Resolve DX operations via requestID (not event ID)

internal/dataexchange/ffdx/dxevent.go

@@ -45,16 +45,21 @@ type wsEvent struct {
 type dxEvent struct {
 	ffdx                *FFDX
 	id                  string
+	requestID           string
 	dxType              dataexchange.DXEventType
 	messageReceived     *dataexchange.MessageReceived
 	privateBlobReceived *dataexchange.PrivateBlobReceived
 	transferResult      *dataexchange.TransferResult
 }
 
-func (e *dxEvent) NamespacedID() string {
+func (e *dxEvent) EventID() string {
 	return e.id
 }
 
+func (e *dxEvent) NamespacedID() string {
+	return e.requestID
+}
+
 func (e *dxEvent) Type() dataexchange.DXEventType {
 	return e.dxType
 }
@@ -88,7 +93,7 @@ func (e *dxEvent) TransferResult() *dataexchange.TransferResult {
 
 func (h *FFDX) dispatchEvent(msg *wsEvent) {
 	var err error
-	e := &dxEvent{ffdx: h, id: msg.EventID}
+	e := &dxEvent{ffdx: h, id: msg.EventID, requestID: msg.RequestID}
 	switch msg.Type {
 	case messageFailed:
 		e.dxType = dataexchange.DXEventTypeTransferResult

internal/dataexchange/ffdx/ffdx_test.go

@@ -446,7 +446,8 @@ func TestEvents(t *testing.T) {
 	h.SetHandler(mcb)
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.NamespacedID() == "1" &&
+		return ev.EventID() == "1" &&
+			ev.NamespacedID() == "tx12345" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusFailed &&
@@ -457,7 +458,7 @@ func TestEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"1"}`, string(msg))
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.NamespacedID() == "2" &&
+		return ev.EventID() == "2" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusSucceeded
@@ -467,7 +468,7 @@ func TestEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"2"}`, string(msg))
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.NamespacedID() == "3" &&
+		return ev.EventID() == "3" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusSucceeded &&
@@ -479,7 +480,7 @@ func TestEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"3"}`, string(msg))
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.NamespacedID() == "4" &&
+		return ev.EventID() == "4" &&
 			ev.Type() == dataexchange.DXEventTypeMessageReceived &&
 			ev.MessageReceived().PeerID == "peer1" &&
 			string(ev.MessageReceived().Data) == "message1"
@@ -489,7 +490,7 @@ func TestEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"4","manifest":"{\"manifest\":true}"}`, string(msg))
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.NamespacedID() == "5" &&
+		return ev.EventID() == "5" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusFailed &&
@@ -500,7 +501,7 @@ func TestEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"5"}`, string(msg))
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.NamespacedID() == "6" &&
+		return ev.EventID() == "6" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusSucceeded &&
@@ -521,7 +522,7 @@ func TestEvents(t *testing.T) {
 
 	hash := fftypes.NewRandB32()
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.NamespacedID() == "9" &&
+		return ev.EventID() == "9" &&
 			ev.Type() == dataexchange.DXEventTypePrivateBlobReceived &&
 			ev.PrivateBlobReceived().Hash.Equals(hash)
 	})).Run(acker()).Return(nil)
@@ -530,7 +531,7 @@ func TestEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"9"}`, string(msg))
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.NamespacedID() == "10" &&
+		return ev.EventID() == "10" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusSucceeded &&
@@ -560,7 +561,7 @@ func TestEventsWithManifest(t *testing.T) {
 	h.SetHandler(mcb)
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.NamespacedID() == "1" &&
+		return ev.EventID() == "1" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().Status == core.OpStatusPending
 	})).Run(acker()).Return(nil)
