Ack message in consumers after handling error

Author: anhthii

pkg/eventconsumer/keygen_consumer.go

@@ -129,13 +129,14 @@ func (sc *keygenConsumer) handleKeygenEvent(msg jetstream.Msg) {
 	if err != nil {
 		logger.Error("SigningConsumer: Failed to unmarshal keygen message", err)
 		sc.handleKeygenError(keygenMsg, event.ErrorCodeUnmarshalFailure, err, sessionID)
-		_ = msg.Nak()
+		_ = msg.Ack()
 		return
 	}
 
 	if !sc.peerRegistry.ArePeersReady() {
 		logger.Warn("KeygenConsumer: Not all peers are ready to gen key, skipping message processing")
 		sc.handleKeygenError(keygenMsg, event.ErrorCodeClusterNotReady, errors.New("not all peers are ready"), sessionID)
+		_ = msg.Ack()
 		return
 	}
 

pkg/eventconsumer/sign_consumer.go

@@ -149,14 +149,15 @@ func (sc *signingConsumer) handleSigningEvent(msg jetstream.Msg) {
 	if err != nil {
 		logger.Error("SigningConsumer: Failed to unmarshal signing message", err)
 		sc.handleSigningError(signingMsg, event.ErrorCodeUnmarshalFailure, err, sessionID)
-		_ = msg.Nak()
+		_ = msg.Ack()
 		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)
+		_ = msg.Ack()
 		return
 	}
 	// Create a reply inbox to receive the signing event response.