Refactor of ChaincodeEvents service implementation to support resume (h…

…yperledger#3283)

Enables the client to (optionally) specify an AfterTransactionId property in addition to a start block number when requesting chaincode events, which causes chaincode events up to that transaction ID (inclusive) to be ignored and not returned to the client. This supports resume of chaincode event listening on client reconnect without duplicating or missing any events.

Signed-off-by: Mark S. Lewis <Mark.S.Lewis@outlook.com>

go.mod

@@ -30,7 +30,7 @@ require (
 	github.com/hyperledger/fabric-chaincode-go v0.0.0-20201119163726-f8ef75b17719
 	github.com/hyperledger/fabric-config v0.1.0
 	github.com/hyperledger/fabric-lib-go v1.0.0
-	github.com/hyperledger/fabric-protos-go v0.0.0-20211118165945-23d738fc3553
+	github.com/hyperledger/fabric-protos-go v0.0.0-20220315113721-7dc293e117f7
 	github.com/kr/pretty v0.2.1
 	github.com/magiconair/properties v1.8.1 // indirect
 	github.com/mattn/go-runewidth v0.0.4 // indirect

go.sum

@@ -171,8 +171,8 @@ github.com/hyperledger/fabric-lib-go v1.0.0/go.mod h1:H362nMlunurmHwkYqR5uHL2UDW
 github.com/hyperledger/fabric-protos-go v0.0.0-20190919234611-2a87503ac7c9/go.mod h1:xVYTjK4DtZRBxZ2D9aE4y6AbLaPwue2o/criQyQbVD0=
 github.com/hyperledger/fabric-protos-go v0.0.0-20200424173316-dd554ba3746e/go.mod h1:xVYTjK4DtZRBxZ2D9aE4y6AbLaPwue2o/criQyQbVD0=
 github.com/hyperledger/fabric-protos-go v0.0.0-20210911123859-041d13f0980c/go.mod h1:xVYTjK4DtZRBxZ2D9aE4y6AbLaPwue2o/criQyQbVD0=
-github.com/hyperledger/fabric-protos-go v0.0.0-20211118165945-23d738fc3553 h1:E9f0v1q4EDfrE+0LdkxVtdYKAZ7PGCaj1bBx45R9yEQ=
-github.com/hyperledger/fabric-protos-go v0.0.0-20211118165945-23d738fc3553/go.mod h1:xVYTjK4DtZRBxZ2D9aE4y6AbLaPwue2o/criQyQbVD0=
+github.com/hyperledger/fabric-protos-go v0.0.0-20220315113721-7dc293e117f7 h1:YV+siZuYQZwENjRH00t7ZS0CTlywt8Qog/SzL/jf6kE=
+github.com/hyperledger/fabric-protos-go v0.0.0-20220315113721-7dc293e117f7/go.mod h1:xVYTjK4DtZRBxZ2D9aE4y6AbLaPwue2o/criQyQbVD0=
 github.com/inconshreveable/mousetrap v1.0.0 h1:Z8tu5sraLXCXIcARxBp/8cbvlwVa7Z1NHg9XEKhtSvM=
 github.com/inconshreveable/mousetrap v1.0.0/go.mod h1:PxqpIevigyE2G7u3NXJIT2ANytuPF1OarO4DADm73n8=
 github.com/jonboulle/clockwork v0.1.0/go.mod h1:Ii8DK3G1RaLaWxj9trq07+26W01tbo22gdxWY5EU2bo=

integration/gateway/gateway_test.go

@@ -212,6 +212,7 @@ var _ = Describe("GatewayService", func() {
 	chaincodeEvents := func(
 		ctx context.Context,
 		startPosition *orderer.SeekPosition,
+		afterTxID string,
 		identity func() ([]byte, error),
 		sign func(msg []byte) ([]byte, error),
 	) (gateway.Gateway_ChaincodeEventsClient, error) {
@@ -226,6 +227,9 @@ var _ = Describe("GatewayService", func() {
 		if startPosition != nil {
 			request.StartPosition = startPosition
 		}
+		if len(afterTxID) > 0 {
+			request.AfterTransactionId = afterTxID
+		}
 
 		requestBytes, err := proto.Marshal(request)
 		Expect(err).NotTo(HaveOccurred())
@@ -375,7 +379,7 @@ var _ = Describe("GatewayService", func() {
 				},
 			}
 
-			eventsClient, err := chaincodeEvents(eventCtx, startPosition, signingIdentity.Serialize, signingIdentity.Sign)
+			eventsClient, err := chaincodeEvents(eventCtx, startPosition, "", signingIdentity.Serialize, signingIdentity.Sign)
 			Expect(err).NotTo(HaveOccurred())
 
 			_, transactionID := submitTransaction("event", []byte("EVENT_NAME"), []byte("EVENT_PAYLOAD"))
@@ -409,7 +413,7 @@ var _ = Describe("GatewayService", func() {
 				},
 			}
 
-			eventsClient, err := chaincodeEvents(eventCtx, startPosition, signingIdentity.Serialize, signingIdentity.Sign)
+			eventsClient, err := chaincodeEvents(eventCtx, startPosition, "", signingIdentity.Serialize, signingIdentity.Sign)
 			Expect(err).NotTo(HaveOccurred())
 
 			event, err := eventsClient.Recv()
@@ -426,13 +430,50 @@ var _ = Describe("GatewayService", func() {
 			Expect(proto.Equal(event.Events[0], expectedEvent)).To(BeTrue(), "Expected\n\t%#v\nto proto.Equal\n\t%#v", event.Events[0], expectedEvent)
 		})
 
+		It("should respond with replayed chaincode events after specified transaction ID", func() {
+			_, afterTransactionID := submitTransaction("event", []byte("WRONG_EVENT_NAME"), []byte("WRONG_EVENT_PAYLOAD"))
+			_, nextTransactionID := submitTransaction("event", []byte("CORRECT_EVENT_NAME"), []byte("CORRECT_EVENT_PAYLOAD"))
+
+			statusResult, err := commitStatus(afterTransactionID, signingIdentity.Serialize, signingIdentity.Sign)
+			Expect(err).NotTo(HaveOccurred())
+
+			_, err = commitStatus(nextTransactionID, signingIdentity.Serialize, signingIdentity.Sign)
+			Expect(err).NotTo(HaveOccurred())
+
+			eventCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
+			defer cancel()
+
+			startPosition := &orderer.SeekPosition{
+				Type: &orderer.SeekPosition_Specified{
+					Specified: &orderer.SeekSpecified{
+						Number: statusResult.BlockNumber,
+					},
+				},
+			}
+
+			eventsClient, err := chaincodeEvents(eventCtx, startPosition, afterTransactionID, signingIdentity.Serialize, signingIdentity.Sign)
+			Expect(err).NotTo(HaveOccurred())
+
+			event, err := eventsClient.Recv()
+			Expect(err).NotTo(HaveOccurred())
+
+			Expect(event.Events).To(HaveLen(1), "number of events")
+			expectedEvent := &peer.ChaincodeEvent{
+				ChaincodeId: "gatewaycc",
+				TxId:        nextTransactionID,
+				EventName:   "CORRECT_EVENT_NAME",
+				Payload:     []byte("CORRECT_EVENT_PAYLOAD"),
+			}
+			Expect(proto.Equal(event.Events[0], expectedEvent)).To(BeTrue(), "Expected\n\t%#v\nto proto.Equal\n\t%#v", event.Events[0], expectedEvent)
+		})
+
 		It("should default to next commit if start position not specified", func() {
 			eventCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
 			defer cancel()
 
 			var startPosition *orderer.SeekPosition
 
-			eventsClient, err := chaincodeEvents(eventCtx, startPosition, signingIdentity.Serialize, signingIdentity.Sign)
+			eventsClient, err := chaincodeEvents(eventCtx, startPosition, "", signingIdentity.Serialize, signingIdentity.Sign)
 			Expect(err).NotTo(HaveOccurred())
 
 			_, transactionID := submitTransaction("event", []byte("EVENT_NAME"), []byte("EVENT_PAYLOAD"))
@@ -462,7 +503,7 @@ var _ = Describe("GatewayService", func() {
 				},
 			}
 
-			eventsClient, err := chaincodeEvents(eventCtx, startPosition, badIdentity.Serialize, signingIdentity.Sign)
+			eventsClient, err := chaincodeEvents(eventCtx, startPosition, "", badIdentity.Serialize, signingIdentity.Sign)
 			Expect(err).NotTo(HaveOccurred())
 
 			event, err := eventsClient.Recv()
@@ -486,7 +527,7 @@ var _ = Describe("GatewayService", func() {
 				},
 			}
 
-			eventsClient, err := chaincodeEvents(eventCtx, startPosition, signingIdentity.Serialize, badSign)
+			eventsClient, err := chaincodeEvents(eventCtx, startPosition, "", signingIdentity.Serialize, badSign)
 			Expect(err).NotTo(HaveOccurred())
 
 			event, err := eventsClient.Recv()

internal/pkg/gateway/api.go

@@ -548,6 +548,8 @@ func (gs *Server) ChaincodeEvents(signedRequest *gp.SignedChaincodeEventsRequest
 		return err
 	}
 
+	isMatch := chaincodeEventMatcher(request)
+
 	ledgerIter, err := ledger.GetBlocksIterator(startBlock)
 	if err != nil {
 		return status.Error(codes.Aborted, err.Error())
@@ -563,9 +565,8 @@ func (gs *Server) ChaincodeEvents(signedRequest *gp.SignedChaincodeEventsRequest
 		}
 
 		var matchingEvents []*peer.ChaincodeEvent
-
 		for _, event := range response.Events {
-			if event.GetChaincodeId() == request.GetChaincodeId() {
+			if isMatch(event) {
 				matchingEvents = append(matchingEvents, event)
 			}
 		}
@@ -586,6 +587,30 @@ func (gs *Server) ChaincodeEvents(signedRequest *gp.SignedChaincodeEventsRequest
 	}
 }
 
+func chaincodeEventMatcher(request *gp.ChaincodeEventsRequest) func(event *peer.ChaincodeEvent) bool {
+	chaincodeID := request.GetChaincodeId()
+	previousTransactionID := request.GetAfterTransactionId()
+
+	if len(previousTransactionID) == 0 {
+		return func(event *peer.ChaincodeEvent) bool {
+			return event.GetChaincodeId() == chaincodeID
+		}
+	}
+
+	passedPreviousTransaction := false
+
+	return func(event *peer.ChaincodeEvent) bool {
+		if !passedPreviousTransaction {
+			if event.TxId == previousTransactionID {
+				passedPreviousTransaction = true
+			}
+			return false
+		}
+
+		return event.GetChaincodeId() == chaincodeID
+	}
+}
+
 func startBlockFromLedgerPosition(ledger ledger.Ledger, position *ab.SeekPosition) (uint64, error) {
 	switch seek := position.GetType().(type) {
 	case nil: