Repository: hyperledger/firefly
Pull request #955 (2022-08-10 20:57:37)

Properly match operation updates to the plugin that generated them


Patch:
diff --git a/internal/blockchain/common/common.go b/internal/blockchain/common/common.go
index d5003146b..9786f6a0a 100644
--- a/internal/blockchain/common/common.go
+++ b/internal/blockchain/common/common.go
@@ -94,7 +94,8 @@ func (cb *callbacks) SetOperationalHandler(namespace string, handler core.Operat
 func (cb *callbacks) OperationUpdate(ctx context.Context, plugin core.Named, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
 	namespace, _, _ := core.ParseNamespacedOpID(ctx, nsOpID)
 	if handler, ok := cb.opHandlers[namespace]; ok {
-		handler.OperationUpdate(plugin, &core.OperationUpdate{
+		handler.OperationUpdate(&core.OperationUpdate{
+			Plugin:         plugin.Name(),
 			NamespacedOpID: nsOpID,
 			Status:         status,
 			BlockchainTXID: blockchainTXID,
diff --git a/internal/dataexchange/ffdx/dxevent.go b/internal/dataexchange/ffdx/dxevent.go
index 835819abd..4b26547e7 100644
--- a/internal/dataexchange/ffdx/dxevent.go
+++ b/internal/dataexchange/ffdx/dxevent.go
@@ -90,6 +90,7 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 	switch msg.Type {
 	case messageFailed:
 		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			Plugin:         h.Name(),
 			NamespacedOpID: msg.RequestID,
 			Status:         core.OpStatusFailed,
 			ErrorMessage:   msg.Error,
@@ -103,6 +104,7 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 			status = core.OpStatusPending
 		}
 		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			Plugin:         h.Name(),
 			NamespacedOpID: msg.RequestID,
 			Status:         status,
 			Output:         msg.Info,
@@ -111,6 +113,7 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 		return
 	case messageAcknowledged:
 		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			Plugin:         h.Name(),
 			NamespacedOpID: msg.RequestID,
 			Status:         core.OpStatusSucceeded,
 			VerifyManifest: h.capabilities.Manifest,
@@ -121,6 +124,7 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 		return
 	case blobFailed:
 		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			Plugin:         h.Name(),
 			NamespacedOpID: msg.RequestID,
 			Status:         core.OpStatusFailed,
 			ErrorMessage:   msg.Error,
@@ -134,6 +138,7 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 			status = core.OpStatusPending
 		}
 		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			Plugin:         h.Name(),
 			NamespacedOpID: msg.RequestID,
 			Status:         status,
 			Output:         msg.Info,
@@ -142,6 +147,7 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 		return
 	case blobAcknowledged:
 		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			Plugin:         h.Name(),
 			NamespacedOpID: msg.RequestID,
 			Status:         core.OpStatusSucceeded,
 			Output:         msg.Info,
diff --git a/internal/dataexchange/ffdx/ffdx.go b/internal/dataexchange/ffdx/ffdx.go
index 12e79b4b4..6c71fd1ae 100644
--- a/internal/dataexchange/ffdx/ffdx.go
+++ b/internal/dataexchange/ffdx/ffdx.go
@@ -65,7 +65,7 @@ type callbacks struct {
 func (cb *callbacks) OperationUpdate(ctx context.Context, update *core.OperationUpdate) {
 	namespace, _, _ := core.ParseNamespacedOpID(ctx, update.NamespacedOpID)
 	if handler, ok := cb.opHandlers[namespace]; ok {
-		handler.OperationUpdate(cb.plugin, update)
+		handler.OperationUpdate(update)
 	} else {
 		log.L(ctx).Errorf("No handler found for DX operation '%s'", update.NamespacedOpID)
 		update.OnComplete()
diff --git a/internal/operations/manager.go b/internal/operations/manager.go
index 905e89920..388547c2a 100644
--- a/internal/operations/manager.go
+++ b/internal/operations/manager.go
@@ -42,7 +42,7 @@ type Manager interface {
 	RunOperation(ctx context.Context, op *core.PreparedOperation, options ...RunOperationOption) (fftypes.JSONObject, error)
 	RetryOperation(ctx context.Context, opID *fftypes.UUID) (*core.Operation, error)
 	AddOrReuseOperation(ctx context.Context, op *core.Operation) error
-	SubmitOperationUpdate(plugin core.Named, update *core.OperationUpdate)
+	SubmitOperationUpdate(update *core.OperationUpdate)
 	ResolveOperationByID(ctx context.Context, opID *fftypes.UUID, op *core.OperationUpdateDTO) error
 	Start() error
 	WaitStop()
@@ -181,12 +181,12 @@ func (om *operationsManager) ResolveOperationByID(ctx context.Context, opID *fft
 	return om.database.ResolveOperation(ctx, om.namespace, opID, op.Status, op.Error, op.Output)
 }
 
-func (om *operationsManager) SubmitOperationUpdate(plugin core.Named, update *core.OperationUpdate) {
+func (om *operationsManager) SubmitOperationUpdate(update *core.OperationUpdate) {
 	errString := ""
 	if update.ErrorMessage != "" {
 		errString = fmt.Sprintf(" error=%s", update.ErrorMessage)
 	}
-	log.L(om.ctx).Debugf("%s updating operation %s status=%s%s", plugin.Name(), update.NamespacedOpID, update.Status, errString)
+	log.L(om.ctx).Debugf("%s updating operation %s status=%s%s", update.Plugin, update.NamespacedOpID, update.Status, errString)
 	om.updater.SubmitOperationUpdate(om.ctx, update)
 }
 
diff --git a/internal/operations/operation_updater.go b/internal/operations/operation_updater.go
index c972e5314..43896a7c5 100644
--- a/internal/operations/operation_updater.go
+++ b/internal/operations/operation_updater.go
@@ -207,6 +207,7 @@ func (ou *operationUpdater) doBatchUpdate(ctx context.Context, updates []*core.O
 	if len(opIDs) == 0 {
 		return nil
 	}
+	// TODO: cache these operation queries
 	opFilter := database.OperationQueryFactory.NewFilter(ctx).In("id", opIDs)
 	ops, _, err := ou.database.GetOperations(ctx, ou.manager.namespace, opFilter)
 	if err != nil {
@@ -241,7 +242,7 @@ func (ou *operationUpdater) doBatchUpdate(ctx context.Context, updates []*core.O
 
 func (ou *operationUpdater) doUpdate(ctx context.Context, update *core.OperationUpdate, ops []*core.Operation, transactions []*core.Transaction) error {
 
-	_, id, err := core.ParseNamespacedOpID(ctx, update.NamespacedOpID)
+	_, updateID, err := core.ParseNamespacedOpID(ctx, update.NamespacedOpID)
 	if err != nil {
 		log.L(ctx).Warnf("Unable to update operation '%s' due to invalid ID: %s", update.NamespacedOpID, err)
 		return nil
@@ -250,7 +251,11 @@ func (ou *operationUpdater) doUpdate(ctx context.Context, update *core.Operation
 	// Find the operation we already retrieved, and do the update
 	var op *core.Operation
 	for _, candidate := range ops {
-		if id.Equals(candidate.ID) {
+		if updateID.Equals(candidate.ID) {
+			if update.Plugin != candidate.Plugin {
+				log.L(ctx).Debugf("Operation update '%s' from '%s' ignored, as it does not match operation source '%s'", update.NamespacedOpID, update.Plugin, candidate.Plugin)
+				return nil
+			}
 			op = candidate
 			break
 		}
diff --git a/internal/orchestrator/bound_callbacks.go b/internal/orchestrator/bound_callbacks.go
index 368974283..addc4ab13 100644
--- a/internal/orchestrator/bound_callbacks.go
+++ b/internal/orchestrator/bound_callbacks.go
@@ -30,8 +30,8 @@ type boundCallbacks struct {
 	om operations.Manager
 }
 
-func (bc *boundCallbacks) OperationUpdate(plugin core.Named, update *core.OperationUpdate) {
-	bc.om.SubmitOperationUpdate(plugin, update)
+func (bc *boundCallbacks) OperationUpdate(update *core.OperationUpdate) {
+	bc.om.SubmitOperationUpdate(update)
 }
 
 func (bc *boundCallbacks) SharedStorageBatchDownloaded(payloadRef string, data []byte) (*fftypes.UUID, error) {
diff --git a/internal/tokens/fftokens/fftokens.go b/internal/tokens/fftokens/fftokens.go
index 905908fd5..e3bc57583 100644
--- a/internal/tokens/fftokens/fftokens.go
+++ b/internal/tokens/fftokens/fftokens.go
@@ -51,7 +51,8 @@ type callbacks struct {
 func (cb *callbacks) OperationUpdate(ctx context.Context, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
 	namespace, _, _ := core.ParseNamespacedOpID(ctx, nsOpID)
 	if handler, ok := cb.opHandlers[namespace]; ok {
-		handler.OperationUpdate(cb.plugin, &core.OperationUpdate{
+		handler.OperationUpdate(&core.OperationUpdate{
+			Plugin:         cb.plugin.Name(),
 			NamespacedOpID: nsOpID,
 			Status:         status,
 			BlockchainTXID: blockchainTXID,
diff --git a/mocks/coremocks/operation_callbacks.go b/mocks/coremocks/operation_callbacks.go
index 37a3aa0e5..97a667c8e 100644
--- a/mocks/coremocks/operation_callbacks.go
+++ b/mocks/coremocks/operation_callbacks.go
@@ -12,7 +12,7 @@ type OperationCallbacks struct {
 	mock.Mock
 }
 
-// OperationUpdate provides a mock function with given fields: plugin, update
-func (_m *OperationCallbacks) OperationUpdate(plugin core.Named, update *core.OperationUpdate) {
-	_m.Called(plugin, update)
+// OperationUpdate provides a mock function with given fields: update
+func (_m *OperationCallbacks) OperationUpdate(update *core.OperationUpdate) {
+	_m.Called(update)
 }
diff --git a/mocks/operationmocks/manager.go b/mocks/operationmocks/manager.go
index 88c74807f..a3555b3f6 100644
--- a/mocks/operationmocks/manager.go
+++ b/mocks/operationmocks/manager.go
@@ -141,9 +141,9 @@ func (_m *Manager) Start() error {
 	return r0
 }
 
-// SubmitOperationUpdate provides a mock function with given fields: plugin, update
-func (_m *Manager) SubmitOperationUpdate(plugin core.Named, update *core.OperationUpdate) {
-	_m.Called(plugin, update)
+// SubmitOperationUpdate provides a mock function with given fields: update
+func (_m *Manager) SubmitOperationUpdate(update *core.OperationUpdate) {
+	_m.Called(update)
 }
 
 // WaitStop provides a mock function with given fields:
diff --git a/pkg/core/operation.go b/pkg/core/operation.go
index 0edf3054a..c90bb7d02 100644
--- a/pkg/core/operation.go
+++ b/pkg/core/operation.go
@@ -143,7 +143,7 @@ func ParseNamespacedOpID(ctx context.Context, nsIDStr string) (string, *fftypes.
 }
 
 type OperationCallbacks interface {
-	OperationUpdate(plugin Named, update *OperationUpdate)
+	OperationUpdate(update *OperationUpdate)
 }
 
 // OperationUpdate notifies FireFly of an update to an operation.
@@ -152,6 +152,7 @@ type OperationCallbacks interface {
 // Note this is an optional hook information, and stored separately to the confirmation of the actual event that was being submitted/sequenced.
 // Only the party submitting the transaction will see this data.
 type OperationUpdate struct {
+	Plugin         string
 	NamespacedOpID string
 	Status         OpStatus
 	BlockchainTXID string


Test patch:
diff --git a/internal/blockchain/common/common_test.go b/internal/blockchain/common/common_test.go
index 8b1c48573..38ae778d8 100644
--- a/internal/blockchain/common/common_test.go
+++ b/internal/blockchain/common/common_test.go
@@ -38,11 +38,13 @@ func TestCallbackOperationUpdate(t *testing.T) {
 	cb := NewBlockchainCallbacks()
 	cb.SetOperationalHandler("ns1", mcb)
 
-	mcb.On("OperationUpdate", mbi, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+	mbi.On("Name").Return("utblockchain")
+	mcb.On("OperationUpdate", mock.MatchedBy(func(update *core.OperationUpdate) bool {
 		return update.NamespacedOpID == nsOpID &&
 			update.Status == core.OpStatusSucceeded &&
 			update.BlockchainTXID == "tx1" &&
-			update.ErrorMessage == "err"
+			update.ErrorMessage == "err" &&
+			update.Plugin == "utblockchain"
 	})).Return().Once()
 	cb.OperationUpdate(context.Background(), mbi, nsOpID, core.OpStatusSucceeded, "tx1", "err", fftypes.JSONObject{})
 
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index cf31eb09e..6aeba046f 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -1336,10 +1336,11 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 		"transactionIndex": "0"
   }`)
 
-	em.On("OperationUpdate", e, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+	em.On("OperationUpdate", mock.MatchedBy(func(update *core.OperationUpdate) bool {
 		return update.NamespacedOpID == "ns1:"+operationID.String() &&
 			update.Status == core.OpStatusSucceeded &&
-			update.BlockchainTXID == "0x71a38acb7a5d4a970854f6d638ceb1fa10a4b59cbf4ed7674273a1a8dc8b36b8"
+			update.BlockchainTXID == "0x71a38acb7a5d4a970854f6d638ceb1fa10a4b59cbf4ed7674273a1a8dc8b36b8" &&
+			update.Plugin == "ethereum"
 	})).Return(nil)
 
 	err := json.Unmarshal(data.Bytes(), &reply)
@@ -1420,10 +1421,11 @@ func TestHandleReceiptTXUpdateEVMConnect(t *testing.T) {
 		"updated": "2022-08-03T18:55:43.781941Z"
 	}`)
 
-	em.On("OperationUpdate", e, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+	em.On("OperationUpdate", mock.MatchedBy(func(update *core.OperationUpdate) bool {
 		return update.NamespacedOpID == "ns1:"+operationID.String() &&
 			update.Status == core.OpStatusPending &&
-			update.BlockchainTXID == "0x929c898a46762d91e9f4b0b8e2800863dcf4a40f694109dc4cd19dbd334fa4cc"
+			update.BlockchainTXID == "0x929c898a46762d91e9f4b0b8e2800863dcf4a40f694109dc4cd19dbd334fa4cc" &&
+			update.Plugin == "ethereum"
 	})).Return(nil)
 
 	err := json.Unmarshal(data.Bytes(), &reply)
@@ -1460,10 +1462,11 @@ func TestHandleBadPayloadsAndThenReceiptFailure(t *testing.T) {
 
 	em := &coremocks.OperationCallbacks{}
 	e.SetOperationHandler("ns1", em)
-	txsu := em.On("OperationUpdate", e, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+	txsu := em.On("OperationUpdate", mock.MatchedBy(func(update *core.OperationUpdate) bool {
 		return update.NamespacedOpID == "ns1:"+operationID.String() &&
 			update.Status == core.OpStatusFailed &&
-			update.ErrorMessage == "Packing arguments for method 'broadcastBatch': abi: cannot use [3]uint8 as type [32]uint8 as argument"
+			update.ErrorMessage == "Packing arguments for method 'broadcastBatch': abi: cannot use [3]uint8 as type [32]uint8 as argument" &&
+			update.Plugin == "ethereum"
 	})).Return(fmt.Errorf("Shutdown"))
 	done := make(chan struct{})
 	txsu.RunFn = func(a mock.Arguments) {
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index 8e2f373de..1e88fda77 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -1255,10 +1255,11 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 		"receivedAt": 1630033474675
   }`)
 
-	em.On("OperationUpdate", e, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+	em.On("OperationUpdate", mock.MatchedBy(func(update *core.OperationUpdate) bool {
 		return update.NamespacedOpID == "ns1:"+operationID.String() &&
 			update.Status == core.OpStatusSucceeded &&
-			update.BlockchainTXID == "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2"
+			update.BlockchainTXID == "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2" &&
+			update.Plugin == "fabric"
 	})).Return(nil)
 
 	err := json.Unmarshal(data, &reply)
@@ -1315,10 +1316,11 @@ func TestHandleReceiptFailedTx(t *testing.T) {
 		"transactionId": "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2"
   }`)
 
-	em.On("OperationUpdate", e, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+	em.On("OperationUpdate", mock.MatchedBy(func(update *core.OperationUpdate) bool {
 		return update.NamespacedOpID == "ns1:"+operationID.String() &&
 			update.Status == core.OpStatusFailed &&
-			update.BlockchainTXID == "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2"
+			update.BlockchainTXID == "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2" &&
+			update.Plugin == "fabric"
 	})).Return(nil)
 
 	err := json.Unmarshal(data, &reply)
diff --git a/internal/dataexchange/ffdx/ffdx_test.go b/internal/dataexchange/ffdx/ffdx_test.go
index 86542474f..bc59ad54b 100644
--- a/internal/dataexchange/ffdx/ffdx_test.go
+++ b/internal/dataexchange/ffdx/ffdx_test.go
@@ -125,7 +125,7 @@ func TestInitMissingURL(t *testing.T) {
 
 func opAcker() func(args mock.Arguments) {
 	return func(args mock.Arguments) {
-		args[1].(*core.OperationUpdate).OnComplete()
+		args[0].(*core.OperationUpdate).OnComplete()
 	}
 }
 
@@ -421,30 +421,33 @@ func TestMessageEvents(t *testing.T) {
 	assert.NoError(t, err)
 
 	namespacedID1 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	ocb.On("OperationUpdate", h, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+	ocb.On("OperationUpdate", mock.MatchedBy(func(ev *core.OperationUpdate) bool {
 		return ev.NamespacedOpID == namespacedID1 &&
 			ev.Status == core.OpStatusFailed &&
-			ev.ErrorMessage == "pop"
+			ev.ErrorMessage == "pop" &&
+			ev.Plugin == "ffdx"
 	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"1","type":"message-failed","requestID":"` + namespacedID1 + `","error":"pop"}`
 	msg := <-toServer
 	assert.Equal(t, `{"action":"ack","id":"1"}`, string(msg))
 
 	namespacedID2 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+	ocb.On("OperationUpdate", mock.MatchedBy(func(ev *core.OperationUpdate) bool {
 		return ev.NamespacedOpID == namespacedID2 &&
-			ev.Status == core.OpStatusSucceeded
+			ev.Status == core.OpStatusSucceeded &&
+			ev.Plugin == "ffdx"
 	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"2","type":"message-delivered","requestID":"` + namespacedID2 + `"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"2"}`, string(msg))
 
 	namespacedID3 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+	ocb.On("OperationUpdate", mock.MatchedBy(func(ev *core.OperationUpdate) bool {
 		return ev.NamespacedOpID == namespacedID3 &&
 			ev.Status == core.OpStatusSucceeded &&
 			ev.DXManifest == `{"manifest":true}` &&
-			ev.Output.String() == `{"signatures":"and stuff"}`
+			ev.Output.String() == `{"signatures":"and stuff"}` &&
+			ev.Plugin == "ffdx"
 	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"3","type":"message-acknowledged","requestID":"` + namespacedID3 + `","info":{"signatures":"and stuff"},"manifest":"{\"manifest\":true}"}`
 	msg = <-toServer
@@ -478,20 +481,22 @@ func TestBlobEvents(t *testing.T) {
 	assert.NoError(t, err)
 
 	namespacedID5 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+	ocb.On("OperationUpdate", mock.MatchedBy(func(ev *core.OperationUpdate) bool {
 		return ev.NamespacedOpID == namespacedID5 &&
 			ev.Status == core.OpStatusFailed &&
-			ev.ErrorMessage == "pop"
+			ev.ErrorMessage == "pop" &&
+			ev.Plugin == "ffdx"
 	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"5","type":"blob-failed","requestID":"` + namespacedID5 + `","error":"pop"}`
 	msg := <-toServer
 	assert.Equal(t, `{"action":"ack","id":"5"}`, string(msg))
 
 	namespacedID6 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+	ocb.On("OperationUpdate", mock.MatchedBy(func(ev *core.OperationUpdate) bool {
 		return ev.NamespacedOpID == namespacedID6 &&
 			ev.Status == core.OpStatusSucceeded &&
-			ev.Output.String() == `{"some":"details"}`
+			ev.Output.String() == `{"some":"details"}` &&
+			ev.Plugin == "ffdx"
 	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"6","type":"blob-delivered","requestID":"` + namespacedID6 + `","info":{"some":"details"}}`
 	msg = <-toServer
@@ -509,10 +514,11 @@ func TestBlobEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"9"}`, string(msg))
 
 	namespacedID10 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+	ocb.On("OperationUpdate", mock.MatchedBy(func(ev *core.OperationUpdate) bool {
 		return ev.NamespacedOpID == namespacedID10 &&
 			ev.Status == core.OpStatusSucceeded &&
-			ev.Output.String() == `{"signatures":"and stuff"}`
+			ev.Output.String() == `{"signatures":"and stuff"}` &&
+			ev.Plugin == "ffdx"
 	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"10","type":"blob-acknowledged","requestID":"` + namespacedID10 + `","info":{"signatures":"and stuff"},"manifest":"{\"manifest\":true}"}`
 	msg = <-toServer
@@ -541,18 +547,20 @@ func TestEventsWithManifest(t *testing.T) {
 	h.SetOperationHandler("ns1", ocb)
 
 	namespacedID1 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+	ocb.On("OperationUpdate", mock.MatchedBy(func(ev *core.OperationUpdate) bool {
 		return ev.NamespacedOpID == namespacedID1 &&
-			ev.Status == core.OpStatusPending
+			ev.Status == core.OpStatusPending &&
+			ev.Plugin == "ffdx"
 	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"1","type":"message-delivered","requestID":"` + namespacedID1 + `"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"1"}`, string(msg))
 
 	namespacedID2 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+	ocb.On("OperationUpdate", mock.MatchedBy(func(ev *core.OperationUpdate) bool {
 		return ev.NamespacedOpID == namespacedID2 &&
-			ev.Status == core.OpStatusPending
+			ev.Status == core.OpStatusPending &&
+			ev.Plugin == "ffdx"
 	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"2","type":"blob-delivered","requestID":"` + namespacedID2 + `"}`
 	msg = <-toServer
diff --git a/internal/operations/operation_updater_test.go b/internal/operations/operation_updater_test.go
index 3b9549057..8bcd9af21 100644
--- a/internal/operations/operation_updater_test.go
+++ b/internal/operations/operation_updater_test.go
@@ -34,12 +34,6 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
-type mockplug struct{}
-
-func (mp *mockplug) Name() string {
-	return "unittest"
-}
-
 func newTestOperationUpdater(t *testing.T) *operationUpdater {
 	return newTestOperationUpdaterCommon(t, &database.Capabilities{Concurrency: true})
 }
@@ -177,18 +171,18 @@ func TestSubmitUpdateWorkerE2ESuccess(t *testing.T) {
 
 	om.Start()
 
-	om.SubmitOperationUpdate(&mockplug{}, &core.OperationUpdate{
+	om.SubmitOperationUpdate(&core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID1.String(),
 		Status:         core.OpStatusSucceeded,
 		BlockchainTXID: "tx12345",
 	})
-	om.SubmitOperationUpdate(&mockplug{}, &core.OperationUpdate{
+	om.SubmitOperationUpdate(&core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID2.String(),
 		Status:         core.OpStatusFailed,
 		ErrorMessage:   "err1",
 		Output:         fftypes.JSONObject{"test": true},
 	})
-	om.SubmitOperationUpdate(&mockplug{}, &core.OperationUpdate{
+	om.SubmitOperationUpdate(&core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID3.String(),
 		Status:         core.OpStatusFailed,
 		ErrorMessage:   "err2",
@@ -305,6 +299,25 @@ func TestDoUpdateIgnoreBadID(t *testing.T) {
 
 }
 
+func TestDoUpdateFailWrongPlugin(t *testing.T) {
+	ou := newTestOperationUpdaterNoConcurrency(t)
+	defer ou.close()
+
+	opID1 := fftypes.NewUUID()
+	txID1 := fftypes.NewUUID()
+
+	ou.initQueues()
+
+	err := ou.doUpdate(ou.ctx, &core.OperationUpdate{
+		NamespacedOpID: "ns1:" + opID1.String(), Status: core.OpStatusSucceeded, BlockchainTXID: "0x12345", Plugin: "plugin2",
+	}, []*core.Operation{
+		{Namespace: "ns1", ID: opID1, Type: core.OpTypeBlockchainInvoke, Transaction: txID1, Plugin: "plugin1"},
+	}, []*core.Transaction{
+		{ID: txID1},
+	})
+	assert.NoError(t, err)
+}
+
 func TestDoUpdateFailTransactionUpdate(t *testing.T) {
 	ou := newTestOperationUpdaterNoConcurrency(t)
 	defer ou.close()
diff --git a/internal/orchestrator/bound_callbacks_test.go b/internal/orchestrator/bound_callbacks_test.go
index 2387a893a..662b4368a 100644
--- a/internal/orchestrator/bound_callbacks_test.go
+++ b/internal/orchestrator/bound_callbacks_test.go
@@ -21,18 +21,15 @@ import (
 	"testing"
 
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
-	"github.com/hyperledger/firefly/mocks/blockchainmocks"
 	"github.com/hyperledger/firefly/mocks/eventmocks"
 	"github.com/hyperledger/firefly/mocks/operationmocks"
 	"github.com/hyperledger/firefly/mocks/sharedstoragemocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
 )
 
 func TestBoundCallbacks(t *testing.T) {
 	mei := &eventmocks.EventManager{}
-	mbi := &blockchainmocks.Plugin{}
 	mss := &sharedstoragemocks.Plugin{}
 	mom := &operationmocks.Manager{}
 	bc := boundCallbacks{ei: mei, ss: mss, om: mom}
@@ -49,8 +46,8 @@ func TestBoundCallbacks(t *testing.T) {
 		ErrorMessage:   "error info",
 		Output:         info,
 	}
-	mom.On("SubmitOperationUpdate", mock.Anything, update).Return().Once()
-	bc.OperationUpdate(mbi, update)
+	mom.On("SubmitOperationUpdate", update).Return().Once()
+	bc.OperationUpdate(update)
 
 	mei.On("SharedStorageBatchDownloaded", mss, "payload1", []byte(`{}`)).Return(nil, fmt.Errorf("pop"))
 	_, err := bc.SharedStorageBatchDownloaded("payload1", []byte(`{}`))
@@ -60,7 +57,6 @@ func TestBoundCallbacks(t *testing.T) {
 	bc.SharedStorageBlobDownloaded(*hash, 12345, "payload1")
 
 	mei.AssertExpectations(t)
-	mbi.AssertExpectations(t)
 	mss.AssertExpectations(t)
 	mom.AssertExpectations(t)
 }
diff --git a/internal/tokens/fftokens/fftokens_test.go b/internal/tokens/fftokens/fftokens_test.go
index bb4891f9f..0645cd010 100644
--- a/internal/tokens/fftokens/fftokens_test.go
+++ b/internal/tokens/fftokens/fftokens_test.go
@@ -828,8 +828,10 @@ func TestReceiptEvents(t *testing.T) {
 	opID := fftypes.NewUUID()
 
 	// receipt: bad ID - passed through
-	mcb.On("OperationUpdate", h, mock.MatchedBy(func(update *core.OperationUpdate) bool {
-		return update.NamespacedOpID == "wrong " && update.Status == core.OpStatusFailed
+	mcb.On("OperationUpdate", mock.MatchedBy(func(update *core.OperationUpdate) bool {
+		return update.NamespacedOpID == "wrong " &&
+			update.Status == core.OpStatusFailed &&
+			update.Plugin == "fftokens"
 	})).Return(nil).Once()
 	fromServer <- fftypes.JSONObject{
 		"id":    "3",
@@ -838,10 +840,11 @@ func TestReceiptEvents(t *testing.T) {
 	}.String()
 
 	// receipt: success
-	mcb.On("OperationUpdate", h, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+	mcb.On("OperationUpdate", mock.MatchedBy(func(update *core.OperationUpdate) bool {
 		return update.NamespacedOpID == "ns1:"+opID.String() &&
 			update.Status == core.OpStatusSucceeded &&
-			update.BlockchainTXID == "0xffffeeee"
+			update.BlockchainTXID == "0xffffeeee" &&
+			update.Plugin == "fftokens"
 	})).Return(nil).Once()
 	fromServer <- fftypes.JSONObject{
 		"id":    "4",
@@ -854,10 +857,11 @@ func TestReceiptEvents(t *testing.T) {
 	}.String()
 
 	// receipt: failure
-	mcb.On("OperationUpdate", h, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+	mcb.On("OperationUpdate", mock.MatchedBy(func(update *core.OperationUpdate) bool {
 		return update.NamespacedOpID == "ns1:"+opID.String() &&
 			update.Status == core.OpStatusFailed &&
-			update.BlockchainTXID == "0xffffeeee"
+			update.BlockchainTXID == "0xffffeeee" &&
+			update.Plugin == "fftokens"
 	})).Return(nil).Once()
 	fromServer <- fftypes.JSONObject{
 		"id":    "5",