diff --git a/CHANGELOG.md b/CHANGELOG.md index 2c1d4b96..7a5c578d 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/README.md b/README.md index e075abc4..4758e25f 100644 --- a/README.md +++ b/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) diff --git a/cmd/main.go b/cmd/main.go index 345e2bd3..58d0961c 100644 --- a/cmd/main.go +++ b/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 { diff --git a/images/diagram.png b/images/diagram.png new file mode 100644 index 00000000..e752bfde Binary files /dev/null and b/images/diagram.png differ diff --git a/pkg/event/sign.go b/pkg/event/sign.go new file mode 100644 index 00000000..cb8d53d8 --- /dev/null +++ b/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"` +} diff --git a/pkg/eventconsumer/consumer.go b/pkg/eventconsumer/consumer.go deleted file mode 100644 index cf17cc82..00000000 --- a/pkg/eventconsumer/consumer.go +++ /dev/null @@ -1,245 +0,0 @@ -package eventconsumer - -import ( - "context" - "encoding/json" - "fmt" - "log" - "math/big" - "sync" - "time" - - "github.com/cryptoniumX/mpcium/pkg/logger" - "github.com/cryptoniumX/mpcium/pkg/messaging" - "github.com/cryptoniumX/mpcium/pkg/mpc" -) - -const ( - MPCGenerateEvent = "mpc:generate" - MPCSignEvent = "mpc:sign" -) - -type EventConsumer interface { - Run() - Close() error -} - -type eventConsumer struct { - node *mpc.Node - pubsub messaging.PubSub - - genKeySucecssQueue messaging.MessageQueue - signingSuccessQueue messaging.MessageQueue - - keyGenerationSub messaging.Subscription - signingSub messaging.Subscription -} - -func NewEventConsumer( - node *mpc.Node, - pubsub messaging.PubSub, - genKeySucecssQueue messaging.MessageQueue, - signingSuccessQueue messaging.MessageQueue, -) EventConsumer { - return &eventConsumer{ - node: node, - pubsub: pubsub, - genKeySucecssQueue: genKeySucecssQueue, - signingSuccessQueue: signingSuccessQueue, - } -} - -func (ec *eventConsumer) Run() { - err := ec.consumeKeyGenerationEvent() - if err != nil { - log.Fatal("Failed to consume key reconstruction event", err) - } - - err = ec.consumeTxSigningEvent() - if err != nil { - log.Fatal("Failed to consume tx signing event", err) - } - - logger.Info("MPC Event consumer started...!") -} - -func (ec *eventConsumer) consumeKeyGenerationEvent() error { - sub, err := ec.pubsub.Subscribe(MPCGenerateEvent, func(msg []byte) { - walletID := string(msg) - // TODO: threshold is configurable - threshold := 1 - session, err := ec.node.CreateKeyGenSession(walletID, threshold, ec.genKeySucecssQueue) - if err != nil { - logger.Error("Failed to create key generation session", err, "walletID", walletID) - return - } - eddsaSession, err := ec.node.CreateEDDSAKeyGenSession(walletID, threshold, ec.genKeySucecssQueue) - if err != nil { - logger.Error("Failed to create key generation session", err, "walletID", walletID) - return - } - - session.Init() - eddsaSession.Init() - - ctx, done := context.WithCancel(context.Background()) - ctxEddsa, doneEddsa := context.WithCancel(context.Background()) - - successEvent := &mpc.KeygenSuccessEvent{ - WalletID: walletID, - } - - var wg sync.WaitGroup - wg.Add(2) - go func() { - for { - select { - case <-ctx.Done(): - successEvent.S256PubKey = session.GetPubKeyResult() - wg.Done() - return - case err := <-session.ErrCh: - logger.Error("Keygen session error", err) - } - } - }() - - go func() { - for { - select { - case <-ctxEddsa.Done(): - successEvent.EDDSAPubKey = eddsaSession.GetPubKeyResult() - wg.Done() - return - case err := <-eddsaSession.ErrCh: - logger.Error("Keygen session error", err) - } - } - }() - - session.ListenToIncomingMessageAsync() - eddsaSession.ListenToIncomingMessageAsync() - // TODO: replace sleep with distributed lock - time.Sleep(1 * time.Second) - - go session.GenerateKey(done) - go eddsaSession.GenerateKey(doneEddsa) - - wg.Wait() - if err != nil { - logger.Error("Errors when closing sessions", err) - } - logger.Info("Closing section successfully!", "event", successEvent) - - successEventBytes, err := json.Marshal(successEvent) - if err != nil { - logger.Error("Failed to marshal keygen success event", err) - return - } - - err = ec.genKeySucecssQueue.Enqueue(fmt.Sprintf(mpc.TypeGenerateWalletSuccess, walletID), successEventBytes, &messaging.EnqueueOptions{ - IdempotententKey: fmt.Sprintf(mpc.TypeGenerateWalletSuccess, walletID), - }) - if err != nil { - logger.Error("Failed to publish key generation success message", err) - return - } - - logger.Info("[COMPLETED KEY GEN] Key generation completed successfully", "walletID", walletID) - if err != nil { - logger.Error("Failed to close session", err) - } - - }) - - ec.keyGenerationSub = sub - if err != nil { - return err - } - return nil -} - -func (ec *eventConsumer) consumeTxSigningEvent() error { - sub, err := ec.pubsub.Subscribe(MPCSignEvent, func(raw []byte) { - var msg SignTxMessage - err := json.Unmarshal(raw, &msg) - if err != nil { - logger.Error("Failed to unmarshal message", err) - return - } - - logger.Info("Received signing event", "waleltID", msg.WalletID, "type", msg.KeyType, "tx", msg.Tx) - threshold := 1 - - var session mpc.ISigningSession - switch msg.KeyType { - case KeyTypeSecp256k1: - session, err = ec.node.CreateSigningSession( - msg.WalletID, - msg.TxID, - msg.NetworkInternalCode, - threshold, - ec.signingSuccessQueue, - ) - case KeyTypeEd25519: - session, err = ec.node.CreateEDDSASigningSession( - msg.WalletID, - msg.TxID, - msg.NetworkInternalCode, - threshold, - ec.signingSuccessQueue, - ) - - } - - if err != nil { - logger.Error("Failed to create signing session", err) - return - } - - txBigInt := new(big.Int).SetBytes(msg.Tx) - err = session.Init(txBigInt) - if err != nil { - logger.Error("Failed to init signing session, terminate session", err, "walletID", msg.WalletID) - return - } - - ctx, done := context.WithCancel(context.Background()) - go func() { - for { - select { - case <-ctx.Done(): - return - case err := <-session.ErrChan(): - logger.Error("Signing session error", err) - } - } - }() - - session.ListenToIncomingMessageAsync() - // TODO: use consul distributed lock here - time.Sleep(1 * time.Second) - go session.Sign(done) // use go routine to not block the event susbscriber - }) - - ec.signingSub = sub - if err != nil { - return err - } - - return nil -} - -// Close and clean up -func (ec *eventConsumer) Close() error { - err := ec.keyGenerationSub.Unsubscribe() - if err != nil { - return err - } - err = ec.signingSub.Unsubscribe() - if err != nil { - return err - } - - return nil -} diff --git a/pkg/eventconsumer/event_consumer.go b/pkg/eventconsumer/event_consumer.go new file mode 100644 index 00000000..0e32494d --- /dev/null +++ b/pkg/eventconsumer/event_consumer.go @@ -0,0 +1,418 @@ +package eventconsumer + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "log" + "math/big" + "sync" + "time" + + "github.com/cryptoniumX/mpcium/pkg/event" + "github.com/cryptoniumX/mpcium/pkg/logger" + "github.com/cryptoniumX/mpcium/pkg/messaging" + "github.com/cryptoniumX/mpcium/pkg/mpc" + "github.com/nats-io/nats.go" +) + +const ( + MPCGenerateEvent = "mpc:generate" + MPCSignEvent = "mpc:sign" +) + +type EventConsumer interface { + Run() + Close() error +} + +type eventConsumer struct { + node *mpc.Node + pubsub messaging.PubSub + + genKeySucecssQueue messaging.MessageQueue + signingResultQueue messaging.MessageQueue + + keyGenerationSub messaging.Subscription + signingSub messaging.Subscription + + // Track active sessions with timestamps for cleanup + activeSessions map[string]time.Time // Maps "walletID-txID" to creation time + sessionsLock sync.RWMutex + cleanupInterval time.Duration // How often to run cleanup + sessionTimeout time.Duration // How long before a session is considered stale + cleanupStopChan chan struct{} // Signal to stop cleanup goroutine +} + +func NewEventConsumer( + node *mpc.Node, + pubsub messaging.PubSub, + genKeySucecssQueue messaging.MessageQueue, + signingResultQueue messaging.MessageQueue, +) EventConsumer { + ec := &eventConsumer{ + node: node, + pubsub: pubsub, + genKeySucecssQueue: genKeySucecssQueue, + signingResultQueue: signingResultQueue, + activeSessions: make(map[string]time.Time), + cleanupInterval: 5 * time.Minute, // Run cleanup every 5 minutes + sessionTimeout: 30 * time.Minute, // Consider sessions older than 30 minutes stale + cleanupStopChan: make(chan struct{}), + } + + // Start background cleanup goroutine + go ec.sessionCleanupRoutine() + + return ec +} + +func (ec *eventConsumer) Run() { + err := ec.consumeKeyGenerationEvent() + if err != nil { + log.Fatal("Failed to consume key reconstruction event", err) + } + + err = ec.consumeTxSigningEvent() + if err != nil { + log.Fatal("Failed to consume tx signing event", err) + } + + logger.Info("MPC Event consumer started...!") +} + +func (ec *eventConsumer) consumeKeyGenerationEvent() error { + sub, err := ec.pubsub.Subscribe(MPCGenerateEvent, func(natMsg *nats.Msg) { + msg := natMsg.Data + walletID := string(msg) + // TODO: threshold is configurable + threshold := 1 + session, err := ec.node.CreateKeyGenSession(walletID, threshold, ec.genKeySucecssQueue) + if err != nil { + logger.Error("Failed to create key generation session", err, "walletID", walletID) + return + } + eddsaSession, err := ec.node.CreateEDDSAKeyGenSession(walletID, threshold, ec.genKeySucecssQueue) + if err != nil { + logger.Error("Failed to create key generation session", err, "walletID", walletID) + return + } + + session.Init() + eddsaSession.Init() + + ctx, done := context.WithCancel(context.Background()) + ctxEddsa, doneEddsa := context.WithCancel(context.Background()) + + successEvent := &mpc.KeygenSuccessEvent{ + WalletID: walletID, + } + + var wg sync.WaitGroup + wg.Add(2) + go func() { + for { + select { + case <-ctx.Done(): + successEvent.S256PubKey = session.GetPubKeyResult() + wg.Done() + return + case err := <-session.ErrCh: + logger.Error("Keygen session error", err) + } + } + }() + + go func() { + for { + select { + case <-ctxEddsa.Done(): + successEvent.EDDSAPubKey = eddsaSession.GetPubKeyResult() + wg.Done() + return + case err := <-eddsaSession.ErrCh: + logger.Error("Keygen session error", err) + } + } + }() + + session.ListenToIncomingMessageAsync() + eddsaSession.ListenToIncomingMessageAsync() + // TODO: replace sleep with distributed lock + time.Sleep(1 * time.Second) + + go session.GenerateKey(done) + go eddsaSession.GenerateKey(doneEddsa) + + wg.Wait() + logger.Info("Closing section successfully!", "event", successEvent) + + successEventBytes, err := json.Marshal(successEvent) + if err != nil { + logger.Error("Failed to marshal keygen success event", err) + return + } + + err = ec.genKeySucecssQueue.Enqueue(fmt.Sprintf(mpc.TypeGenerateWalletSuccess, walletID), successEventBytes, &messaging.EnqueueOptions{ + IdempotententKey: fmt.Sprintf(mpc.TypeGenerateWalletSuccess, walletID), + }) + if err != nil { + logger.Error("Failed to publish key generation success message", err) + return + } + + logger.Info("[COMPLETED KEY GEN] Key generation completed successfully", "walletID", walletID) + + }) + + ec.keyGenerationSub = sub + if err != nil { + return err + } + return nil +} + +func (ec *eventConsumer) consumeTxSigningEvent() error { + sub, err := ec.pubsub.Subscribe(MPCSignEvent, func(natMsg *nats.Msg) { + raw := natMsg.Data + var msg SignTxMessage + err := json.Unmarshal(raw, &msg) + if err != nil { + logger.Error("Failed to unmarshal signing message", err) + return + } + + logger.Info( + "Received signing event", + "waleltID", + msg.WalletID, + "type", + msg.KeyType, + "tx", + msg.TxID, + "Id", + ec.node.ID(), + ) + threshold := 1 + + // Check for duplicate session and track if new + if ec.checkDuplicateSession(msg.WalletID, msg.TxID) { + natMsg.Term() + return + } + + var session mpc.ISigningSession + switch msg.KeyType { + case KeyTypeSecp256k1: + session, err = ec.node.CreateSigningSession( + msg.WalletID, + msg.TxID, + msg.NetworkInternalCode, + threshold, + ec.signingResultQueue, + ) + case KeyTypeEd25519: + session, err = ec.node.CreateEDDSASigningSession( + msg.WalletID, + msg.TxID, + msg.NetworkInternalCode, + threshold, + ec.signingResultQueue, + ) + + } + + if err != nil { + ec.handleSigningSessionError( + msg.WalletID, + msg.TxID, + msg.NetworkInternalCode, + err, + "Failed to create signing session", + natMsg, + ) + return + } + + txBigInt := new(big.Int).SetBytes(msg.Tx) + err = session.Init(txBigInt) + if err != nil { + if errors.Is(err, mpc.ErrNotEnoughParticipants) { + logger.Info("RETRY LATER: Not enough participants to sign") + //Return for retry later + return + } + ec.handleSigningSessionError( + msg.WalletID, + msg.TxID, + msg.NetworkInternalCode, + err, + "Failed to init signing session", + natMsg, + ) + return + } + + // Mark session as already processed + ec.addSession(msg.WalletID, msg.TxID) + + ctx, done := context.WithCancel(context.Background()) + go func() { + for { + select { + case <-ctx.Done(): + return + case err := <-session.ErrChan(): + if err != nil { + ec.handleSigningSessionError( + msg.WalletID, + msg.TxID, + msg.NetworkInternalCode, + err, + "Failed to sign tx", + natMsg, + ) + return + } + } + } + }() + + session.ListenToIncomingMessageAsync() + // TODO: use consul distributed lock here, only sign after all nodes has already completed listing to incoming message async + // The purpose of the sleep is to be ensuring that the node has properly set up its message listeners + // before it starts the signing process. If the signing process starts sending messages before other nodes + // have set up their listeners, those messages might be missed, potentially causing the signing process to fail. + // One solution: + // The messaging includes mechanisms for direct point-to-point communication (in point2point.go). + // The nodes could explicitly coordinate through request-response patterns before starting signing + time.Sleep(1 * time.Second) + + onSuccess := func(data []byte) { + done() + if natMsg.Reply != "" { + err = ec.pubsub.Publish(natMsg.Reply, data) + if err != nil { + logger.Error("Failed to publish reply", err) + } else { + logger.Info("Reply to the original message", "reply", natMsg.Reply) + } + } + } + go session.Sign(onSuccess) + }) + + ec.signingSub = sub + if err != nil { + return err + } + + return nil +} + +func (ec *eventConsumer) handleSigningSessionError(walletID, txID, NetworkInternalCode string, err error, errMsg string, natMsg *nats.Msg) { + logger.Error("Signing session error", err, "walletID", walletID, "txID", txID, "error", errMsg) + signingResult := event.SigningResultEvent{ + ResultType: event.SigningResultTypeError, + NetworkInternalCode: NetworkInternalCode, + WalletID: walletID, + TxID: txID, + ErrorReason: errMsg, + } + + signingResultBytes, err := json.Marshal(signingResult) + if err != nil { + logger.Error("Failed to marshal signing result event", err) + return + } + + natMsg.Ack() + err = ec.signingResultQueue.Enqueue(event.SigningResultCompleteTopic, signingResultBytes, &messaging.EnqueueOptions{ + IdempotententKey: txID, + }) + if err != nil { + logger.Error("Failed to publish signing result event", err) + return + } +} + +// Add a cleanup routine that runs periodically +func (ec *eventConsumer) sessionCleanupRoutine() { + ticker := time.NewTicker(ec.cleanupInterval) + defer ticker.Stop() + + for { + select { + case <-ticker.C: + ec.cleanupStaleSessions() + case <-ec.cleanupStopChan: + return + } + } +} + +// Cleanup stale sessions +func (ec *eventConsumer) cleanupStaleSessions() { + now := time.Now() + ec.sessionsLock.Lock() + defer ec.sessionsLock.Unlock() + + for sessionID, creationTime := range ec.activeSessions { + if now.Sub(creationTime) > ec.sessionTimeout { + logger.Info("Cleaning up stale session", "sessionID", sessionID, "age", now.Sub(creationTime)) + delete(ec.activeSessions, sessionID) + } + } +} + +// markSessionAsActive marks a session as active with the current timestamp +func (ec *eventConsumer) addSession(walletID, txID string) { + sessionID := fmt.Sprintf("%s-%s", walletID, txID) + ec.sessionsLock.Lock() + ec.activeSessions[sessionID] = time.Now() + ec.sessionsLock.Unlock() +} + +// Remove a session from tracking +func (ec *eventConsumer) removeSession(walletID, txID string) { + sessionID := fmt.Sprintf("%s-%s", walletID, txID) + ec.sessionsLock.Lock() + delete(ec.activeSessions, sessionID) + ec.sessionsLock.Unlock() +} + +// checkAndTrackSession checks if a session already exists and tracks it if new. +// Returns true if the session is a duplicate. +func (ec *eventConsumer) checkDuplicateSession(walletID, txID string) bool { + sessionID := fmt.Sprintf("%s-%s", walletID, txID) + + // Check for duplicate + ec.sessionsLock.RLock() + _, isDuplicate := ec.activeSessions[sessionID] + ec.sessionsLock.RUnlock() + + if isDuplicate { + logger.Info("Duplicate signing request detected", "walletID", walletID, "txID", txID) + return true + } + + return false +} + +// Close and clean up +func (ec *eventConsumer) Close() error { + // Signal cleanup routine to stop + close(ec.cleanupStopChan) + + err := ec.keyGenerationSub.Unsubscribe() + if err != nil { + return err + } + err = ec.signingSub.Unsubscribe() + if err != nil { + return err + } + + return nil +} diff --git a/pkg/eventconsumer/sign_consumer.go b/pkg/eventconsumer/sign_consumer.go new file mode 100644 index 00000000..fad3e23b --- /dev/null +++ b/pkg/eventconsumer/sign_consumer.go @@ -0,0 +1,144 @@ +package eventconsumer + +import ( + "context" + "fmt" + "time" + + "github.com/cryptoniumX/mpcium/pkg/event" + "github.com/cryptoniumX/mpcium/pkg/logger" + "github.com/cryptoniumX/mpcium/pkg/messaging" + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" +) + +const ( + // Maximum time to wait for a signing response. + signingResponseTimeout = 30 * time.Second + // How often to poll for the reply message. + signingPollingInterval = 500 * time.Millisecond +) + +// SigningConsumer represents a consumer that processes signing events. +type SigningConsumer interface { + // Run starts the consumer and blocks until the provided context is canceled. + Run(ctx context.Context) error + // Close performs a graceful shutdown of the consumer. + Close() error +} + +// signingConsumer implements SigningConsumer. +type signingConsumer struct { + natsConn *nats.Conn + pubsub messaging.PubSub + jsPubsub messaging.StreamPubsub + + // 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, jsPubsub messaging.StreamPubsub, pubsub messaging.PubSub) SigningConsumer { + return &signingConsumer{ + natsConn: natsConn, + pubsub: pubsub, + jsPubsub: jsPubsub, + } +} + +// Run subscribes to signing events and processes them until the context is canceled. +func (sc *signingConsumer) Run(ctx context.Context) error { + sub, err := sc.jsPubsub.Subscribe( + event.SigningConsumerStream, + event.SigningRequestEventTopic, + sc.handleSigningEvent, + ) + if err != nil { + return fmt.Errorf("failed to subscribe to signing events: %w", err) + } + sc.jsSub = sub + logger.Info("SigningConsumer: Subscribed to signing events") + + // Block until context cancellation. + <-ctx.Done() + + // When context is canceled, close subscription. + return sc.Close() +} + +// The handleSigningEvent function in sign_consumer.go acts as a bridge between the JetStream-based event queue and the MPC (Multi-Party Computation) signing system +// Creates a reply channel: It generates a unique inbox address using nats.NewInbox() to receive the signing response. +// Sets up response handling: It creates a synchronous subscription to listen for replies on this inbox. +// Forwards the signing request: It publishes the original signing event data to the MPCSigningEventTopic with the reply inbox attached, which triggers the MPC signing process. +// Polls for completion: It enters a polling loop that checks for a reply message, continuing until either: +// A reply is received (successful signing) +// An error occurs (failed signing) +// The timeout is reached (30 seconds) +// Completes the transaction: It either acknowledges (Ack) the message if signing was successful or negatively acknowledges (Nak) it if there was a timeout or error. +// MPC Session Interaction +// The signing consumer doesn't directly interact with MPC sessions. Instead: +// It publishes the signing request to the MPCSigningEventTopic, which is consumed by the eventconsumer.consumeTxSigningEvent handler. +// This handler creates the appropriate signing session (SigningSession for ECDSA or EDDSASigningSession for EdDSA) via the MPC node's creation methods. +// The MPC signing sessions manage the distributed cryptographic operations across multiple nodes, handling message routing, party updates, and signature verification. +// 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) { + // Create a reply inbox to receive the signing event response. + replyInbox := nats.NewInbox() + + // Use a synchronous subscription for the reply inbox. + replySub, err := sc.natsConn.SubscribeSync(replyInbox) + if err != nil { + logger.Error("SigningConsumer: Failed to subscribe to reply inbox", err) + _ = msg.Nak() + return + } + defer func() { + if err := replySub.Unsubscribe(); err != nil { + logger.Warn("SigningConsumer: Failed to unsubscribe from reply inbox", err) + } + }() + + // Publish the signing event with the reply inbox. + if err := sc.pubsub.PublishWithReply(event.MPCSigningEventTopic, replyInbox, msg.Data()); err != nil { + logger.Error("SigningConsumer: Failed to publish signing event with reply", err) + _ = msg.Nak() + return + } + + // Poll for the reply message until timeout. + deadline := time.Now().Add(signingResponseTimeout) + for time.Now().Before(deadline) { + replyMsg, err := replySub.NextMsg(signingPollingInterval) + if err != nil { + // If timeout occurs, continue trying. + if err == nats.ErrTimeout { + continue + } + logger.Error("SigningConsumer: Error receiving reply message", err) + break + } + if replyMsg != nil { + logger.Info("SigningConsumer: Completed signing event; reply received") + if ackErr := msg.Ack(); ackErr != nil { + logger.Error("SigningConsumer: ACK failed", ackErr) + } + return + } + } + + logger.Warn("SigningConsumer: Timeout waiting for signing event response") + _ = msg.Nak() +} + +// Close unsubscribes from the JetStream subject and cleans up resources. +func (sc *signingConsumer) Close() error { + if sc.jsSub != nil { + if err := sc.jsSub.Unsubscribe(); err != nil { + logger.Error("SigningConsumer: Failed to unsubscribe from JetStream", err) + return err + } + logger.Info("SigningConsumer: Unsubscribed from JetStream") + } + return nil +} diff --git a/pkg/eventconsumer/timeout_consumer.go b/pkg/eventconsumer/timeout_consumer.go new file mode 100644 index 00000000..6c4a1802 --- /dev/null +++ b/pkg/eventconsumer/timeout_consumer.go @@ -0,0 +1,99 @@ +package eventconsumer + +import ( + "encoding/json" + "fmt" + + "github.com/cryptoniumX/mpcium/pkg/event" + "github.com/cryptoniumX/mpcium/pkg/logger" + "github.com/cryptoniumX/mpcium/pkg/messaging" + "github.com/nats-io/nats.go" +) + +// Other service not listen to this subject that make loss of message +const maxDeliveriesExceededSubject = "$JS.EVENT.ADVISORY.CONSUMER.MAX_DELIVERIES.>" + +type timeOutConsumer struct { + natsConn *nats.Conn + resultQueue messaging.MessageQueue + advisorySub messaging.Subscription +} + +func NewTimeOutConsumer(natsConn *nats.Conn, resultQueue messaging.MessageQueue) *timeOutConsumer { + return &timeOutConsumer{ + natsConn: natsConn, + resultQueue: resultQueue, + } +} + +func (tc *timeOutConsumer) Run() { + logger.Info("Starting advisory consumer for max deliveries exceeded") + sub, err := tc.natsConn.Subscribe(maxDeliveriesExceededSubject, func(msg *nats.Msg) { + data := msg.Data + var advisory struct { + Stream string `json:"stream"` + StreamSeq uint64 `json:"stream_seq"` + } + + err := json.Unmarshal(data, &advisory) + if err != nil { + logger.Error("Failed to unmarshal advisory message", err) + return + } + logger.Info("Received advisory message", "stream", advisory.Stream, "stream_seq", advisory.StreamSeq) + + if advisory.Stream == event.SigningPublisherStream { + logger.Info("Received max deliveries exceeded advisory", "stream", advisory.Stream, "stream_seq", advisory.StreamSeq) + js, _ := tc.natsConn.JetStream() + failedMsg, err := js.GetMsg(advisory.Stream, advisory.StreamSeq) + + if err != nil { + logger.Error("Failed to retrieve message", err) + return + } + + data := failedMsg.Data + var signErrorResult event.SigningResultEvent + err = json.Unmarshal(data, &signErrorResult) + + if err != nil { + logger.Error("Failed to unmarshal signing result event", err) + return + } + + signErrorResult.ResultType = event.SigningResultTypeError + signErrorResult.IsTimeout = true + signErrorResult.ErrorReason = fmt.Sprintf("Message delivery exceeded for stream %s", advisory.Stream) + + signErrorResultBytes, err := json.Marshal(signErrorResult) + if err != nil { + logger.Error("Failed to marshal signing result event", err) + return + } + + err = tc.resultQueue.Enqueue(event.SigningResultTopic, signErrorResultBytes, &messaging.EnqueueOptions{ + IdempotententKey: signErrorResult.TxID, + }) + if err != nil { + logger.Error("Failed to publish signing result event", err) + return + } + logger.Info("Published signing result event for timeout", "txID", signErrorResult.TxID) + return + } + }) + if err != nil { + logger.Error("Failed to subscribe to max deliveries exceeded subject", err) + return + } + + tc.advisorySub = sub +} + +func (tc *timeOutConsumer) Close() error { + err := tc.advisorySub.Unsubscribe() + if err != nil { + return err + } + return nil +} diff --git a/pkg/logger/logger.go b/pkg/logger/logger.go index dd42bb6c..543c314a 100644 --- a/pkg/logger/logger.go +++ b/pkg/logger/logger.go @@ -12,7 +12,7 @@ var Log zerolog.Logger func Init(env string) { zerolog.ErrorStackMarshaler = pkgerrors.MarshalStack if env != "production" { - Log = zerolog.New(zerolog.ConsoleWriter{Out: os.Stderr, NoColor: false}).With().Timestamp().Logger() + Log = zerolog.New(zerolog.ConsoleWriter{Out: os.Stderr, NoColor: false, TimeFormat: "2006-01-02 15:04:05.000"}).With().Timestamp().Logger() } else { Log = zerolog.New(os.Stdout).With().Timestamp().Logger() } diff --git a/pkg/messaging/pubsub.go b/pkg/messaging/pubsub.go index a956ac6b..ab4f4d0f 100644 --- a/pkg/messaging/pubsub.go +++ b/pkg/messaging/pubsub.go @@ -1,7 +1,15 @@ package messaging import ( + "context" + "fmt" + "strings" + "time" + "unicode" + + "github.com/cryptoniumX/mpcium/pkg/logger" "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" ) type Subscription interface { @@ -10,7 +18,8 @@ type Subscription interface { type PubSub interface { Publish(topic string, message []byte) error - Subscribe(topic string, handler func(message []byte)) (Subscription, error) + PublishWithReply(ttopic, reply string, data []byte) error + Subscribe(topic string, handler func(msg *nats.Msg)) (Subscription, error) } type natsPubSub struct { @@ -21,23 +30,40 @@ type natsSubscription struct { subscription *nats.Subscription } +type jetstreamSubscription struct { + consumer jetstream.Consumer +} + func (ns *natsSubscription) Unsubscribe() error { return ns.subscription.Unsubscribe() } +func (js *jetstreamSubscription) Unsubscribe() error { + return nil +} + func NewNATSPubSub(natsConn *nats.Conn) PubSub { return &natsPubSub{natsConn} } func (n *natsPubSub) Publish(topic string, message []byte) error { + fmt.Println("Publishing message to topic:", topic) return n.natsConn.Publish(topic, message) } -func (n *natsPubSub) Subscribe(topic string, handler func(message []byte)) (Subscription, error) { +func (n *natsPubSub) PublishWithReply(topic, reply string, data []byte) error { + return n.natsConn.PublishMsg(&nats.Msg{ + Subject: topic, + Reply: reply, + Data: data, + }) +} + +func (n *natsPubSub) Subscribe(topic string, handler func(msg *nats.Msg)) (Subscription, error) { // TODO: Handle subscription // handle more fields in msg sub, err := n.natsConn.Subscribe(topic, func(msg *nats.Msg) { - handler(msg.Data) + handler(msg) }) if err != nil { return nil, err @@ -45,3 +71,186 @@ func (n *natsPubSub) Subscribe(topic string, handler func(message []byte)) (Subs return &natsSubscription{subscription: sub}, nil } + +type StreamPubsub interface { + Publish(topic string, message []byte) error + Subscribe(name string, topic string, handler func(msg jetstream.Msg)) (Subscription, error) +} + +type StreamPubsubOption func(*streamPubSubConfig) + +type streamPubSubConfig struct { + streamName string + subjects []string + description string + retention nats.RetentionPolicy + storage nats.StorageType + maxAge time.Duration + discard nats.DiscardPolicy + ackWait time.Duration + maxDeliveryAttempts int + consumerNamePrefix string +} + +func WithDescription(description string) StreamPubsubOption { + return func(cfg *streamPubSubConfig) { + cfg.description = description + } +} + +func WithRetention(policy nats.RetentionPolicy) StreamPubsubOption { + return func(cfg *streamPubSubConfig) { + cfg.retention = policy + } +} + +func WithStorage(storage nats.StorageType) StreamPubsubOption { + return func(cfg *streamPubSubConfig) { + cfg.storage = storage + } +} + +func WithMaxAge(maxAge time.Duration) StreamPubsubOption { + return func(cfg *streamPubSubConfig) { + cfg.maxAge = maxAge + } +} + +func WithDiscardPolicy(policy nats.DiscardPolicy) StreamPubsubOption { + return func(cfg *streamPubSubConfig) { + cfg.discard = policy + } +} + +func WithAckWait(ackWait time.Duration) StreamPubsubOption { + return func(cfg *streamPubSubConfig) { + cfg.ackWait = ackWait + } +} + +func WithMaxDeliveryAttempts(maxAttempts int) StreamPubsubOption { + return func(cfg *streamPubSubConfig) { + cfg.maxDeliveryAttempts = maxAttempts + } +} + +func WithConsumerNamePrefix(prefix string) StreamPubsubOption { + return func(cfg *streamPubSubConfig) { + cfg.consumerNamePrefix = prefix + } +} + +type jetStreamPubSub struct { + name string + config streamPubSubConfig + js jetstream.JetStream +} + +func NewJetStreamPubSub(natsConn *nats.Conn, streamName string, subjects []string, opts ...StreamPubsubOption) (StreamPubsub, error) { + config := streamPubSubConfig{ + streamName: streamName, + subjects: subjects, + retention: nats.InterestPolicy, + storage: nats.MemoryStorage, + discard: nats.DiscardOld, + ackWait: 60 * time.Second, + maxDeliveryAttempts: 3, + consumerNamePrefix: "consumer", + } + for _, opt := range opts { + opt(&config) + } + + js, err := jetstream.New(natsConn) + if err != nil { + logger.Fatal("Error creating JetStream context: ", err) + } + + ctx := context.Background() + stream, err := js.Stream(ctx, streamName) + if err != nil { + logger.Warn("Stream not found, creating new stream", "stream", streamName) + } + if stream != nil { + info, _ := stream.Info(ctx) + logger.Info("Stream found", "info", info) + + } + + _, err = js.CreateOrUpdateStream(context.Background(), jetstream.StreamConfig{ + Name: streamName, + Description: "Stream for " + streamName, + Subjects: subjects, + }) + + if err != nil { + logger.Fatal("Error creating JetStream stream: ", err) + } + + logger.Info("Creating apex NATs Jetstream context successfully!") + + return &jetStreamPubSub{ + name: streamName, + config: config, + js: js, + }, nil +} + +func (j *jetStreamPubSub) Publish(topic string, message []byte) error { + _, err := j.js.Publish(context.Background(), topic, message) + return err +} + +func sanitizeConsumerName(name string) string { + // Replace invalid characters + name = strings.ReplaceAll(name, ".", "_") + name = strings.ReplaceAll(name, ":", "_") + name = strings.ReplaceAll(name, " ", "_") + name = strings.ReplaceAll(name, "-", "_") + name = strings.ReplaceAll(name, ">", "all") + name = strings.ReplaceAll(name, "*", "any") + + // Ensure it starts with a letter or underscore + if len(name) > 0 && !unicode.IsLetter(rune(name[0])) && name[0] != '_' { + name = "_" + name + } + + return name +} + +func (j *jetStreamPubSub) Subscribe(name string, topic string, handler func(msg jetstream.Msg)) (Subscription, error) { + + logger.Info("Subscribing to topic", sanitizeConsumerName(name), topic) + consumerConfig := jetstream.ConsumerConfig{ + Name: sanitizeConsumerName(name), + Durable: sanitizeConsumerName(name), + AckPolicy: jetstream.AckExplicitPolicy, + MaxDeliver: 3, + BackOff: []time.Duration{30 * time.Second, 30 * time.Second, 30 * time.Second}, + DeliverPolicy: jetstream.DeliverAllPolicy, // Deliver all messages + FilterSubject: topic, + AckWait: 30 * time.Second, // explicitly set ack wait here + } + + logger.Info("Creating consumer", "config", consumerConfig, "stream", j.config.streamName) + consumer, err := j.js.CreateOrUpdateConsumer(context.Background(), j.config.streamName, consumerConfig) + + if err != nil { + logger.Error("❌ Failed to create or update consumer:", err) + } + + if consumer != nil { + logger.Info("✅ Successfully created or updated consumer", "consumer", consumer) + } + + _, err = consumer.Consume(func(msg jetstream.Msg) { + logger.Info("Received jetStreamPubSub message", "subject", msg.Data()) + handler(msg) + }) + + if err != nil { + logger.Error("❌ Failed to consume message:", err) + } + + return &jetstreamSubscription{consumer: consumer}, nil +} diff --git a/pkg/mpc/ecdsa_keygen_session.go b/pkg/mpc/ecdsa_keygen_session.go index 15b122f8..ad1dff9e 100644 --- a/pkg/mpc/ecdsa_keygen_session.go +++ b/pkg/mpc/ecdsa_keygen_session.go @@ -40,7 +40,7 @@ func NewKeygenSession( preParams *keygen.LocalPreParams, kvstore kvstore.KVStore, keyinfoStore keyinfo.Store, - successQueue messaging.MessageQueue, + resultQueue messaging.MessageQueue, ) *KeygenSession { return &KeygenSession{ Session: Session{ @@ -68,7 +68,7 @@ func NewKeygenSession( return fmt.Sprintf("ecdsa:%s", walletID) }, getRoundFunc: GetEcdsaMsgRound, - successQueue: successQueue, + resultQueue: resultQueue, sessionType: SessionTypeEcdsa, }, endCh: make(chan *keygen.LocalPartySaveData), diff --git a/pkg/mpc/ecdsa_signing_session.go b/pkg/mpc/ecdsa_signing_session.go index 402d0b31..a44f14d6 100644 --- a/pkg/mpc/ecdsa_signing_session.go +++ b/pkg/mpc/ecdsa_signing_session.go @@ -11,6 +11,7 @@ import ( "github.com/bnb-chain/tss-lib/v2/ecdsa/signing" "github.com/bnb-chain/tss-lib/v2/tss" "github.com/cryptoniumX/mpcium/pkg/common/errors" + "github.com/cryptoniumX/mpcium/pkg/event" "github.com/cryptoniumX/mpcium/pkg/keyinfo" "github.com/cryptoniumX/mpcium/pkg/kvstore" "github.com/cryptoniumX/mpcium/pkg/logger" @@ -18,10 +19,6 @@ import ( "github.com/samber/lo" ) -const ( - SignSuccessTopic = "mpc.mpc_sign_success.completed" -) - // Ecdsa signing session type SigningSession struct { Session @@ -41,19 +38,7 @@ type ISigningSession interface { ISession Init(tx *big.Int) error - Sign(done func()) -} - -type SigningSuccessEvent 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"` + Sign(onSuccess func(data []byte)) } func NewSigningSession( @@ -69,7 +54,7 @@ func NewSigningSession( preParams *keygen.LocalPreParams, kvstore kvstore.KVStore, keyinfoStore keyinfo.Store, - succesQueue messaging.MessageQueue, + resultQueue messaging.MessageQueue, ) *SigningSession { return &SigningSession{ Session: Session{ @@ -97,7 +82,7 @@ func NewSigningSession( return fmt.Sprintf("ecdsa:%s", waleltID) }, getRoundFunc: GetEcdsaMsgRound, - successQueue: succesQueue, + resultQueue: resultQueue, }, endCh: make(chan *common.SignatureData), txID: txID, @@ -121,7 +106,8 @@ func (s *SigningSession) Init(tx *big.Int) error { } if len(s.participantPeerIDs) < keyInfo.Threshold+1 { - return fmt.Errorf("Not enough participants to sign, expected %d, got %d", keyInfo.Threshold+1, len(s.participantPeerIDs)) + logger.Warn("Not enough participants to sign", "participants", s.participantPeerIDs, "expected", keyInfo.Threshold+1) + return ErrNotEnoughParticipants } // check if t+1 participants are present @@ -149,7 +135,7 @@ func (s *SigningSession) Init(tx *big.Int) error { return nil } -func (s *SigningSession) Sign(done func()) { +func (s *SigningSession) Sign(onSuccess func(data []byte)) { logger.Info("Starting signing", "walletID", s.walletID) go func() { if err := s.party.Start(); err != nil { @@ -176,7 +162,8 @@ func (s *SigningSession) Sign(done func()) { return } - r := SigningSuccessEvent{ + r := event.SigningResultEvent{ + ResultType: event.SigningResultTypeSuccess, NetworkInternalCode: s.networkInternalCode, WalletID: s.walletID, TxID: s.txID, @@ -191,7 +178,7 @@ func (s *SigningSession) Sign(done func()) { return } - err = s.successQueue.Enqueue(SignSuccessTopic, bytes, &messaging.EnqueueOptions{ + err = s.resultQueue.Enqueue(event.SigningResultCompleteTopic, bytes, &messaging.EnqueueOptions{ IdempotententKey: s.txID, }) if err != nil { @@ -206,7 +193,7 @@ func (s *SigningSession) Sign(done func()) { logger.Error("Failed to close session", err) } - done() + onSuccess(bytes) return } diff --git a/pkg/mpc/eddsa_keygen_session.go b/pkg/mpc/eddsa_keygen_session.go index 0bccff46..2a6152df 100644 --- a/pkg/mpc/eddsa_keygen_session.go +++ b/pkg/mpc/eddsa_keygen_session.go @@ -33,7 +33,7 @@ func NewEDDSAKeygenSession( threshold int, kvstore kvstore.KVStore, keyinfoStore keyinfo.Store, - successQueue messaging.MessageQueue, + resultQueue messaging.MessageQueue, ) *EDDSAKeygenSession { return &EDDSAKeygenSession{Session: Session{ walletID: walletID, @@ -59,7 +59,7 @@ func NewEDDSAKeygenSession( return fmt.Sprintf("eddsa:%s", waleltID) }, getRoundFunc: GetEddsaMsgRound, - successQueue: successQueue, + resultQueue: resultQueue, sessionType: SessionTypeEddsa, }, endCh: make(chan *keygen.LocalPartySaveData), diff --git a/pkg/mpc/eddsa_signing_session.go b/pkg/mpc/eddsa_signing_session.go index 95e26e48..07495c24 100644 --- a/pkg/mpc/eddsa_signing_session.go +++ b/pkg/mpc/eddsa_signing_session.go @@ -10,6 +10,7 @@ import ( "github.com/bnb-chain/tss-lib/v2/eddsa/signing" "github.com/bnb-chain/tss-lib/v2/tss" "github.com/cryptoniumX/mpcium/pkg/common/errors" + "github.com/cryptoniumX/mpcium/pkg/event" "github.com/cryptoniumX/mpcium/pkg/keyinfo" "github.com/cryptoniumX/mpcium/pkg/kvstore" "github.com/cryptoniumX/mpcium/pkg/logger" @@ -39,7 +40,7 @@ func NewEDDSASigningSession( threshold int, kvstore kvstore.KVStore, keyinfoStore keyinfo.Store, - succesQueue messaging.MessageQueue, + resultQueue messaging.MessageQueue, ) *EDDSASigningSession { return &EDDSASigningSession{ Session: Session{ @@ -67,7 +68,7 @@ func NewEDDSASigningSession( return fmt.Sprintf("eddsa:%s", waleltID) }, getRoundFunc: GetEddsaMsgRound, - successQueue: succesQueue, + resultQueue: resultQueue, }, endCh: make(chan *common.SignatureData), txID: txID, @@ -91,7 +92,8 @@ func (s *EDDSASigningSession) Init(tx *big.Int) error { } if len(s.participantPeerIDs) < keyInfo.Threshold+1 { - return fmt.Errorf("Not enough participants to sign, expected %d, got %d", keyInfo.Threshold+1, len(s.participantPeerIDs)) + logger.Warn("Not enough participants to sign, expected %d, got %d", keyInfo.Threshold+1, len(s.participantPeerIDs)) + return ErrNotEnoughParticipants } // check if t+1 participants are present @@ -119,7 +121,7 @@ func (s *EDDSASigningSession) Init(tx *big.Int) error { return nil } -func (s *EDDSASigningSession) Sign(done func()) { +func (s *EDDSASigningSession) Sign(onSuccess func(data []byte)) { logger.Info("Starting signing", "walletID", s.walletID) go func() { if err := s.party.Start(); err != nil { @@ -146,7 +148,8 @@ func (s *EDDSASigningSession) Sign(done func()) { return } - r := SigningSuccessEvent{ + r := event.SigningResultEvent{ + ResultType: event.SigningResultTypeSuccess, NetworkInternalCode: s.networkInternalCode, WalletID: s.walletID, TxID: s.txID, @@ -159,7 +162,7 @@ func (s *EDDSASigningSession) Sign(done func()) { return } - err = s.successQueue.Enqueue(SignSuccessTopic, bytes, &messaging.EnqueueOptions{ + err = s.resultQueue.Enqueue(event.SigningResultCompleteTopic, bytes, &messaging.EnqueueOptions{ IdempotententKey: s.txID, }) if err != nil { @@ -168,12 +171,13 @@ func (s *EDDSASigningSession) Sign(done func()) { } logger.Info("[SIGN] Sign successfully", "walletID", s.walletID) + err = s.Close() if err != nil { logger.Error("Failed to close session", err) } - done() + onSuccess(bytes) return } diff --git a/pkg/mpc/node.go b/pkg/mpc/node.go index 468c0a2b..e9041854 100644 --- a/pkg/mpc/node.go +++ b/pkg/mpc/node.go @@ -68,7 +68,6 @@ func NewNode( } logger.Info("Starting new node, preparams is generated successfully!") - peerRegistry.Ready() go peerRegistry.WatchPeersReady() return &Node{ @@ -141,7 +140,7 @@ func (p *Node) CreateSigningSession( txID string, networkInternalCode string, threshold int, - successQueue messaging.MessageQueue, + resultQueue messaging.MessageQueue, ) (*SigningSession, error) { readyPeerIDs := p.peerRegistry.GetReadyPeersIncludeSelf() selfPartyID, allPartyIDs := p.generatePartyIDs(PurposeKeygen, readyPeerIDs) @@ -158,7 +157,7 @@ func (p *Node) CreateSigningSession( p.ecdsaPreParams, p.kvstore, p.keyinfoStore, - successQueue, + resultQueue, ) return session, nil } @@ -168,7 +167,7 @@ func (p *Node) CreateEDDSASigningSession( txID string, networkInternalCode string, threshold int, - successQueue messaging.MessageQueue, + resultQueue messaging.MessageQueue, ) (*EDDSASigningSession, error) { readyPeerIDs := p.peerRegistry.GetReadyPeersIncludeSelf() selfPartyID, allPartyIDs := p.generatePartyIDs(PurposeKeygen, readyPeerIDs) @@ -184,7 +183,7 @@ func (p *Node) CreateEDDSASigningSession( threshold, p.kvstore, p.keyinfoStore, - successQueue, + resultQueue, ) return session, nil } diff --git a/pkg/mpc/session.go b/pkg/mpc/session.go index 6e24914e..7a62e342 100644 --- a/pkg/mpc/session.go +++ b/pkg/mpc/session.go @@ -12,6 +12,11 @@ import ( "github.com/cryptoniumX/mpcium/pkg/kvstore" "github.com/cryptoniumX/mpcium/pkg/logger" "github.com/cryptoniumX/mpcium/pkg/messaging" + "github.com/nats-io/nats.go" +) + +var ( + ErrNotEnoughParticipants = errors.New("Not enough participants to sign") ) type TopicComposer struct { @@ -47,7 +52,7 @@ type Session struct { keyinfoStore keyinfo.Store broadcastSub messaging.Subscription directSub messaging.Subscription - successQueue messaging.MessageQueue + resultQueue messaging.MessageQueue topicComposer *TopicComposer composeKey KeyComposerFn @@ -136,9 +141,23 @@ func (s *Session) receiveTssMessage(rawMsg []byte) { } } +func (s *Session) SendReplySignSuccess(natMsg *nats.Msg) { + msg := natMsg.Data + s.mu.Lock() + defer s.mu.Unlock() + + err := s.pubSub.Publish(natMsg.Reply, msg) + if err != nil { + s.ErrCh <- fmt.Errorf("Failed to reply sign sucess message: %w", err) + return + } + logger.Info("Sent reply sign sucess message", "reply", natMsg.Reply) +} + func (s *Session) ListenToIncomingMessageAsync() { go func() { - sub, err := s.pubSub.Subscribe(s.topicComposer.ComposeBroadcastTopic(), func(msg []byte) { + sub, err := s.pubSub.Subscribe(s.topicComposer.ComposeBroadcastTopic(), func(natMsg *nats.Msg) { + msg := natMsg.Data s.receiveTssMessage(msg) })