Add additional idempotence check to cover Kafka server restart, while…

… EthConnect stays running (#227) * Idempotency check on the way into the inflight pool Signed-off-by: Peter Broadhurst <[email protected]> * Update initialization for idempotency check receipt store, and txprocessor impl Signed-off-by: Peter Broadhurst <[email protected]> * Add unit tests for idempotence Signed-off-by: Peter Broadhurst <[email protected]> * Check on TX Hash, in case redelivery happens after full receipt has been persisted Signed-off-by: Peter Broadhurst <[email protected]> * Provide more detailed error when idempotency check fails on receipt store Signed-off-by: Peter Broadhurst <[email protected]> * Better logging and handle redelivery with extra receipt Signed-off-by: Peter Broadhurst <[email protected]> * Do not store receipts when we get a redelivery notification Signed-off-by: Peter Broadhurst <[email protected]> * Store special record if we lose the reply Signed-off-by: Peter Broadhurst <[email protected]> Signed-off-by: Peter Broadhurst <[email protected]>
hyperledger · Aug 31, 2022 · 0e6f5b0 · 0e6f5b0
1 parent 5ee45fd
commit 0e6f5b0
Show file tree

Hide file tree

Showing 26 changed files with 750 additions and 229 deletions.
diff --git a/.vscode/settings.json b/.vscode/settings.json
@@ -9,8 +9,10 @@
     "ffcmocks",
     "fftypes",
     "hashicorp",
+    "idempotency",
     "Infof",
     "kvstore",
+    "receiptsmocks",
     "smconf",
     "stretchr",
     "Tracef",

diff --git a/Makefile b/Makefile
@@ -61,11 +61,12 @@ mocks-$(strip $(1))-$(strip $(2)): ${MOCKERY} sarama
 	${MOCKERY} --case underscore --dir $(1) --name $(2) --outpkg $(3) --output mocks/$(strip $(3))
 endef
 
-$(eval $(call makemock, internal/contractregistry, ContractStore,        contractregistrymocks))
-$(eval $(call makemock, internal/contractregistry, RemoteRegistry,       contractregistrymocks))
-$(eval $(call makemock, internal/eth,              RPCClient,            ethmocks))
-$(eval $(call makemock, internal/ffcapiconnector,  FFCServer,            ffcapiconnectormocks))
-$(eval $(call makemock, $$(SARAMA_PATH),           Client,               saramamocks))
-$(eval $(call makemock, $$(SARAMA_PATH),           ConsumerGroup,        saramamocks))
-$(eval $(call makemock, $$(SARAMA_PATH),           ConsumerGroupSession, saramamocks))
-$(eval $(call makemock, $$(SARAMA_PATH),           ConsumerGroupClaim,   saramamocks))
+$(eval $(call makemock, internal/contractregistry, ContractStore,           contractregistrymocks))
+$(eval $(call makemock, internal/contractregistry, RemoteRegistry,          contractregistrymocks))
+$(eval $(call makemock, internal/eth,              RPCClient,               ethmocks))
+$(eval $(call makemock, internal/ffcapiconnector,  FFCServer,               ffcapiconnectormocks))
+$(eval $(call makemock, internal/receipts,         ReceiptStorePersistence, receiptsmocks))
+$(eval $(call makemock, $$(SARAMA_PATH),           Client,                  saramamocks))
+$(eval $(call makemock, $$(SARAMA_PATH),           ConsumerGroup,           saramamocks))
+$(eval $(call makemock, $$(SARAMA_PATH),           ConsumerGroupSession,    saramamocks))
+$(eval $(call makemock, $$(SARAMA_PATH),           ConsumerGroupClaim,      saramamocks))
diff --git a/cmd/ethconnect.go b/cmd/ethconnect.go
@@ -26,6 +26,7 @@ import (
 
 	"github.com/hyperledger/firefly-ethconnect/internal/errors"
 	"github.com/hyperledger/firefly-ethconnect/internal/kafka"
+	"github.com/hyperledger/firefly-ethconnect/internal/receipts"
 	"github.com/hyperledger/firefly-ethconnect/internal/rest"
 	"github.com/hyperledger/firefly-ethconnect/internal/utils"
 	"github.com/icza/dyno"
@@ -56,19 +57,14 @@ func initLogging(debugLevel int) {
 	switch debugLevel {
 	case 0:
 		log.SetLevel(log.ErrorLevel)
-		break
 	case 1:
 		log.SetLevel(log.InfoLevel)
-		break
 	case 2:
 		log.SetLevel(log.DebugLevel)
-		break
 	case 3:
 		log.SetLevel(log.TraceLevel)
-		break
 	default:
 		log.SetLevel(log.DebugLevel)
-		break
 	}
 	log.Debugf("Log level set to %d", debugLevel)
 }
@@ -168,33 +164,57 @@ func startServer() (err error) {
 
 	anyRoutineFinished := make(chan bool)
 	var dontPrintYaml = false
-	for name, conf := range serverConfig.KafkaBridges {
-		kafkaBridge := kafka.NewKafkaBridge(&dontPrintYaml)
-		kafkaBridge.SetConf(conf)
-		if err := kafkaBridge.ValidateConf(); err != nil {
-			return err
-		}
-		go func(name string, anyRoutineFinished chan bool) {
-			log.Infof("Starting Kafka->Ethereum bridge '%s'", name)
-			if err := kafkaBridge.Start(); err != nil {
-				log.Errorf("Kafka->Ethereum bridge failed: %s", err)
-			}
-			anyRoutineFinished <- true
-		}(name, anyRoutineFinished)
-	}
+
 	// Merge in legacy named 'webbhooks' configs
 	if serverConfig.RESTGateways == nil {
 		serverConfig.RESTGateways = make(map[string]*rest.RESTGatewayConf)
 	}
 	for name, conf := range serverConfig.Webhooks {
 		serverConfig.RESTGateways[name] = conf
 	}
+	var idempotencyCheckReceiptStore receipts.ReceiptStorePersistence
+	restGateways := make(map[string]*rest.RESTGateway)
 	for name, conf := range serverConfig.RESTGateways {
 		restGateway := rest.NewRESTGateway(&dontPrintYaml)
+		restGateways[name] = restGateway
 		restGateway.SetConf(conf)
 		if err := restGateway.ValidateConf(); err != nil {
 			return err
 		}
+		// This is a slightly awkward cross-component call, to account for the most popular pattern of usage:
+		// - Run in server mode
+		// - Single REST API Gateway
+		// - Single Kafka bridge co-located in the same process
+		// In this scenario, we can pass the receipt store to the Kafka bridge for it to do
+		// additional idempotency checks that prevent res-submission of transactions.
+		if idempotencyCheckReceiptStore == nil {
+			idempotencyCheckReceiptStore, err = restGateway.InitReceiptStore()
+			if err != nil {
+				return err
+			}
+		}
+
+	}
+
+	// Start the kafka bridges, passing in the receipt store if we have one
+	for name, conf := range serverConfig.KafkaBridges {
+		kafkaBridge := kafka.NewKafkaBridge(&dontPrintYaml)
+		kafkaBridge.SetConf(conf)
+		if err := kafkaBridge.ValidateConf(); err != nil {
+			return err
+		}
+		go func(name string, anyRoutineFinished chan bool) {
+			log.Infof("Starting Kafka->Ethereum bridge '%s'", name)
+			if err := kafkaBridge.Start(idempotencyCheckReceiptStore); err != nil {
+				log.Errorf("Kafka->Ethereum bridge failed: %s", err)
+			}
+			anyRoutineFinished <- true
+		}(name, anyRoutineFinished)
+	}
+
+	// Start the rest gateways
+	for name, rgw := range restGateways {
+		restGateway := rgw
 		go func(name string, anyRoutineFinished chan bool) {
 			log.Infof("Starting REST gateway '%s'", name)
 			if err := restGateway.Start(); err != nil {

diff --git a/internal/contractgateway/rest2eth.go b/internal/contractgateway/rest2eth.go
@@ -582,13 +582,14 @@ func (r *rest2eth) deployContract(res http.ResponseWriter, req *http.Request, fr
 		}
 	} else {
 		ack := !getFlyParamBool("noack", req) // turn on ack's by default
-		immediateReceipt := strings.EqualFold(getFlyParam("acktype", req), "receipt")
+		deployMsg.AckType = strings.ToLower(getFlyParam("acktype", req))
+		immediateReceipt := deployMsg.AckType == "receipt"
 
 		// Async messages are dispatched as generic map payloads.
 		// We are confident in the re-serialization here as we've deserialized from JSON then built our own structure
 		msgBytes, _ := json.Marshal(deployMsg)
 		var mapMsg map[string]interface{}
-		json.Unmarshal(msgBytes, &mapMsg)
+		_ = json.Unmarshal(msgBytes, &mapMsg)
 		if asyncResponse, status, err := r.asyncDispatcher.DispatchMsgAsync(req.Context(), mapMsg, ack, immediateReceipt); err != nil {
 			r.restErrReply(res, req, err, status)
 		} else {
@@ -630,7 +631,8 @@ func (r *rest2eth) sendTransaction(res http.ResponseWriter, req *http.Request, f
 		}
 	} else {
 		ack := !getFlyParamBool("noack", req) // turn on ack's by default
-		immediateReceipt := strings.EqualFold(getFlyParam("acktype", req), "receipt")
+		msg.AckType = strings.ToLower(getFlyParam("acktype", req))
+		immediateReceipt := msg.AckType == "receipt"
 
 		// Async messages are dispatched as generic map payloads.
 		// We are confident in the re-serialization here as we've deserialized from JSON then built our own structure

diff --git a/internal/contractgateway/syncdispatcher_test.go b/internal/contractgateway/syncdispatcher_test.go
@@ -21,6 +21,7 @@ import (
 
 	"github.com/hyperledger/firefly-ethconnect/internal/eth"
 	"github.com/hyperledger/firefly-ethconnect/internal/messages"
+	"github.com/hyperledger/firefly-ethconnect/internal/receipts"
 	"github.com/hyperledger/firefly-ethconnect/internal/tx"
 	"github.com/stretchr/testify/assert"
 )
@@ -58,6 +59,8 @@ func (p *mockProcessor) OnMessage(c tx.TxnContext) {
 	}
 }
 func (p *mockProcessor) Init(eth.RPCClient) {}
+func (p *mockProcessor) SetReceiptStoreForIdempotencyCheck(receiptStore receipts.ReceiptStorePersistence) {
+}
 
 type mockReplyProcessor struct {
 	err     error

diff --git a/internal/errors/errors.go b/internal/errors/errors.go
@@ -513,6 +513,10 @@ var (
 
 	// ReceiptStoreKeyNotUnique non-unique request ID
 	ReceiptStoreKeyNotUnique = e(100219, "Request ID is not unique")
+	// ReceiptErrorIdempotencyCheck failed to query receipt during idempotency check
+	ReceiptErrorIdempotencyCheck = e(100220, "Failed querying the receipt store, performing duplicate message check on ackmode=receipt for id %s: %s")
+	// ResubmissionPreventedCheckTransactionHash redelivery was prevented by the processor
+	ResubmissionPreventedCheckTransactionHash = e(100221, "Resubmission of this transaction was prevented by the REST API Gateway. Check the status of the transaction by the transaction hash")
 )
 
 type EthconnectError interface {

diff --git a/internal/kafka/kafkabridge.go b/internal/kafka/kafkabridge.go
@@ -27,6 +27,7 @@ import (
 	"github.com/hyperledger/firefly-ethconnect/internal/errors"
 	"github.com/hyperledger/firefly-ethconnect/internal/eth"
 	"github.com/hyperledger/firefly-ethconnect/internal/messages"
+	"github.com/hyperledger/firefly-ethconnect/internal/receipts"
 	"github.com/hyperledger/firefly-ethconnect/internal/tx"
 	"github.com/hyperledger/firefly-ethconnect/internal/utils"
 	log "github.com/sirupsen/logrus"
@@ -87,7 +88,7 @@ func (k *KafkaBridge) CobraInit() (cmd *cobra.Command) {
 		Short: "Kafka->Ethereum (JSON/RPC) Bridge",
 		RunE: func(cmd *cobra.Command, args []string) (err error) {
 			log.Infof("Starting Kafka bridge")
-			err = k.Start()
+			err = k.Start(nil /* only available in full server mode with co-located REST API Gateway */)
 			return
 		},
 		PreRunE: func(cmd *cobra.Command, args []string) (err error) {
@@ -145,7 +146,7 @@ func (k *KafkaBridge) addInflightMsg(msg *sarama.ConsumerMessage, producer Kafka
 	// is very important that the consumer of the wrapped context object calls Reply
 	pCtx = &ctx
 	k.inFlight[ctx.reqOffset] = pCtx
-	log.Infof("Message now in-flight: %s", pCtx)
+	log.Debugf("Message now in-flight: %s", pCtx)
 	// Attempt to process the headers from the original message,
 	// which could fail. In which case we still have a msgContext inflight
 	// that needs Reply (and offset commit). So our caller must
@@ -200,19 +201,19 @@ func (k *KafkaBridge) setInFlightComplete(ctx *msgContext, consumer KafkaConsume
 
 	// Build an offset sorted list of the inflight
 	ctx.complete = true
-	var completeInParition []*msgContext
+	var completeInPartition []*msgContext
 	for _, inflight := range k.inFlight {
 		if inflight.saramaMsg.Partition == ctx.saramaMsg.Partition {
-			completeInParition = append(completeInParition, inflight)
+			completeInPartition = append(completeInPartition, inflight)
 		}
 	}
-	sort.Sort(ctxByOffset(completeInParition))
+	sort.Sort(ctxByOffset(completeInPartition))
 
 	// Go forwards until the first that isn't complete
 	var readyToAck []*msgContext
-	for i := 0; i < len(completeInParition); i++ {
-		if completeInParition[i].complete {
-			readyToAck = append(readyToAck, completeInParition[i])
+	for i := 0; i < len(completeInPartition); i++ {
+		if completeInPartition[i].complete {
+			readyToAck = append(readyToAck, completeInPartition[i])
 		} else {
 			break
 		}
@@ -221,7 +222,7 @@ func (k *KafkaBridge) setInFlightComplete(ctx *msgContext, consumer KafkaConsume
 	canMark := len(readyToAck) > 0
 	log.Debugf("Ready=%d:%d CanMark=%t Infight=%d InflightSamePartition=%d ReadyToAck=%d",
 		ctx.saramaMsg.Partition, ctx.saramaMsg.Offset, canMark,
-		len(k.inFlight), len(completeInParition), len(readyToAck))
+		len(k.inFlight), len(completeInPartition), len(readyToAck))
 	if canMark {
 		// Remove all the ready-to-acks from the in-flight list
 		for i := 0; i < len(readyToAck); i++ {
@@ -345,7 +346,7 @@ func (k *KafkaBridge) ConsumerMessagesLoop(consumer KafkaConsumer, producer Kafk
 	log.Debugf("Kafka consumer loop started")
 	for msg := range consumer.Messages() {
 		k.inFlightCond.L.Lock()
-		log.Infof("Kafka consumer received message: Partition=%d Offset=%d", msg.Partition, msg.Offset)
+		log.Debugf("Kafka consumer received message: Partition=%d Offset=%d", msg.Partition, msg.Offset)
 
 		// We cannot build up an infinite number of messages in memory
 		for len(k.inFlight) >= k.conf.MaxInFlight {
@@ -361,6 +362,8 @@ func (k *KafkaBridge) ConsumerMessagesLoop(consumer KafkaConsumer, producer Kafk
 			// This was a dup
 		} else if err == nil {
 			// Dispatch for processing if we parsed the message successfully
+			headers := msgCtx.Headers()
+			log.Infof("Kafka consumer dispatching message '%s' Type=%s ID=%s", msgCtx.reqOffset, headers.MsgType, headers.ID)
 			k.processor.OnMessage(msgCtx)
 		} else {
 			// Dispatch a generic 'bad data' reply
@@ -399,7 +402,8 @@ func (k *KafkaBridge) ProducerSuccessLoop(consumer KafkaConsumer, producer Kafka
 		k.inFlightCond.L.Lock()
 		reqOffset := msg.Metadata.(string)
 		if ctx, ok := k.inFlight[reqOffset]; ok {
-			log.Infof("Reply sent: %s", ctx)
+			headers := ctx.Headers()
+			log.Infof("Kafka consumer replying to message '%s' Type=%s ID=%s", ctx.reqOffset, headers.MsgType, headers.ID)
 			// While still holding the lock, add this to the completed list
 			_ = k.setInFlightComplete(ctx, consumer)
 			// We've reduced the in-flight count - wake any waiting consumer go func
@@ -423,7 +427,9 @@ func (k *KafkaBridge) connect() (err error) {
 }
 
 // Start kicks off the bridge
-func (k *KafkaBridge) Start() (err error) {
+func (k *KafkaBridge) Start(receiptStore receipts.ReceiptStorePersistence) (err error) {
+
+	k.processor.SetReceiptStoreForIdempotencyCheck(receiptStore)
 
 	if *k.printYAML {
 		b, err := utils.MarshalToYAML(&k.conf)

diff --git a/internal/kafka/kafkabridge_test.go b/internal/kafka/kafkabridge_test.go
@@ -29,6 +29,7 @@ import (
 	"github.com/hyperledger/firefly-ethconnect/internal/errors"
 	"github.com/hyperledger/firefly-ethconnect/internal/eth"
 	"github.com/hyperledger/firefly-ethconnect/internal/messages"
+	"github.com/hyperledger/firefly-ethconnect/internal/receipts"
 	"github.com/hyperledger/firefly-ethconnect/internal/tx"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
@@ -93,6 +94,10 @@ func (p *testKafkaMsgProcessor) OnMessage(msg tx.TxnContext) {
 	p.messages <- msg
 	return
 }
+
+func (p *testKafkaMsgProcessor) SetReceiptStoreForIdempotencyCheck(receiptStore receipts.ReceiptStorePersistence) {
+}
+
 func TestNewKafkaBridge(t *testing.T) {
 	assert := assert.New(t)
 

diff --git a/internal/messages/messages.go b/internal/messages/messages.go
@@ -36,6 +36,8 @@ const (
 	MsgTypeTransactionSuccess = "TransactionSuccess"
 	// MsgTypeTransactionFailure - a transaction receipt where status is 0
 	MsgTypeTransactionFailure = "TransactionFailure"
+	// MsgTypeTransactionRedeliveryPrevented - idempotency check caught a redelivery of the message
+	MsgTypeTransactionRedeliveryPrevented = "TransactionRedeliveryPrevented"
 	// RecordHeaderAccessToken - record header name for passing JWT token over messaging
 	RecordHeaderAccessToken = "fly-accesstoken"
 )
@@ -194,6 +196,14 @@ type TransactionReceipt struct {
 	RegisterAs           string                `json:"registerAs,omitempty"`
 }
 
+// TransactionRedeliveryNotification is sent on redelivery of a message, when the ackmode=receipt
+// idempotency check is enabled. The REST API Gateway (or other consumer), should avoid overwriting
+// any received receipt when it gets this.
+type TransactionRedeliveryNotification struct {
+	ReplyCommon
+	TransactionHash string `json:"transactionHash"`
+}
+
 // TransactionInfo is the detailed transaction info returned by eth_getTransactionByXXXXX
 // For the big numbers, we pass a simple string as well as a full
 // ethereum hex encoding version