@@ -569,7 +570,7 @@ func TestEventsWithManifest(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"1"}`, string(msg))
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.NamespacedID() == "2" &&
+		return ev.EventID() == "2" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().Status == core.OpStatusPending
 	})).Run(acker()).Return(nil)

pkg/dataexchange/plugin.go

@@ -108,6 +108,7 @@ type DXEventType int
 
 // DXEvent is a single interface that can be passed to all events
 type DXEvent interface {
+	EventID() string
 	NamespacedID() string
 	Ack()
 	AckWithManifest(manifest string)

test/e2e/e2e_test.go

@@ -437,3 +437,26 @@ func checkObject(t *testing.T, expected interface{}, actual interface{}) bool {
 	}
 	return match
 }
+
+func verifyAllOperationsSucceeded(t *testing.T, clients []*resty.Client, startTime time.Time) {
+	tries := 3
+	delay := 2 * time.Second
+
+	var pending string
+	for i := 0; i < tries; i++ {
+		pending = ""
+		for _, client := range clients {
+			for _, op := range GetOperations(t, client, startTime) {
+				if op.Status != core.OpStatusSucceeded {
+					pending += fmt.Sprintf("Operation '%s' (%s) on '%s' is not successful\n", op.ID, op.Type, client.BaseURL)
+				}
+			}
+		}
+		if pending == "" {
+			return
+		}
+		time.Sleep(delay)
+	}
+
+	assert.Fail(t, pending)
+}

test/e2e/ethereum_contract_test.go

@@ -136,6 +136,10 @@ func (suite *EthereumContractTestSuite) BeforeTest(suiteName, testName string) {
 	suite.testState = beforeE2ETest(suite.T())
 }
 
+func (suite *EthereumContractTestSuite) AfterTest(suiteName, testName string) {
+	verifyAllOperationsSucceeded(suite.T(), []*resty.Client{suite.testState.client1, suite.testState.client2}, suite.testState.startTime)
+}
+
 func (suite *EthereumContractTestSuite) TestDirectInvokeMethod() {
 	defer suite.testState.done()
 

test/e2e/fabric_contract_test.go

@@ -138,6 +138,10 @@ func (suite *FabricContractTestSuite) BeforeTest(suiteName, testName string) {
 	suite.testState = beforeE2ETest(suite.T())
 }
 
+func (suite *FabricContractTestSuite) AfterTest(suiteName, testName string) {
+	verifyAllOperationsSucceeded(suite.T(), []*resty.Client{suite.testState.client1, suite.testState.client2}, suite.testState.startTime)
+}
+
 func (suite *FabricContractTestSuite) TestE2EContractEvents() {
 	defer suite.testState.done()
 

test/e2e/identity_test.go

@@ -22,6 +22,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/go-resty/resty/v2"
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
@@ -38,6 +39,10 @@ func (suite *IdentityTestSuite) BeforeTest(suiteName, testName string) {
 	suite.testState = beforeE2ETest(suite.T())
 }
 
+func (suite *IdentityTestSuite) AfterTest(suiteName, testName string) {
+	verifyAllOperationsSucceeded(suite.T(), []*resty.Client{suite.testState.client1, suite.testState.client2}, suite.testState.startTime)
+}
+
 func (suite *IdentityTestSuite) TestCustomChildIdentityBroadcasts() {
 	defer suite.testState.done()
 

test/e2e/onchain_offchain_test.go

@@ -44,6 +44,10 @@ func (suite *OnChainOffChainTestSuite) BeforeTest(suiteName, testName string) {
 	suite.testState = beforeE2ETest(suite.T())
 }
 
+func (suite *OnChainOffChainTestSuite) AfterTest(suiteName, testName string) {
+	verifyAllOperationsSucceeded(suite.T(), []*resty.Client{suite.testState.client1, suite.testState.client2}, suite.testState.startTime)
+}
+
 func (suite *OnChainOffChainTestSuite) TestE2EBroadcast() {
 	defer suite.testState.done()
 

test/e2e/restclient_test.go

@@ -64,6 +64,7 @@ var (
 	urlContractListeners = "/namespaces/default/contracts/listeners"
 	urlContractAPI       = "/namespaces/default/apis"
 	urlBlockchainEvents  = "/namespaces/default/blockchainevents"
+	urlOperations        = "/namespaces/default/operations"
 	urlGetOrganizations  = "/namespaces/default/network/organizations"
 	urlGetOrgKeys        = "/namespaces/default/identities/%s/verifiers"
 )
@@ -817,3 +818,14 @@ func GetBlockchainEvent(t *testing.T, client *resty.Client, eventID string) (int
 	require.Equal(t, 200, resp.StatusCode(), "GET %s [%d]: %s", path, resp.StatusCode(), resp.String())
 	return res, err
 }
+
+func GetOperations(t *testing.T, client *resty.Client, startTime time.Time) (operations []*core.Operation) {
+	path := urlOperations
+	resp, err := client.R().
+		SetQueryParam("created", fmt.Sprintf(">%d", startTime.UnixNano())).
+		SetResult(&operations).
+		Get(path)
+	require.NoError(t, err)
+	require.Equal(t, 200, resp.StatusCode(), "GET %s [%d]: %s", path, resp.StatusCode(), resp.String())
+	return operations
+}

test/e2e/tokens_test.go

@@ -21,6 +21,7 @@ import (
 	"math/rand"
 	"time"
 
+	"github.com/go-resty/resty/v2"
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
@@ -43,6 +44,10 @@ func (suite *TokensTestSuite) BeforeTest(suiteName, testName string) {
 	suite.testState = beforeE2ETest(suite.T())
 }
 
+func (suite *TokensTestSuite) AfterTest(suiteName, testName string) {
+	verifyAllOperationsSucceeded(suite.T(), []*resty.Client{suite.testState.client1, suite.testState.client2}, suite.testState.startTime)
+}
+
 func (suite *TokensTestSuite) TestE2EFungibleTokensAsync() {
 	defer suite.testState.done()