Repository: hyperledger/firefly
Pull request #915 (2022-07-30 17:35:10)

Allow multiple copies of things on different namespaces
Everything that can be sent (messages, data, all definition types) needs to only be unique within a namespace. The IDs can no longer be considered globally unique, and the database needs to reflect that.

Move "node" configuration to the "multiparty" section (next to "org"), as different namespaces may want to advertise different names for their local node.

Clean up operation handling in DX to make it consistent with other plugins.


Patch:
diff --git a/cmd/firefly.go b/cmd/firefly.go
index 054595c84..54f111b15 100644
--- a/cmd/firefly.go
+++ b/cmd/firefly.go
@@ -101,7 +101,6 @@ func run() error {
 	// Setup logging after reading config (even if failed), to output header correctly
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	ctx = log.WithLogger(ctx, logrus.WithField("pid", fmt.Sprintf("%d", os.Getpid())))
-	ctx = log.WithLogger(ctx, logrus.WithField("prefix", config.GetString(coreconfig.NodeName)))
 
 	config.SetupLogging(ctx)
 	log.L(ctx).Infof("Hyperledger FireFly")
diff --git a/db/migrations/postgres/000096_add_namespace_to_message_indexes.down.sql b/db/migrations/postgres/000096_add_namespace_to_message_indexes.down.sql
new file mode 100644
index 000000000..853061a7b
--- /dev/null
+++ b/db/migrations/postgres/000096_add_namespace_to_message_indexes.down.sql
@@ -0,0 +1,37 @@
+BEGIN;
+DROP INDEX identities_id;
+CREATE UNIQUE INDEX identities_id ON identities(id);
+
+DROP INDEX verifiers_hash;
+CREATE UNIQUE INDEX verifiers_hash on verifiers(hash);
+
+DROP INDEX verifiers_identity;
+CREATE UNIQUE INDEX verifiers_identity on verifiers(identity);
+
+DROP INDEX tokenpool_locator;
+CREATE UNIQUE INDEX tokenpool_locator ON tokenpool(connector, locator);
+
+DROP INDEX ffi_id;
+CREATE UNIQUE INDEX ffi_id ON ffi(id);
+
+DROP INDEX messages_id;
+CREATE UNIQUE INDEX messages_id ON messages(id);
+
+DROP INDEX data_id;
+CREATE UNIQUE INDEX data_id ON data(id);
+
+ALTER TABLE messages_data RENAME TO messages_data_old;
+CREATE TABLE messages_data (
+  seq         SERIAL          PRIMARY KEY,
+  message_id  UUID            NOT NULL,
+  data_id     UUID            NOT NULL,
+  data_hash   CHAR(64)        NOT NULL,
+  data_idx    INT             NOT NULL
+);
+INSERT INTO messages_data(message_id, data_id, data_hash, data_idx)
+  SELECT message_id, data_id, data_hash, data_idx FROM messages_data_old;
+DROP TABLE messages_data_old;
+
+CREATE INDEX messages_data_message ON messages_data(message_id);
+CREATE INDEX messages_data_data ON messages_data(data_id);
+COMMIT;
diff --git a/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql b/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql
new file mode 100644
index 000000000..84b5d7e8c
--- /dev/null
+++ b/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql
@@ -0,0 +1,41 @@
+BEGIN;
+DROP INDEX identities_id;
+CREATE UNIQUE INDEX identities_id ON identities(namespace, id);
+
+DROP INDEX verifiers_hash;
+CREATE UNIQUE INDEX verifiers_hash on verifiers(namespace, hash);
+
+DROP INDEX verifiers_identity;
+CREATE UNIQUE INDEX verifiers_identity on verifiers(namespace, identity);
+
+DROP INDEX tokenpool_locator;
+CREATE UNIQUE INDEX tokenpool_locator ON tokenpool(namespace, connector, locator);
+
+DROP INDEX ffi_id;
+CREATE UNIQUE INDEX ffi_id ON ffi(namespace, id);
+
+DROP INDEX messages_id;
+CREATE UNIQUE INDEX messages_id ON messages(namespace_local, id);
+
+DROP INDEX data_id;
+CREATE UNIQUE INDEX data_id ON data(namespace, id);
+
+ALTER TABLE messages_data RENAME TO messages_data_old;
+CREATE TABLE messages_data (
+  seq         SERIAL          PRIMARY KEY,
+  message_id  UUID            NOT NULL,
+  data_id     UUID            NOT NULL,
+  data_hash   CHAR(64)        NOT NULL,
+  data_idx    INT             NOT NULL,
+  namespace   VARCHAR(64)
+);
+INSERT INTO messages_data(message_id, data_id, data_hash, data_idx)
+  SELECT message_id, data_id, data_hash, data_idx FROM messages_data_old;
+DROP TABLE messages_data_old;
+UPDATE messages_data SET namespace = msg.namespace
+  FROM (SELECT namespace, id FROM messages) AS msg
+  WHERE messages_data.message_id = msg.id;
+
+CREATE INDEX messages_data_message ON messages_data(namespace, message_id);
+CREATE INDEX messages_data_data ON messages_data(namespace, data_id);
+COMMIT;
diff --git a/db/migrations/sqlite/000096_add_namespace_to_message_indexes.down.sql b/db/migrations/sqlite/000096_add_namespace_to_message_indexes.down.sql
new file mode 100644
index 000000000..6979f6f1a
--- /dev/null
+++ b/db/migrations/sqlite/000096_add_namespace_to_message_indexes.down.sql
@@ -0,0 +1,35 @@
+DROP INDEX identities_id;
+CREATE UNIQUE INDEX identities_id ON identities(id);
+
+DROP INDEX verifiers_hash;
+CREATE UNIQUE INDEX verifiers_hash on verifiers(hash);
+
+DROP INDEX verifiers_identity;
+CREATE UNIQUE INDEX verifiers_identity on verifiers(identity);
+
+DROP INDEX tokenpool_locator;
+CREATE UNIQUE INDEX tokenpool_locator ON tokenpool(connector, locator);
+
+DROP INDEX ffi_id;
+CREATE UNIQUE INDEX ffi_id ON ffi(id);
+
+DROP INDEX messages_id;
+CREATE UNIQUE INDEX messages_id ON messages(id);
+
+DROP INDEX data_id;
+CREATE UNIQUE INDEX data_id ON data(id);
+
+ALTER TABLE messages_data RENAME TO messages_data_old;
+CREATE TABLE messages_data (
+  seq         SERIAL          PRIMARY KEY,
+  message_id  UUID            NOT NULL,
+  data_id     UUID            NOT NULL,
+  data_hash   CHAR(64)        NOT NULL,
+  data_idx    INT             NOT NULL
+);
+INSERT INTO messages_data(message_id, data_id, data_hash, data_idx)
+  SELECT message_id, data_id, data_hash, data_idx FROM messages_data_old;
+DROP TABLE messages_data_old;
+
+CREATE INDEX messages_data_message ON messages_data(message_id);
+CREATE INDEX messages_data_data ON messages_data(data_id);
diff --git a/db/migrations/sqlite/000096_add_namespace_to_message_indexes.up.sql b/db/migrations/sqlite/000096_add_namespace_to_message_indexes.up.sql
new file mode 100644
index 000000000..aff5238e0
--- /dev/null
+++ b/db/migrations/sqlite/000096_add_namespace_to_message_indexes.up.sql
@@ -0,0 +1,39 @@
+DROP INDEX identities_id;
+CREATE UNIQUE INDEX identities_id ON identities(namespace, id);
+
+DROP INDEX verifiers_hash;
+CREATE UNIQUE INDEX verifiers_hash on verifiers(namespace, hash);
+
+DROP INDEX verifiers_identity;
+CREATE UNIQUE INDEX verifiers_identity on verifiers(namespace, identity);
+
+DROP INDEX tokenpool_locator;
+CREATE UNIQUE INDEX tokenpool_locator ON tokenpool(namespace, connector, locator);
+
+DROP INDEX ffi_id;
+CREATE UNIQUE INDEX ffi_id ON ffi(namespace, id);
+
+DROP INDEX messages_id;
+CREATE UNIQUE INDEX messages_id ON messages(namespace_local, id);
+
+DROP INDEX data_id;
+CREATE UNIQUE INDEX data_id ON data(namespace, id);
+
+ALTER TABLE messages_data RENAME TO messages_data_old;
+CREATE TABLE messages_data (
+  seq         INTEGER         PRIMARY KEY AUTOINCREMENT,
+  message_id  UUID            NOT NULL,
+  data_id     UUID            NOT NULL,
+  data_hash   CHAR(64)        NOT NULL,
+  data_idx    INT             NOT NULL,
+  namespace   VARCHAR(64)
+);
+INSERT INTO messages_data(message_id, data_id, data_hash, data_idx)
+  SELECT message_id, data_id, data_hash, data_idx FROM messages_data_old;
+DROP TABLE messages_data_old;
+UPDATE messages_data SET namespace = msg.namespace
+  FROM (SELECT namespace, id FROM messages) AS msg
+  WHERE messages_data.message_id = msg.id;
+
+CREATE INDEX messages_data_message ON messages_data(namespace, message_id);
+CREATE INDEX messages_data_data ON messages_data(namespace, data_id);
diff --git a/docs/reference/config.md b/docs/reference/config.md
index b62a378a1..e1009783d 100644
--- a/docs/reference/config.md
+++ b/docs/reference/config.md
@@ -635,6 +635,13 @@ nav_order: 2
 |firstEvent|The first event the contract should process. Valid options are `oldest` or `newest`|`string`|`<nil>`
 |location|A blockchain-specific contract location. For example, an Ethereum contract address, or a Fabric chaincode name and channe|`string`|`<nil>`
 
+## namespaces.predefined[].multiparty.node
+
+|Key|Description|Type|Default Value|
+|---|-----------|----|-------------|
+|description|A description for the node in this namespace|`string`|`<nil>`
+|name|The node name for this namespace|`string`|`<nil>`
+
 ## namespaces.predefined[].multiparty.org
 
 |Key|Description|Type|Default Value|
diff --git a/internal/assets/operations.go b/internal/assets/operations.go
index 92bb2ec34..e3389ff7c 100644
--- a/internal/assets/operations.go
+++ b/internal/assets/operations.go
@@ -24,7 +24,6 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/coremsgs"
-	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/core"
 )
@@ -146,7 +145,7 @@ func (am *assetManager) RunOperation(ctx context.Context, op *core.PreparedOpera
 	}
 }
 
-func (am *assetManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *operations.OperationUpdate) error {
+func (am *assetManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *core.OperationUpdate) error {
 	// Write an event for failed pool operations
 	if op.Type == core.OpTypeTokenCreatePool && update.Status == core.OpStatusFailed {
 		tokenPool, err := txcommon.RetrieveTokenPoolCreateInputs(ctx, op)
diff --git a/internal/blockchain/common/common.go b/internal/blockchain/common/common.go
index 2de04f288..d5003146b 100644
--- a/internal/blockchain/common/common.go
+++ b/internal/blockchain/common/common.go
@@ -94,7 +94,13 @@ 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, nsOpID, status, blockchainTXID, errorMessage, opOutput)
+		handler.OperationUpdate(plugin, &core.OperationUpdate{
+			NamespacedOpID: nsOpID,
+			Status:         status,
+			BlockchainTXID: blockchainTXID,
+			ErrorMessage:   errorMessage,
+			Output:         opOutput,
+		})
 		return
 	}
 	log.L(ctx).Errorf("No handler found for blockchain operation '%s'", nsOpID)
diff --git a/internal/broadcast/operations.go b/internal/broadcast/operations.go
index f4b0dc04e..f79cdb6f0 100644
--- a/internal/broadcast/operations.go
+++ b/internal/broadcast/operations.go
@@ -25,7 +25,6 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/coremsgs"
-	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 )
@@ -169,7 +168,7 @@ func (bm *broadcastManager) uploadBlob(ctx context.Context, data uploadBlobData)
 	return getUploadBlobOutputs(data.Data.Blob.Public), true, nil
 }
 
-func (bm *broadcastManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *operations.OperationUpdate) error {
+func (bm *broadcastManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *core.OperationUpdate) error {
 	return nil
 }
 
diff --git a/internal/contracts/operations.go b/internal/contracts/operations.go
index b68700bc6..c378180ff 100644
--- a/internal/contracts/operations.go
+++ b/internal/contracts/operations.go
@@ -23,7 +23,6 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly/internal/coremsgs"
-	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/pkg/core"
 )
 
@@ -73,7 +72,7 @@ func (cm *contractManager) RunOperation(ctx context.Context, op *core.PreparedOp
 	}
 }
 
-func (cm *contractManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *operations.OperationUpdate) error {
+func (cm *contractManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *core.OperationUpdate) error {
 	// Special handling for OpTypeBlockchainInvoke, which writes an event when it succeeds or fails
 	if op.Type == core.OpTypeBlockchainInvoke {
 		if update.Status == core.OpStatusSucceeded {
diff --git a/internal/coreconfig/coreconfig.go b/internal/coreconfig/coreconfig.go
index a9109d371..6e04aa28d 100644
--- a/internal/coreconfig/coreconfig.go
+++ b/internal/coreconfig/coreconfig.go
@@ -51,6 +51,10 @@ const (
 	NamespaceMultipartyOrgDescription = "org.description"
 	// NamespaceMultipartyOrgKey is the signing key allocated to the local root org within a namespace
 	NamespaceMultipartyOrgKey = "org.key"
+	// NamespaceMultipartyNodeName is the name for the local node within a namespace
+	NamespaceMultipartyNodeName = "node.name"
+	// NamespaceMultipartyNodeName is a description for the local node within a namespace
+	NamespaceMultipartyNodeDescription = "node.description"
 	// NamespaceMultipartyContract is a list of firefly contract configurations for this namespace
 	NamespaceMultipartyContract = "contract"
 	// NamespaceMultipartyContractFirstEvent is the first event to process for this contract
diff --git a/internal/coremsgs/en_config_descriptions.go b/internal/coremsgs/en_config_descriptions.go
index fbdd452a5..cd26b5283 100644
--- a/internal/coremsgs/en_config_descriptions.go
+++ b/internal/coremsgs/en_config_descriptions.go
@@ -261,6 +261,8 @@ var (
 	ConfigNamespacesMultipartyOrgName            = ffc("config.namespaces.predefined[].multiparty.org.name", "A short name for the local root organization within this namespace", i18n.StringType)
 	ConfigNamespacesMultipartyOrgDesc            = ffc("config.namespaces.predefined[].multiparty.org.description", "A description for the local root organization within this namespace", i18n.StringType)
 	ConfigNamespacesMultipartyOrgKey             = ffc("config.namespaces.predefined[].multiparty.org.key", "The signing key allocated to the root organization within this namespace", i18n.StringType)
+	ConfigNamespacesMultipartyNodeName           = ffc("config.namespaces.predefined[].multiparty.node.name", "The node name for this namespace", i18n.StringType)
+	ConfigNamespacesMultipartyNodeDescription    = ffc("config.namespaces.predefined[].multiparty.node.description", "A description for the node in this namespace", i18n.StringType)
 	ConfigNamespacesMultipartyContract           = ffc("config.namespaces.predefined[].contract", "A list containing configuration for the multi-party blockchain contract", i18n.StringType)
 	ConfigNamespacesMultipartyContractFirstEvent = ffc("config.namespaces.predefined[].multiparty.contract[].firstEvent", "The first event the contract should process. Valid options are `oldest` or `newest`", i18n.StringType)
 	ConfigNamespacesMultipartyContractLocation   = ffc("config.namespaces.predefined[].multiparty.contract[].location", "A blockchain-specific contract location. For example, an Ethereum contract address, or a Fabric chaincode name and channe", i18n.StringType)
diff --git a/internal/database/sqlcommon/identity_sql.go b/internal/database/sqlcommon/identity_sql.go
index d5563c7f0..e2311a91c 100644
--- a/internal/database/sqlcommon/identity_sql.go
+++ b/internal/database/sqlcommon/identity_sql.go
@@ -72,7 +72,8 @@ func (s *SQLCommon) attemptIdentityUpdate(ctx context.Context, tx *txWrapper, id
 			Set("messages_update", identity.Messages.Update).
 			Set("updated", identity.Updated).
 			Where(sq.Eq{
-				"id": identity.ID,
+				"id":        identity.ID,
+				"namespace": identity.Namespace,
 			}),
 		func() {
 			s.callbacks.UUIDCollectionNSEvent(database.CollectionIdentities, core.ChangeEventTypeUpdated, identity.Namespace, identity.ID)
@@ -127,7 +128,7 @@ func (s *SQLCommon) UpsertIdentity(ctx context.Context, identity *core.Identity,
 		msgRows, _, err := s.queryTx(ctx, identitiesTable, tx,
 			sq.Select("id").
 				From(identitiesTable).
-				Where(sq.Eq{"id": identity.ID}),
+				Where(sq.Eq{"id": identity.ID, "namespace": identity.Namespace}),
 		)
 		if err != nil {
 			return err
diff --git a/internal/dataexchange/ffdx/dxevent.go b/internal/dataexchange/ffdx/dxevent.go
index e23a1ab65..f715fb4b8 100644
--- a/internal/dataexchange/ffdx/dxevent.go
+++ b/internal/dataexchange/ffdx/dxevent.go
@@ -46,21 +46,15 @@ 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) EventID() string {
 	return e.id
 }
 
-func (e *dxEvent) NamespacedID() string {
-	return e.requestID
-}
-
 func (e *dxEvent) Type() dataexchange.DXEventType {
 	return e.dxType
 }
@@ -88,48 +82,42 @@ func (e *dxEvent) PrivateBlobReceived() *dataexchange.PrivateBlobReceived {
 	return e.privateBlobReceived
 }
 
-func (e *dxEvent) TransferResult() *dataexchange.TransferResult {
-	return e.transferResult
-}
-
 func (h *FFDX) dispatchEvent(msg *wsEvent) {
 	var namespace string
 	var err error
-	e := &dxEvent{ffdx: h, id: msg.EventID, requestID: msg.RequestID}
+	e := &dxEvent{ffdx: h, id: msg.EventID}
 	switch msg.Type {
 	case messageFailed:
-		e.dxType = dataexchange.DXEventTypeTransferResult
-		e.transferResult = &dataexchange.TransferResult{
-			TrackingID: msg.RequestID,
-			Status:     core.OpStatusFailed,
-			TransportStatusUpdate: core.TransportStatusUpdate{
-				Error: msg.Error,
-				Info:  msg.Info,
-			},
-		}
+		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			NamespacedOpID: msg.RequestID,
+			Status:         core.OpStatusFailed,
+			ErrorMessage:   msg.Error,
+			Output:         msg.Info,
+			OnComplete:     e.Ack,
+		})
+		return
 	case messageDelivered:
 		status := core.OpStatusSucceeded
 		if h.capabilities.Manifest {
 			status = core.OpStatusPending
 		}
-		e.dxType = dataexchange.DXEventTypeTransferResult
-		e.transferResult = &dataexchange.TransferResult{
-			TrackingID: msg.RequestID,
-			Status:     status,
-			TransportStatusUpdate: core.TransportStatusUpdate{
-				Info: msg.Info,
-			},
-		}
+		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			NamespacedOpID: msg.RequestID,
+			Status:         status,
+			Output:         msg.Info,
+			OnComplete:     e.Ack,
+		})
+		return
 	case messageAcknowledged:
-		e.dxType = dataexchange.DXEventTypeTransferResult
-		e.transferResult = &dataexchange.TransferResult{
-			TrackingID: msg.RequestID,
-			Status:     core.OpStatusSucceeded,
-			TransportStatusUpdate: core.TransportStatusUpdate{
-				Manifest: msg.Manifest,
-				Info:     msg.Info,
-			},
-		}
+		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			NamespacedOpID: msg.RequestID,
+			Status:         core.OpStatusSucceeded,
+			VerifyManifest: h.capabilities.Manifest,
+			DXManifest:     msg.Manifest,
+			Output:         msg.Info,
+			OnComplete:     e.Ack,
+		})
+		return
 	case messageReceived:
 		// De-serialize the transport wrapper
 		var wrapper *core.TransportWrapper
@@ -148,28 +136,26 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 			}
 		}
 	case blobFailed:
-		e.dxType = dataexchange.DXEventTypeTransferResult
-		e.transferResult = &dataexchange.TransferResult{
-			TrackingID: msg.RequestID,
-			Status:     core.OpStatusFailed,
-			TransportStatusUpdate: core.TransportStatusUpdate{
-				Error: msg.Error,
-				Info:  msg.Info,
-			},
-		}
+		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			NamespacedOpID: msg.RequestID,
+			Status:         core.OpStatusFailed,
+			ErrorMessage:   msg.Error,
+			Output:         msg.Info,
+			OnComplete:     e.Ack,
+		})
+		return
 	case blobDelivered:
 		status := core.OpStatusSucceeded
 		if h.capabilities.Manifest {
 			status = core.OpStatusPending
 		}
-		e.dxType = dataexchange.DXEventTypeTransferResult
-		e.transferResult = &dataexchange.TransferResult{
-			TrackingID: msg.RequestID,
-			Status:     status,
-			TransportStatusUpdate: core.TransportStatusUpdate{
-				Info: msg.Info,
-			},
-		}
+		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			NamespacedOpID: msg.RequestID,
+			Status:         status,
+			Output:         msg.Info,
+			OnComplete:     e.Ack,
+		})
+		return
 	case blobReceived:
 		var hash *fftypes.Bytes32
 		hash, err = fftypes.ParseBytes32(h.ctx, msg.Hash)
@@ -185,27 +171,24 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 			}
 		}
 	case blobAcknowledged:
-		e.dxType = dataexchange.DXEventTypeTransferResult
-		e.transferResult = &dataexchange.TransferResult{
-			TrackingID: msg.RequestID,
-			Status:     core.OpStatusSucceeded,
-			TransportStatusUpdate: core.TransportStatusUpdate{
-				Hash: msg.Hash,
-				Info: msg.Info,
-			},
-		}
+		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			NamespacedOpID: msg.RequestID,
+			Status:         core.OpStatusSucceeded,
+			Output:         msg.Info,
+			VerifyManifest: h.capabilities.Manifest,
+			DXHash:         msg.Hash,
+			OnComplete:     e.Ack,
+		})
+		return
 	default:
 		err = i18n.NewError(h.ctx, coremsgs.MsgUnexpectedDXMessageType, msg.Type)
 	}
 
 	// If we couldn't dispatch the event we received, we still ack it
-	if err != nil {
+	if err == nil {
+		h.callbacks.DXEvent(h.ctx, namespace, e)
+	} else {
 		log.L(h.ctx).Warnf("Failed to dispatch DX event: %s", err)
 		e.Ack()
-	} else {
-		if namespace == "" && msg.RequestID != "" {
-			namespace, _, _ = core.ParseNamespacedOpID(h.ctx, msg.RequestID)
-		}
-		h.callbacks.DXEvent(h.ctx, namespace, e)
 	}
 }
diff --git a/internal/dataexchange/ffdx/ffdx.go b/internal/dataexchange/ffdx/ffdx.go
index e44acd4a6..d37be52e2 100644
--- a/internal/dataexchange/ffdx/ffdx.go
+++ b/internal/dataexchange/ffdx/ffdx.go
@@ -34,6 +34,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly-common/pkg/wsclient"
 	"github.com/hyperledger/firefly/internal/coremsgs"
+	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/dataexchange"
 )
 
@@ -51,14 +52,26 @@ type FFDX struct {
 }
 
 type callbacks struct {
-	handlers map[string]dataexchange.Callbacks
+	plugin     dataexchange.Plugin
+	handlers   map[string]dataexchange.Callbacks
+	opHandlers map[string]core.OperationCallbacks
+}
+
+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)
+	} else {
+		log.L(ctx).Errorf("No handler found for DX operation '%s'", update.NamespacedOpID)
+		update.OnComplete()
+	}
 }
 
 func (cb *callbacks) DXEvent(ctx context.Context, namespace string, event dataexchange.DXEvent) {
 	if handler, ok := cb.handlers[namespace]; ok {
-		handler.DXEvent(ctx, event)
+		handler.DXEvent(cb.plugin, event)
 	} else {
-		log.L(ctx).Errorf("unknown namespace on event '%s'", event.EventID())
+		log.L(ctx).Errorf("No handler found for DX event '%s'", event.EventID())
 		event.Ack()
 	}
 }
@@ -143,7 +156,11 @@ func (h *FFDX) Name() string {
 func (h *FFDX) Init(ctx context.Context, config config.Section) (err error) {
 	h.ctx = log.WithLogField(ctx, "dx", "https")
 	h.ackChannel = make(chan *ack)
-	h.callbacks.handlers = make(map[string]dataexchange.Callbacks)
+	h.callbacks = callbacks{
+		plugin:     h,
+		handlers:   make(map[string]dataexchange.Callbacks),
+		opHandlers: make(map[string]core.OperationCallbacks),
+	}
 	h.needsInit = config.GetBool(DataExchangeInitEnabled)
 
 	if config.GetString(ffresty.HTTPConfigURL) == "" {
@@ -174,6 +191,10 @@ func (h *FFDX) SetHandler(namespace string, handler dataexchange.Callbacks) {
 	h.callbacks.handlers[namespace] = handler
 }
 
+func (h *FFDX) SetOperationHandler(namespace string, handler core.OperationCallbacks) {
+	h.callbacks.opHandlers[namespace] = handler
+}
+
 func (h *FFDX) Start() error {
 	return h.wsconn.Connect()
 }
diff --git a/internal/events/dx_callbacks.go b/internal/events/dx_callbacks.go
index ca9accfe9..381fe7d97 100644
--- a/internal/events/dx_callbacks.go
+++ b/internal/events/dx_callbacks.go
@@ -183,7 +183,7 @@ func (em *eventManager) DXEvent(dx dataexchange.Plugin, event dataexchange.DXEve
 		// Batches are significant items of work in their own right, so get dispatched to their own routines
 		go em.messageReceived(dx, event)
 	default:
-		log.L(em.ctx).Errorf("Invalid DX event type from %s: %d", dx.Name(), event.Type())
+		log.L(em.ctx).Errorf("Invalid data exchange event type from %s: %d", dx.Name(), event.Type())
 		event.Ack() // still ack
 	}
 }
@@ -208,7 +208,7 @@ func (em *eventManager) privateBlobReceived(dx dataexchange.Plugin, event dataex
 	log.L(em.ctx).Infof("Blob received event from data exchange %s: Peer='%s' Hash='%v' PayloadRef='%s'", dx.Name(), br.PeerID, &br.Hash, br.PayloadRef)
 
 	if br.PeerID == "" || len(br.PeerID) > 256 || br.PayloadRef == "" || len(br.PayloadRef) > 1024 {
-		log.L(em.ctx).Errorf("Invalid blob received event from data exhange: Peer='%s' Hash='%v' PayloadRef='%s'", br.PeerID, &br.Hash, br.PayloadRef)
+		log.L(em.ctx).Errorf("Invalid blob received event from data exchange: Peer='%s' Hash='%v' PayloadRef='%s'", br.PeerID, &br.Hash, br.PayloadRef)
 		event.Ack() // Still confirm the event
 		return
 	}
diff --git a/internal/events/event_manager.go b/internal/events/event_manager.go
index d49c577cc..b22bd8f6d 100644
--- a/internal/events/event_manager.go
+++ b/internal/events/event_manager.go
@@ -70,7 +70,7 @@ type EventManager interface {
 	BlockchainNetworkAction(action string, location *fftypes.JSONAny, event *blockchain.Event, signingKey *core.VerifierRef) error
 
 	// Bound dataexchange callbacks
-	DXEvent(dx dataexchange.Plugin, event dataexchange.DXEvent)
+	DXEvent(plugin dataexchange.Plugin, event dataexchange.DXEvent)
 
 	// Bound sharedstorage callbacks
 	SharedStorageBatchDownloaded(ss sharedstorage.Plugin, payloadRef string, data []byte) (*fftypes.UUID, error)
diff --git a/internal/identity/identitymanager.go b/internal/identity/identitymanager.go
index 78d44b510..38dd73318 100644
--- a/internal/identity/identitymanager.go
+++ b/internal/identity/identitymanager.go
@@ -48,9 +48,9 @@ type Manager interface {
 	CachedIdentityLookupByID(ctx context.Context, id *fftypes.UUID) (identity *core.Identity, err error)
 	CachedIdentityLookupMustExist(ctx context.Context, did string) (identity *core.Identity, retryable bool, err error)
 	CachedIdentityLookupNilOK(ctx context.Context, did string) (identity *core.Identity, retryable bool, err error)
-	CachedVerifierLookup(ctx context.Context, vType core.VerifierType, value string) (verifier *core.Verifier, err error)
 	GetMultipartyRootVerifier(ctx context.Context) (*core.VerifierRef, error)
 	GetMultipartyRootOrg(ctx context.Context) (*core.Identity, error)
+	GetLocalNode(ctx context.Context) (node *core.Identity, err error)
 	VerifyIdentityChain(ctx context.Context, identity *core.Identity) (immediateParent *core.Identity, retryable bool, err error)
 }
 
@@ -61,6 +61,7 @@ type identityManager struct {
 	namespace              string
 	defaultKey             string
 	multipartyRootVerifier *core.VerifierRef
+	localNodeID            *fftypes.UUID
 	identityCacheTTL       time.Duration
 	identityCache          *ccache.Cache
 	signingKeyCacheTTL     time.Duration
@@ -100,6 +101,19 @@ func ParseKeyNormalizationConfig(strConfigVal string) int {
 	}
 }
 
+func (im *identityManager) GetLocalNode(ctx context.Context) (node *core.Identity, err error) {
+	if im.localNodeID != nil {
+		return im.CachedIdentityLookupByID(ctx, im.localNodeID)
+	}
+	nodeName := im.multiparty.LocalNode().Name
+	node, err = im.database.GetIdentityByName(ctx, core.IdentityTypeNode, im.namespace, nodeName)
+	if err == nil && node != nil {
+		im.localNodeID = node.ID
+		im.addToIdentityCache(node)
+	}
+	return node, err
+}
+
 // NormalizeSigningKey takes in only a "key" (which may be empty to use the default) to be normalized and returned.
 // This is for cases where keys are used directly without an "author" field alongside them (custom contracts, tokens),
 // or when the author is known by the caller and should not / cannot be confirmed prior to sending (identity claims)
@@ -479,6 +493,11 @@ func (im *identityManager) CachedIdentityLookupMustExist(ctx context.Context, di
 	return identity, false, nil
 }
 
+func (im *identityManager) addToIdentityCache(identity *core.Identity) {
+	cacheKey := fmt.Sprintf("id=%s", identity.ID)
+	im.identityCache.Set(cacheKey, identity, im.identityCacheTTL)
+}
+
 func (im *identityManager) CachedIdentityLookupByID(ctx context.Context, id *fftypes.UUID) (identity *core.Identity, err error) {
 	// Use an LRU cache for the author identity, as it's likely for the same identity to be re-used over and over
 	cacheKey := fmt.Sprintf("id=%s", id)
@@ -491,24 +510,7 @@ func (im *identityManager) CachedIdentityLookupByID(ctx context.Context, id *fft
 			return identity, err
 		}
 		// Cache the result
-		im.identityCache.Set(cacheKey, identity, im.identityCacheTTL)
+		im.addToIdentityCache(identity)
 	}
 	return identity, nil
 }
-
-func (im *identityManager) CachedVerifierLookup(ctx context.Context, vType core.VerifierType, value string) (verifier *core.Verifier, err error) {
-	// Use an LRU cache for the author identity, as it's likely for the same identity to be re-used over and over
-	cacheKey := fmt.Sprintf("v=%s|%s", vType, value)
-	if cached := im.identityCache.Get(cacheKey); cached != nil {
-		cached.Extend(im.identityCacheTTL)
-		verifier = cached.Value().(*core.Verifier)
-	} else {
-		verifier, err = im.database.GetVerifierByValue(ctx, vType, im.namespace, value)
-		if err != nil || verifier == nil {
-			return verifier, err
-		}
-		// Cache the result
-		im.identityCache.Set(cacheKey, verifier, im.identityCacheTTL)
-	}
-	return verifier, nil
-}
diff --git a/internal/multiparty/manager.go b/internal/multiparty/manager.go
index dcc6a05b0..7a03e677b 100644
--- a/internal/multiparty/manager.go
+++ b/internal/multiparty/manager.go
@@ -38,6 +38,9 @@ type Manager interface {
 	// RootOrg returns configuration details for the root organization identity
 	RootOrg() RootOrg
 
+	// LocalNode returns configuration details for the local node identity
+	LocalNode() LocalNode
+
 	// ConfigureContract initializes the subscription to the FireFly contract
 	// - Determines the active multiparty contract entry from the config, and updates the namespace with contract info
 	// - Resolves the multiparty contract address and version, and initializes subscriptions for contract events
@@ -66,6 +69,7 @@ type Manager interface {
 type Config struct {
 	Enabled   bool
 	Org       RootOrg
+	Node      LocalNode
 	Contracts []Contract
 }
 
@@ -75,6 +79,11 @@ type RootOrg struct {
 	Key         string
 }
 
+type LocalNode struct {
+	Name        string
+	Description string
+}
+
 type Contract struct {
 	Location   *fftypes.JSONAny
 	FirstEvent string
@@ -118,6 +127,10 @@ func (mm *multipartyManager) RootOrg() RootOrg {
 	return mm.config.Org
 }
 
+func (mm *multipartyManager) LocalNode() LocalNode {
+	return mm.config.Node
+}
+
 func (mm *multipartyManager) ConfigureContract(ctx context.Context) (err error) {
 	return mm.configureContractCommon(ctx, false)
 }
diff --git a/internal/multiparty/operations.go b/internal/multiparty/operations.go
index 5dfe0beb5..7e39f9aa2 100644
--- a/internal/multiparty/operations.go
+++ b/internal/multiparty/operations.go
@@ -22,7 +22,6 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly/internal/coremsgs"
-	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/pkg/blockchain"
 	"github.com/hyperledger/firefly/pkg/core"
 )
@@ -126,7 +125,7 @@ func (mm *multipartyManager) RunOperation(ctx context.Context, op *core.Prepared
 	}
 }
 
-func (mm *multipartyManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *operations.OperationUpdate) error {
+func (mm *multipartyManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *core.OperationUpdate) error {
 	return nil
 }
 
diff --git a/internal/namespace/config.go b/internal/namespace/config.go
index b6da65d2d..1a8909464 100644
--- a/internal/namespace/config.go
+++ b/internal/namespace/config.go
@@ -45,6 +45,8 @@ func InitConfig(withDefaults bool) {
 	multipartyConf.AddKnownKey(coreconfig.NamespaceMultipartyOrgName)
 	multipartyConf.AddKnownKey(coreconfig.NamespaceMultipartyOrgDescription)
 	multipartyConf.AddKnownKey(coreconfig.NamespaceMultipartyOrgKey)
+	multipartyConf.AddKnownKey(coreconfig.NamespaceMultipartyNodeName)
+	multipartyConf.AddKnownKey(coreconfig.NamespaceMultipartyNodeDescription)
 
 	contractConf := multipartyConf.SubArray(coreconfig.NamespaceMultipartyContract)
 	contractConf.AddKnownKey(coreconfig.NamespaceMultipartyContractFirstEvent, core.SubOptsFirstEventOldest)
diff --git a/internal/namespace/manager.go b/internal/namespace/manager.go
index e22a2a28a..07e22cad9 100644
--- a/internal/namespace/manager.go
+++ b/internal/namespace/manager.go
@@ -796,9 +796,13 @@ func (nm *namespaceManager) loadNamespace(ctx context.Context, name string, inde
 	orgName := multipartyConf.GetString(coreconfig.NamespaceMultipartyOrgName)
 	orgKey := multipartyConf.GetString(coreconfig.NamespaceMultipartyOrgKey)
 	orgDesc := multipartyConf.GetString(coreconfig.NamespaceMultipartyOrgDescription)
+	nodeName := multipartyConf.GetString(coreconfig.NamespaceMultipartyNodeName)
+	nodeDesc := multipartyConf.GetString(coreconfig.NamespaceMultipartyNodeDescription)
 	deprecatedOrgName := config.GetString(coreconfig.OrgName)
 	deprecatedOrgKey := config.GetString(coreconfig.OrgKey)
 	deprecatedOrgDesc := config.GetString(coreconfig.OrgDescription)
+	deprecatedNodeName := config.GetString(coreconfig.NodeName)
+	deprecatedNodeDesc := config.GetString(coreconfig.NodeDescription)
 	if deprecatedOrgName != "" || deprecatedOrgKey != "" || deprecatedOrgDesc != "" {
 		log.L(ctx).Warnf("Your org config uses a deprecated configuration structure - the org configuration has been moved under the 'namespaces.predefined[].multiparty' section")
 	}
@@ -811,6 +815,13 @@ func (nm *namespaceManager) loadNamespace(ctx context.Context, name string, inde
 	if orgDesc == "" {
 		orgDesc = deprecatedOrgDesc
 	}
+	if nodeName == "" {
+		nodeName = deprecatedNodeName
+	}
+	if nodeDesc == "" {
+		nodeDesc = deprecatedNodeDesc
+	}
+
 	multipartyEnabled := multipartyConf.Get(coreconfig.NamespaceMultipartyEnabled)
 	if multipartyEnabled == nil {
 		multipartyEnabled = orgName != "" || orgKey != ""
@@ -874,6 +885,8 @@ func (nm *namespaceManager) loadNamespace(ctx context.Context, name string, inde
 		config.Multiparty.Org.Key = orgKey
 		config.Multiparty.Org.Description = orgDesc
 		config.Multiparty.Contracts = contracts
+		config.Multiparty.Node.Name = nodeName
+		config.Multiparty.Node.Description = nodeDesc
 	}
 
 	return &namespace{
diff --git a/internal/networkmap/register_node.go b/internal/networkmap/register_node.go
index c5a56ead4..7f516c9a0 100644
--- a/internal/networkmap/register_node.go
+++ b/internal/networkmap/register_node.go
@@ -18,10 +18,9 @@ package networkmap
 
 import (
 	"context"
-	"fmt"
 
-	"github.com/hyperledger/firefly-common/pkg/config"
-	"github.com/hyperledger/firefly/internal/coreconfig"
+	"github.com/hyperledger/firefly-common/pkg/i18n"
+	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/pkg/core"
 )
 
@@ -32,19 +31,18 @@ func (nm *networkMap) RegisterNode(ctx context.Context, waitConfirm bool) (ident
 		return nil, err
 	}
 
+	localNodeName := nm.multiparty.LocalNode().Name
+	if localNodeName == "" {
+		return nil, i18n.NewError(ctx, coremsgs.MsgNodeAndOrgIDMustBeSet)
+	}
 	nodeRequest := &core.IdentityCreateDTO{
 		Parent: nodeOwningOrg.ID.String(),
-		Name:   config.GetString(coreconfig.NodeName),
+		Name:   localNodeName,
 		Type:   core.IdentityTypeNode,
 		IdentityProfile: core.IdentityProfile{
-			Description: config.GetString(coreconfig.NodeDescription),
+			Description: nm.multiparty.LocalNode().Description,
 		},
 	}
-	if nodeRequest.Name == "" {
-		if nodeOwningOrg.Name != "" {
-			nodeRequest.Name = fmt.Sprintf("%s.node", nodeOwningOrg.Name)
-		}
-	}
 
 	dxInfo, err := nm.exchange.GetEndpointInfo(ctx)
 	if err != nil {
diff --git a/internal/operations/manager.go b/internal/operations/manager.go
index 25e1bb4cc..905e89920 100644
--- a/internal/operations/manager.go
+++ b/internal/operations/manager.go
@@ -33,7 +33,7 @@ type OperationHandler interface {
 	core.Named
 	PrepareOperation(ctx context.Context, op *core.Operation) (*core.PreparedOperation, error)
 	RunOperation(ctx context.Context, op *core.PreparedOperation) (outputs fftypes.JSONObject, complete bool, err error)
-	OnOperationUpdate(ctx context.Context, op *core.Operation, update *OperationUpdate) error
+	OnOperationUpdate(ctx context.Context, op *core.Operation, update *core.OperationUpdate) error
 }
 
 type Manager interface {
@@ -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 *OperationUpdate)
+	SubmitOperationUpdate(plugin core.Named, update *core.OperationUpdate)
 	ResolveOperationByID(ctx context.Context, opID *fftypes.UUID, op *core.OperationUpdateDTO) error
 	Start() error
 	WaitStop()
@@ -181,7 +181,7 @@ 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 *OperationUpdate) {
+func (om *operationsManager) SubmitOperationUpdate(plugin core.Named, update *core.OperationUpdate) {
 	errString := ""
 	if update.ErrorMessage != "" {
 		errString = fmt.Sprintf(" error=%s", update.ErrorMessage)
diff --git a/internal/operations/operation_updater.go b/internal/operations/operation_updater.go
index b96383196..c972e5314 100644
--- a/internal/operations/operation_updater.go
+++ b/internal/operations/operation_updater.go
@@ -34,21 +34,8 @@ import (
 	"github.com/hyperledger/firefly/pkg/database"
 )
 
-// OperationUpdate is dispatched asynchronously to perform an update.
-type OperationUpdate struct {
-	NamespacedOpID string
-	Status         core.OpStatus
-	BlockchainTXID string
-	ErrorMessage   string
-	Output         fftypes.JSONObject
-	VerifyManifest bool
-	DXManifest     string
-	DXHash         string
-	OnComplete     func()
-}
-
 type operationUpdaterBatch struct {
-	updates        []*OperationUpdate
+	updates        []*core.OperationUpdate
 	timeoutContext context.Context
 	timeoutCancel  func()
 }
@@ -60,7 +47,7 @@ type operationUpdater struct {
 	manager     *operationsManager
 	database    database.Plugin
 	txHelper    txcommon.Helper
-	workQueues  []chan *OperationUpdate
+	workQueues  []chan *core.OperationUpdate
 	workersDone []chan struct{}
 	conf        operationUpdaterConf
 	closed      bool
@@ -100,13 +87,13 @@ func newOperationUpdater(ctx context.Context, om *operationsManager, di database
 }
 
 // pickWorker ensures multiple updates for the same ID go to the same worker
-func (ou *operationUpdater) pickWorker(ctx context.Context, id *fftypes.UUID, update *OperationUpdate) chan *OperationUpdate {
+func (ou *operationUpdater) pickWorker(ctx context.Context, id *fftypes.UUID, update *core.OperationUpdate) chan *core.OperationUpdate {
 	worker := id.HashBucket(ou.conf.workerCount)
 	log.L(ctx).Debugf("Submitting operation update id=%s status=%s blockchainTX=%s worker=opu_%.3d", id, update.Status, update.BlockchainTXID, worker)
 	return ou.workQueues[worker]
 }
 
-func (ou *operationUpdater) SubmitOperationUpdate(ctx context.Context, update *OperationUpdate) {
+func (ou *operationUpdater) SubmitOperationUpdate(ctx context.Context, update *core.OperationUpdate) {
 	ns, id, 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)
@@ -126,17 +113,17 @@ func (ou *operationUpdater) SubmitOperationUpdate(ctx context.Context, update *O
 		return
 	}
 	// Otherwise do it in-line on this context
-	err = ou.doBatchUpdateWithRetry(ctx, []*OperationUpdate{update})
+	err = ou.doBatchUpdateWithRetry(ctx, []*core.OperationUpdate{update})
 	if err != nil {
 		log.L(ctx).Warnf("Exiting while updating operation: %s", err)
 	}
 }
 
 func (ou *operationUpdater) initQueues() {
-	ou.workQueues = make([]chan *OperationUpdate, ou.conf.workerCount)
+	ou.workQueues = make([]chan *core.OperationUpdate, ou.conf.workerCount)
 	ou.workersDone = make([]chan struct{}, ou.conf.workerCount)
 	for i := 0; i < ou.conf.workerCount; i++ {
-		ou.workQueues[i] = make(chan *OperationUpdate, ou.conf.queueLength)
+		ou.workQueues[i] = make(chan *core.OperationUpdate, ou.conf.queueLength)
 		ou.workersDone[i] = make(chan struct{})
 	}
 }
@@ -188,7 +175,7 @@ func (ou *operationUpdater) updaterLoop(index int) {
 	}
 }
 
-func (ou *operationUpdater) doBatchUpdateWithRetry(ctx context.Context, updates []*OperationUpdate) error {
+func (ou *operationUpdater) doBatchUpdateWithRetry(ctx context.Context, updates []*core.OperationUpdate) error {
 	return ou.retry.Do(ctx, "operation update", func(attempt int) (retry bool, err error) {
 		err = ou.database.RunAsGroup(ctx, func(ctx context.Context) error {
 			return ou.doBatchUpdate(ctx, updates)
@@ -205,7 +192,7 @@ func (ou *operationUpdater) doBatchUpdateWithRetry(ctx context.Context, updates
 	})
 }
 
-func (ou *operationUpdater) doBatchUpdate(ctx context.Context, updates []*OperationUpdate) error {
+func (ou *operationUpdater) doBatchUpdate(ctx context.Context, updates []*core.OperationUpdate) error {
 
 	// Get all the operations that match
 	opIDs := make([]driver.Value, 0, len(updates))
@@ -252,7 +239,7 @@ func (ou *operationUpdater) doBatchUpdate(ctx context.Context, updates []*Operat
 	return nil
 }
 
-func (ou *operationUpdater) doUpdate(ctx context.Context, update *OperationUpdate, ops []*core.Operation, transactions []*core.Transaction) error {
+func (ou *operationUpdater) doUpdate(ctx context.Context, update *core.OperationUpdate, ops []*core.Operation, transactions []*core.Transaction) error {
 
 	_, id, err := core.ParseNamespacedOpID(ctx, update.NamespacedOpID)
 	if err != nil {
@@ -309,7 +296,7 @@ func (ou *operationUpdater) doUpdate(ctx context.Context, update *OperationUpdat
 	return nil
 }
 
-func (ou *operationUpdater) verifyManifest(ctx context.Context, update *OperationUpdate, op *core.Operation) error {
+func (ou *operationUpdater) verifyManifest(ctx context.Context, update *core.OperationUpdate, op *core.Operation) error {
 
 	if op.Type == core.OpTypeDataExchangeSendBatch && update.Status == core.OpStatusSucceeded {
 		batchID, _ := fftypes.ParseUUID(ctx, op.Input.GetString("batch"))
diff --git a/internal/orchestrator/bound_callbacks.go b/internal/orchestrator/bound_callbacks.go
index 3eeaeee3a..368974283 100644
--- a/internal/orchestrator/bound_callbacks.go
+++ b/internal/orchestrator/bound_callbacks.go
@@ -17,66 +17,21 @@
 package orchestrator
 
 import (
-	"context"
-
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
-	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/events"
 	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/pkg/core"
-	"github.com/hyperledger/firefly/pkg/dataexchange"
 	"github.com/hyperledger/firefly/pkg/sharedstorage"
 )
 
 type boundCallbacks struct {
-	dx dataexchange.Plugin
 	ss sharedstorage.Plugin
 	ei events.EventManager
 	om operations.Manager
 }
 
-func (bc *boundCallbacks) OperationUpdate(plugin core.Named, nsOpID string, status core.OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject) {
-	bc.om.SubmitOperationUpdate(plugin, &operations.OperationUpdate{
-		NamespacedOpID: nsOpID,
-		Status:         status,
-		BlockchainTXID: blockchainTXID,
-		ErrorMessage:   errorMessage,
-		Output:         opOutput,
-	})
-}
-
-func (bc *boundCallbacks) DXEvent(ctx context.Context, event dataexchange.DXEvent) {
-	switch event.Type() {
-	case dataexchange.DXEventTypeTransferResult:
-		tr := event.TransferResult()
-		log.L(ctx).Infof("Transfer result %s=%s error='%s' manifest='%s' info='%s'", tr.TrackingID, tr.Status, tr.Error, tr.Manifest, tr.Info)
-
-		opUpdate := &operations.OperationUpdate{
-			NamespacedOpID: event.NamespacedID(),
-			Status:         tr.Status,
-			VerifyManifest: bc.dx.Capabilities().Manifest,
-			ErrorMessage:   tr.Error,
-			Output:         tr.Info,
-			OnComplete: func() {
-				event.Ack()
-			},
-		}
-
-		// Pass manifest verification code to the background worker, for once it has loaded the operation
-		if opUpdate.VerifyManifest {
-			if tr.Manifest != "" {
-				// For batches DX passes us a manifest to compare.
-				opUpdate.DXManifest = tr.Manifest
-			} else if tr.Hash != "" {
-				// For blobs DX passes us a hash to compare.
-				opUpdate.DXHash = tr.Hash
-			}
-		}
-
-		bc.om.SubmitOperationUpdate(bc.dx, opUpdate)
-	default:
-		bc.ei.DXEvent(bc.dx, event)
-	}
+func (bc *boundCallbacks) OperationUpdate(plugin core.Named, update *core.OperationUpdate) {
+	bc.om.SubmitOperationUpdate(plugin, update)
 }
 
 func (bc *boundCallbacks) SharedStorageBatchDownloaded(payloadRef string, data []byte) (*fftypes.UUID, error) {
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index 04be69396..5f44e3b0d 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -224,7 +224,6 @@ func (or *orchestrator) Init(ctx context.Context, cancelCtx context.CancelFunc)
 	}
 	// Bind together the blockchain interface callbacks, with the events manager
 	or.bc.ei = or.events
-	or.bc.dx = or.plugins.DataExchange.Plugin
 	or.bc.ss = or.plugins.SharedStorage.Plugin
 	or.bc.om = or.operations
 	return err
@@ -376,7 +375,8 @@ func (or *orchestrator) initHandlers(ctx context.Context) (err error) {
 			nodeInfo[i] = node.Profile
 		}
 		or.plugins.DataExchange.Plugin.SetNodes(nodeInfo)
-		or.plugins.DataExchange.Plugin.SetHandler(or.namespace.RemoteName, &or.bc)
+		or.plugins.DataExchange.Plugin.SetHandler(or.namespace.RemoteName, or.events)
+		or.plugins.DataExchange.Plugin.SetOperationHandler(or.namespace.LocalName, &or.bc)
 	}
 
 	for _, token := range or.plugins.Tokens {
diff --git a/internal/orchestrator/status.go b/internal/orchestrator/status.go
index 04cf28191..86b7ee1ca 100644
--- a/internal/orchestrator/status.go
+++ b/internal/orchestrator/status.go
@@ -20,10 +20,8 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/hyperledger/firefly-common/pkg/config"
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/log"
-	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 )
@@ -102,16 +100,14 @@ func (or *orchestrator) GetStatus(ctx context.Context) (status *core.NamespaceSt
 
 	status = &core.NamespaceStatus{
 		Namespace: or.namespace,
-		Node: &core.NamespaceStatusNode{
-			Name: config.GetString(coreconfig.NodeName),
-		},
-		Plugins: or.getPlugins(),
+		Plugins:   or.getPlugins(),
 		Multiparty: core.NamespaceStatusMultiparty{
 			Enabled: or.config.Multiparty.Enabled,
 		},
 	}
 
 	if or.config.Multiparty.Enabled {
+		status.Node = &core.NamespaceStatusNode{Name: or.config.Multiparty.Node.Name}
 		status.Org = &core.NamespaceStatusOrg{Name: or.config.Multiparty.Org.Name}
 		status.Multiparty.Contracts = or.namespace.Contracts
 
diff --git a/internal/privatemessaging/operations.go b/internal/privatemessaging/operations.go
index 2530475a2..6fc4de3c9 100644
--- a/internal/privatemessaging/operations.go
+++ b/internal/privatemessaging/operations.go
@@ -23,7 +23,6 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly/internal/coremsgs"
-	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/pkg/core"
 )
 
@@ -144,7 +143,7 @@ func (pm *privateMessaging) RunOperation(ctx context.Context, op *core.PreparedO
 	}
 }
 
-func (pm *privateMessaging) OnOperationUpdate(ctx context.Context, op *core.Operation, update *operations.OperationUpdate) error {
+func (pm *privateMessaging) OnOperationUpdate(ctx context.Context, op *core.Operation, update *core.OperationUpdate) error {
 	return nil
 }
 
diff --git a/internal/privatemessaging/privatemessaging.go b/internal/privatemessaging/privatemessaging.go
index f215ccabb..322c27304 100644
--- a/internal/privatemessaging/privatemessaging.go
+++ b/internal/privatemessaging/privatemessaging.go
@@ -71,8 +71,6 @@ type privateMessaging struct {
 	syncasync             syncasync.Bridge
 	multiparty            multiparty.Manager
 	retry                 retry.Retry
-	localNodeName         string
-	localNodeID           *fftypes.UUID // lookup and cached on first use, as might not be registered at startup
 	maxBatchPayloadLength int64
 	metrics               metrics.Manager
 	operations            operations.Manager
@@ -91,16 +89,15 @@ func NewPrivateMessaging(ctx context.Context, ns core.NamespaceRef, di database.
 	}
 
 	pm := &privateMessaging{
-		ctx:           ctx,
-		namespace:     ns,
-		database:      di,
-		identity:      im,
-		exchange:      dx,
-		blockchain:    bi,
-		data:          dm,
-		syncasync:     sa,
-		multiparty:    mult,
-		localNodeName: config.GetString(coreconfig.NodeName),
+		ctx:        ctx,
+		namespace:  ns,
+		database:   di,
+		identity:   im,
+		exchange:   dx,
+		blockchain: bi,
+		data:       dm,
+		syncasync:  sa,
+		multiparty: mult,
 		groupManager: groupManager{
 			namespace:     ns,
 			database:      di,
diff --git a/internal/privatemessaging/recipients.go b/internal/privatemessaging/recipients.go
index 7bb192af6..81c375b54 100644
--- a/internal/privatemessaging/recipients.go
+++ b/internal/privatemessaging/recipients.go
@@ -113,6 +113,10 @@ func (pm *privateMessaging) getRecipients(ctx context.Context, in *core.MessageI
 	}
 
 	foundLocal := false
+	localNode, err := pm.identity.GetLocalNode(ctx)
+	if err != nil {
+		return nil, err
+	}
 	gi = &core.GroupIdentity{
 		Namespace: in.Message.Header.Namespace,
 		Name:      in.Group.Name,
@@ -129,7 +133,7 @@ func (pm *privateMessaging) getRecipients(ctx context.Context, in *core.MessageI
 		if err != nil {
 			return nil, err
 		}
-		isLocal := (node.Parent.Equals(localOrg.ID) && node.Name == pm.localNodeName)
+		isLocal := (node.Parent.Equals(localOrg.ID) && node.Name == localNode.Name)
 		foundLocal = foundLocal || isLocal
 		log.L(ctx).Debugf("Resolved group identity %s node=%s to identity %s node=%s local=%t", rInput.Identity, rInput.Node, identity.DID, node.ID, isLocal)
 		gi.Members[i] = &core.Member{
@@ -138,40 +142,15 @@ func (pm *privateMessaging) getRecipients(ctx context.Context, in *core.MessageI
 		}
 	}
 	if !foundLocal {
-		// Add in the local org identity
-		localNodeID, err := pm.resolveLocalNode(ctx, localOrg)
-		if err != nil {
-			return nil, err
-		}
+		// Add in the local org/node identity
 		gi.Members = append(gi.Members, &core.Member{
 			Identity: localOrg.DID,
-			Node:     localNodeID,
+			Node:     localNode.ID,
 		})
 	}
 	return gi, nil
 }
 
-func (pm *privateMessaging) resolveLocalNode(ctx context.Context, localOrg *core.Identity) (*fftypes.UUID, error) {
-	if pm.localNodeID != nil {
-		return pm.localNodeID, nil
-	}
-	fb := database.IdentityQueryFactory.NewFilterLimit(ctx, 1)
-	filter := fb.And(
-		fb.Eq("parent", localOrg.ID),
-		fb.Eq("type", core.IdentityTypeNode),
-		fb.Eq("name", pm.localNodeName),
-	)
-	nodes, _, err := pm.database.GetIdentities(ctx, pm.namespace.LocalName, filter)
-	if err != nil {
-		return nil, err
-	}
-	if len(nodes) == 0 {
-		return nil, i18n.NewError(ctx, coremsgs.MsgLocalNodeResolveFailed)
-	}
-	pm.localNodeID = nodes[0].ID
-	return pm.localNodeID, nil
-}
-
 func (pm *privateMessaging) findOrGenerateGroup(ctx context.Context, in *core.MessageInOut) (group *core.Group, isNew bool, err error) {
 	gi, err := pm.getRecipients(ctx, in)
 	if err != nil {
diff --git a/internal/shareddownload/operations.go b/internal/shareddownload/operations.go
index f48463e6a..a63757402 100644
--- a/internal/shareddownload/operations.go
+++ b/internal/shareddownload/operations.go
@@ -26,7 +26,6 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/coremsgs"
-	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/pkg/core"
 )
 
@@ -161,7 +160,7 @@ func (dm *downloadManager) downloadBlob(ctx context.Context, data downloadBlobDa
 	return getDownloadBlobOutputs(hash, blobSize, dxPayloadRef), true, nil
 }
 
-func (dm *downloadManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *operations.OperationUpdate) error {
+func (dm *downloadManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *core.OperationUpdate) error {
 	return nil
 }
 
diff --git a/internal/tokens/fftokens/fftokens.go b/internal/tokens/fftokens/fftokens.go
index 15d125a29..905908fd5 100644
--- a/internal/tokens/fftokens/fftokens.go
+++ b/internal/tokens/fftokens/fftokens.go
@@ -51,7 +51,13 @@ 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, nsOpID, status, blockchainTXID, errorMessage, opOutput)
+		handler.OperationUpdate(cb.plugin, &core.OperationUpdate{
+			NamespacedOpID: nsOpID,
+			Status:         status,
+			BlockchainTXID: blockchainTXID,
+			ErrorMessage:   errorMessage,
+			Output:         opOutput,
+		})
 	} else {
 		log.L(ctx).Errorf("No handler found for token operation '%s'", nsOpID)
 	}
diff --git a/mocks/coremocks/operation_callbacks.go b/mocks/coremocks/operation_callbacks.go
index 8116d9cb7..37a3aa0e5 100644
--- a/mocks/coremocks/operation_callbacks.go
+++ b/mocks/coremocks/operation_callbacks.go
@@ -3,9 +3,7 @@
 package coremocks
 
 import (
-	fftypes "github.com/hyperledger/firefly-common/pkg/fftypes"
 	core "github.com/hyperledger/firefly/pkg/core"
-
 	mock "github.com/stretchr/testify/mock"
 )
 
@@ -14,7 +12,7 @@ type OperationCallbacks struct {
 	mock.Mock
 }
 
-// OperationUpdate provides a mock function with given fields: plugin, nsOpID, status, blockchainTXID, errorMessage, opOutput
-func (_m *OperationCallbacks) OperationUpdate(plugin core.Named, nsOpID string, status core.OpStatus, blockchainTXID string, errorMessage string, opOutput fftypes.JSONObject) {
-	_m.Called(plugin, nsOpID, status, blockchainTXID, errorMessage, opOutput)
+// OperationUpdate provides a mock function with given fields: plugin, update
+func (_m *OperationCallbacks) OperationUpdate(plugin core.Named, update *core.OperationUpdate) {
+	_m.Called(plugin, update)
 }
diff --git a/mocks/dataexchangemocks/callbacks.go b/mocks/dataexchangemocks/callbacks.go
index af463b028..559b61f05 100644
--- a/mocks/dataexchangemocks/callbacks.go
+++ b/mocks/dataexchangemocks/callbacks.go
@@ -3,8 +3,6 @@
 package dataexchangemocks
 
 import (
-	context "context"
-
 	dataexchange "github.com/hyperledger/firefly/pkg/dataexchange"
 	mock "github.com/stretchr/testify/mock"
 )
@@ -14,7 +12,7 @@ type Callbacks struct {
 	mock.Mock
 }
 
-// DXEvent provides a mock function with given fields: ctx, event
-func (_m *Callbacks) DXEvent(ctx context.Context, event dataexchange.DXEvent) {
-	_m.Called(ctx, event)
+// DXEvent provides a mock function with given fields: plugin, event
+func (_m *Callbacks) DXEvent(plugin dataexchange.Plugin, event dataexchange.DXEvent) {
+	_m.Called(plugin, event)
 }
diff --git a/mocks/dataexchangemocks/dx_event.go b/mocks/dataexchangemocks/dx_event.go
index d251dd534..e7e2fcf52 100644
--- a/mocks/dataexchangemocks/dx_event.go
+++ b/mocks/dataexchangemocks/dx_event.go
@@ -52,20 +52,6 @@ func (_m *DXEvent) MessageReceived() *dataexchange.MessageReceived {
 	return r0
 }
 
-// NamespacedID provides a mock function with given fields:
-func (_m *DXEvent) NamespacedID() string {
-	ret := _m.Called()
-
-	var r0 string
-	if rf, ok := ret.Get(0).(func() string); ok {
-		r0 = rf()
-	} else {
-		r0 = ret.Get(0).(string)
-	}
-
-	return r0
-}
-
 // PrivateBlobReceived provides a mock function with given fields:
 func (_m *DXEvent) PrivateBlobReceived() *dataexchange.PrivateBlobReceived {
 	ret := _m.Called()
@@ -82,22 +68,6 @@ func (_m *DXEvent) PrivateBlobReceived() *dataexchange.PrivateBlobReceived {
 	return r0
 }
 
-// TransferResult provides a mock function with given fields:
-func (_m *DXEvent) TransferResult() *dataexchange.TransferResult {
-	ret := _m.Called()
-
-	var r0 *dataexchange.TransferResult
-	if rf, ok := ret.Get(0).(func() *dataexchange.TransferResult); ok {
-		r0 = rf()
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*dataexchange.TransferResult)
-		}
-	}
-
-	return r0
-}
-
 // Type provides a mock function with given fields:
 func (_m *DXEvent) Type() dataexchange.DXEventType {
 	ret := _m.Called()
diff --git a/mocks/dataexchangemocks/plugin.go b/mocks/dataexchangemocks/plugin.go
index db294a069..450185f35 100644
--- a/mocks/dataexchangemocks/plugin.go
+++ b/mocks/dataexchangemocks/plugin.go
@@ -7,6 +7,8 @@ import (
 
 	config "github.com/hyperledger/firefly-common/pkg/config"
 
+	core "github.com/hyperledger/firefly/pkg/core"
+
 	dataexchange "github.com/hyperledger/firefly/pkg/dataexchange"
 
 	fftypes "github.com/hyperledger/firefly-common/pkg/fftypes"
@@ -184,6 +186,11 @@ func (_m *Plugin) SetNodes(nodes []fftypes.JSONObject) {
 	_m.Called(nodes)
 }
 
+// SetOperationHandler provides a mock function with given fields: namespace, handler
+func (_m *Plugin) SetOperationHandler(namespace string, handler core.OperationCallbacks) {
+	_m.Called(namespace, handler)
+}
+
 // Start provides a mock function with given fields:
 func (_m *Plugin) Start() error {
 	ret := _m.Called()
diff --git a/mocks/eventmocks/event_manager.go b/mocks/eventmocks/event_manager.go
index fb59c83c5..5cf46353f 100644
--- a/mocks/eventmocks/event_manager.go
+++ b/mocks/eventmocks/event_manager.go
@@ -97,9 +97,9 @@ func (_m *EventManager) CreateUpdateDurableSubscription(ctx context.Context, sub
 	return r0
 }
 
-// DXEvent provides a mock function with given fields: dx, event
-func (_m *EventManager) DXEvent(dx dataexchange.Plugin, event dataexchange.DXEvent) {
-	_m.Called(dx, event)
+// DXEvent provides a mock function with given fields: plugin, event
+func (_m *EventManager) DXEvent(plugin dataexchange.Plugin, event dataexchange.DXEvent) {
+	_m.Called(plugin, event)
 }
 
 // DeleteDurableSubscription provides a mock function with given fields: ctx, subDef
diff --git a/mocks/identitymanagermocks/manager.go b/mocks/identitymanagermocks/manager.go
index d857118c9..b184f4e0e 100644
--- a/mocks/identitymanagermocks/manager.go
+++ b/mocks/identitymanagermocks/manager.go
@@ -99,22 +99,22 @@ func (_m *Manager) CachedIdentityLookupNilOK(ctx context.Context, did string) (*
 	return r0, r1, r2
 }
 
-// CachedVerifierLookup provides a mock function with given fields: ctx, vType, value
-func (_m *Manager) CachedVerifierLookup(ctx context.Context, vType fftypes.FFEnum, value string) (*core.Verifier, error) {
-	ret := _m.Called(ctx, vType, value)
+// FindIdentityForVerifier provides a mock function with given fields: ctx, iTypes, verifier
+func (_m *Manager) FindIdentityForVerifier(ctx context.Context, iTypes []fftypes.FFEnum, verifier *core.VerifierRef) (*core.Identity, error) {
+	ret := _m.Called(ctx, iTypes, verifier)
 
-	var r0 *core.Verifier
-	if rf, ok := ret.Get(0).(func(context.Context, fftypes.FFEnum, string) *core.Verifier); ok {
-		r0 = rf(ctx, vType, value)
+	var r0 *core.Identity
+	if rf, ok := ret.Get(0).(func(context.Context, []fftypes.FFEnum, *core.VerifierRef) *core.Identity); ok {
+		r0 = rf(ctx, iTypes, verifier)
 	} else {
 		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*core.Verifier)
+			r0 = ret.Get(0).(*core.Identity)
 		}
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, fftypes.FFEnum, string) error); ok {
-		r1 = rf(ctx, vType, value)
+	if rf, ok := ret.Get(1).(func(context.Context, []fftypes.FFEnum, *core.VerifierRef) error); ok {
+		r1 = rf(ctx, iTypes, verifier)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -122,13 +122,13 @@ func (_m *Manager) CachedVerifierLookup(ctx context.Context, vType fftypes.FFEnu
 	return r0, r1
 }
 
-// FindIdentityForVerifier provides a mock function with given fields: ctx, iTypes, verifier
-func (_m *Manager) FindIdentityForVerifier(ctx context.Context, iTypes []fftypes.FFEnum, verifier *core.VerifierRef) (*core.Identity, error) {
-	ret := _m.Called(ctx, iTypes, verifier)
+// GetLocalNode provides a mock function with given fields: ctx
+func (_m *Manager) GetLocalNode(ctx context.Context) (*core.Identity, error) {
+	ret := _m.Called(ctx)
 
 	var r0 *core.Identity
-	if rf, ok := ret.Get(0).(func(context.Context, []fftypes.FFEnum, *core.VerifierRef) *core.Identity); ok {
-		r0 = rf(ctx, iTypes, verifier)
+	if rf, ok := ret.Get(0).(func(context.Context) *core.Identity); ok {
+		r0 = rf(ctx)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(*core.Identity)
@@ -136,8 +136,8 @@ func (_m *Manager) FindIdentityForVerifier(ctx context.Context, iTypes []fftypes
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, []fftypes.FFEnum, *core.VerifierRef) error); ok {
-		r1 = rf(ctx, iTypes, verifier)
+	if rf, ok := ret.Get(1).(func(context.Context) error); ok {
+		r1 = rf(ctx)
 	} else {
 		r1 = ret.Error(1)
 	}
diff --git a/mocks/multipartymocks/manager.go b/mocks/multipartymocks/manager.go
index 4b3a75d1a..052034f9c 100644
--- a/mocks/multipartymocks/manager.go
+++ b/mocks/multipartymocks/manager.go
@@ -49,6 +49,20 @@ func (_m *Manager) GetNetworkVersion() int {
 	return r0
 }
 
+// LocalNode provides a mock function with given fields:
+func (_m *Manager) LocalNode() multiparty.LocalNode {
+	ret := _m.Called()
+
+	var r0 multiparty.LocalNode
+	if rf, ok := ret.Get(0).(func() multiparty.LocalNode); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Get(0).(multiparty.LocalNode)
+	}
+
+	return r0
+}
+
 // Name provides a mock function with given fields:
 func (_m *Manager) Name() string {
 	ret := _m.Called()
diff --git a/mocks/operationmocks/manager.go b/mocks/operationmocks/manager.go
index 5029ad825..88c74807f 100644
--- a/mocks/operationmocks/manager.go
+++ b/mocks/operationmocks/manager.go
@@ -142,7 +142,7 @@ func (_m *Manager) Start() error {
 }
 
 // SubmitOperationUpdate provides a mock function with given fields: plugin, update
-func (_m *Manager) SubmitOperationUpdate(plugin core.Named, update *operations.OperationUpdate) {
+func (_m *Manager) SubmitOperationUpdate(plugin core.Named, update *core.OperationUpdate) {
 	_m.Called(plugin, update)
 }
 
diff --git a/pkg/core/namespace_status.go b/pkg/core/namespace_status.go
index de35fb745..7259ea45d 100644
--- a/pkg/core/namespace_status.go
+++ b/pkg/core/namespace_status.go
@@ -21,7 +21,7 @@ import "github.com/hyperledger/firefly-common/pkg/fftypes"
 // NamespaceStatus is a set of information that represents the configuration and status of a given namespace
 type NamespaceStatus struct {
 	Namespace  *Namespace                `ffstruct:"NamespaceStatus" json:"namespace"`
-	Node       *NamespaceStatusNode      `ffstruct:"NamespaceStatus" json:"node"`
+	Node       *NamespaceStatusNode      `ffstruct:"NamespaceStatus" json:"node,omitempty"`
 	Org        *NamespaceStatusOrg       `ffstruct:"NamespaceStatus" json:"org,omitempty"`
 	Plugins    NamespaceStatusPlugins    `ffstruct:"NamespaceStatus" json:"plugins"`
 	Multiparty NamespaceStatusMultiparty `ffstruct:"NamespaceStatus" json:"multiparty"`
diff --git a/pkg/core/operation.go b/pkg/core/operation.go
index cde2e4a33..0edf3054a 100644
--- a/pkg/core/operation.go
+++ b/pkg/core/operation.go
@@ -143,10 +143,22 @@ func ParseNamespacedOpID(ctx context.Context, nsIDStr string) (string, *fftypes.
 }
 
 type OperationCallbacks interface {
-	// OperationUpdate notifies FireFly of an update to an operation.
-	// Only success/failure and errorMessage (for errors) are modeled.
-	// opOutput can be used to add opaque protocol-specific JSON from the plugin (protocol transaction ID etc.)
-	// 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.
-	OperationUpdate(plugin Named, nsOpID string, status OpStatus, blockchainTXID, errorMessage string, opOutput fftypes.JSONObject)
+	OperationUpdate(plugin Named, update *OperationUpdate)
+}
+
+// OperationUpdate notifies FireFly of an update to an operation.
+// Only success/failure and errorMessage (for errors) are modeled.
+// Output can be used to add opaque protocol-specific JSON from the plugin (protocol transaction ID etc.)
+// 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 {
+	NamespacedOpID string
+	Status         OpStatus
+	BlockchainTXID string
+	ErrorMessage   string
+	Output         fftypes.JSONObject
+	VerifyManifest bool
+	DXManifest     string
+	DXHash         string
+	OnComplete     func()
 }
diff --git a/pkg/core/transport_wrapper.go b/pkg/core/transport_wrapper.go
index 2674972b6..c15dee5ce 100644
--- a/pkg/core/transport_wrapper.go
+++ b/pkg/core/transport_wrapper.go
@@ -30,10 +30,3 @@ type TransportWrapper struct {
 	Group *Group `json:"group,omitempty"`
 	Batch *Batch `json:"batch,omitempty"`
 }
-
-type TransportStatusUpdate struct {
-	Error    string             `json:"error,omitempty"`
-	Manifest string             `json:"manifest,omitempty"`
-	Info     fftypes.JSONObject `json:"info,omitempty"`
-	Hash     string             `json:"hash,omitempty"`
-}
diff --git a/pkg/dataexchange/plugin.go b/pkg/dataexchange/plugin.go
index 6f8a22fab..f301d06c4 100644
--- a/pkg/dataexchange/plugin.go
+++ b/pkg/dataexchange/plugin.go
@@ -70,6 +70,10 @@ type Plugin interface {
 	// Plugin will attempt (but is not guaranteed) to deliver events only for the given namespace
 	SetHandler(namespace string, handler Callbacks)
 
+	// SetOperationHandler registers a handler to receive async operation status
+	// If namespace is set, plugin will attempt to deliver only events for that namespace
+	SetOperationHandler(namespace string, handler core.OperationCallbacks)
+
 	// Data exchange interface must not deliver any events until start is called
 	Start() error
 
@@ -102,7 +106,7 @@ type Plugin interface {
 // Callbacks is the interface provided to the data exchange plugin, to allow it to pass events back to firefly.
 type Callbacks interface {
 	// Event has sub-types as defined below, and can be processed and ack'd asynchronously
-	DXEvent(ctx context.Context, event DXEvent)
+	DXEvent(plugin Plugin, event DXEvent)
 }
 
 type DXEventType int
@@ -110,19 +114,16 @@ 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)
 	Type() DXEventType
 	MessageReceived() *MessageReceived
 	PrivateBlobReceived() *PrivateBlobReceived
-	TransferResult() *TransferResult
 }
 
 const (
 	DXEventTypeMessageReceived DXEventType = iota
 	DXEventTypePrivateBlobReceived
-	DXEventTypeTransferResult
 )
 
 type MessageReceived struct {
@@ -138,12 +139,6 @@ type PrivateBlobReceived struct {
 	PayloadRef string
 }
 
-type TransferResult struct {
-	TrackingID string
-	Status     core.OpStatus
-	core.TransportStatusUpdate
-}
-
 // Capabilities the supported featureset of the data exchange
 // interface implemented by the plugin, with the specified config
 type Capabilities struct {


Test patch:
diff --git a/internal/assets/operations_test.go b/internal/assets/operations_test.go
index ef6d0ee9f..2d132580d 100644
--- a/internal/assets/operations_test.go
+++ b/internal/assets/operations_test.go
@@ -21,7 +21,6 @@ import (
 	"testing"
 
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
-	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/tokenmocks"
@@ -551,7 +550,7 @@ func TestOperationUpdatePool(t *testing.T) {
 	err := txcommon.AddTokenPoolCreateInputs(op, pool)
 	assert.NoError(t, err)
 
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusFailed,
 	}
 
@@ -575,7 +574,7 @@ func TestOperationUpdatePoolBadInput(t *testing.T) {
 		ID:   fftypes.NewUUID(),
 		Type: core.OpTypeTokenCreatePool,
 	}
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusFailed,
 	}
 
@@ -599,7 +598,7 @@ func TestOperationUpdatePoolEventFail(t *testing.T) {
 		ID:   fftypes.NewUUID(),
 		Type: core.OpTypeTokenCreatePool,
 	}
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusFailed,
 	}
 
@@ -629,7 +628,7 @@ func TestOperationUpdateTransfer(t *testing.T) {
 	err := txcommon.AddTokenTransferInputs(op, transfer)
 	assert.NoError(t, err)
 
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusFailed,
 	}
 
@@ -653,7 +652,7 @@ func TestOperationUpdateTransferBadInput(t *testing.T) {
 		ID:   fftypes.NewUUID(),
 		Type: core.OpTypeTokenTransfer,
 	}
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusFailed,
 	}
 
@@ -677,7 +676,7 @@ func TestOperationUpdateTransferEventFail(t *testing.T) {
 		ID:   fftypes.NewUUID(),
 		Type: core.OpTypeTokenTransfer,
 	}
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusFailed,
 	}
 
@@ -706,7 +705,7 @@ func TestOperationUpdateApproval(t *testing.T) {
 	err := txcommon.AddTokenApprovalInputs(op, approval)
 	assert.NoError(t, err)
 
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusFailed,
 	}
 
@@ -730,7 +729,7 @@ func TestOperationUpdateApprovalBadInput(t *testing.T) {
 		ID:   fftypes.NewUUID(),
 		Type: core.OpTypeTokenApproval,
 	}
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusFailed,
 	}
 
@@ -754,7 +753,7 @@ func TestOperationUpdateApprovalEventFail(t *testing.T) {
 		ID:   fftypes.NewUUID(),
 		Type: core.OpTypeTokenApproval,
 	}
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusFailed,
 	}
 
diff --git a/internal/blockchain/common/common_test.go b/internal/blockchain/common/common_test.go
index 92e974ec9..8b1c48573 100644
--- a/internal/blockchain/common/common_test.go
+++ b/internal/blockchain/common/common_test.go
@@ -38,7 +38,12 @@ func TestCallbackOperationUpdate(t *testing.T) {
 	cb := NewBlockchainCallbacks()
 	cb.SetOperationalHandler("ns1", mcb)
 
-	mcb.On("OperationUpdate", mbi, nsOpID, core.OpStatusSucceeded, "tx1", "err", mock.Anything).Return().Once()
+	mcb.On("OperationUpdate", mbi, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+		return update.NamespacedOpID == nsOpID &&
+			update.Status == core.OpStatusSucceeded &&
+			update.BlockchainTXID == "tx1" &&
+			update.ErrorMessage == "err"
+	})).Return().Once()
 	cb.OperationUpdate(context.Background(), mbi, nsOpID, core.OpStatusSucceeded, "tx1", "err", fftypes.JSONObject{})
 
 	nsOpID = "ns2:" + fftypes.NewUUID().String()
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index 7a4676363..0aa7292fd 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -1339,13 +1339,11 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 		"transactionIndex": "0"
   }`)
 
-	em.On("OperationUpdate",
-		e,
-		"ns1:"+operationID.String(),
-		core.OpStatusSucceeded,
-		"0x71a38acb7a5d4a970854f6d638ceb1fa10a4b59cbf4ed7674273a1a8dc8b36b8",
-		"",
-		mock.Anything).Return(nil)
+	em.On("OperationUpdate", e, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+		return update.NamespacedOpID == "ns1:"+operationID.String() &&
+			update.Status == core.OpStatusSucceeded &&
+			update.BlockchainTXID == "0x71a38acb7a5d4a970854f6d638ceb1fa10a4b59cbf4ed7674273a1a8dc8b36b8"
+	})).Return(nil)
 
 	err := json.Unmarshal(data.Bytes(), &reply)
 	assert.NoError(t, err)
@@ -1381,13 +1379,11 @@ func TestHandleBadPayloadsAndThenReceiptFailure(t *testing.T) {
 
 	em := &coremocks.OperationCallbacks{}
 	e.SetOperationHandler("ns1", em)
-	txsu := em.On("OperationUpdate",
-		e,
-		"ns1:"+operationID.String(),
-		core.OpStatusFailed,
-		"",
-		"Packing arguments for method 'broadcastBatch': abi: cannot use [3]uint8 as type [32]uint8 as argument",
-		mock.Anything).Return(fmt.Errorf("Shutdown"))
+	txsu := em.On("OperationUpdate", e, 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"
+	})).Return(fmt.Errorf("Shutdown"))
 	done := make(chan struct{})
 	txsu.RunFn = func(a mock.Arguments) {
 		close(done)
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index ec42f1720..fe038adbc 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -1256,13 +1256,11 @@ func TestHandleReceiptTXSuccess(t *testing.T) {
 		"receivedAt": 1630033474675
   }`)
 
-	em.On("OperationUpdate",
-		e,
-		"ns1:"+operationID.String(),
-		core.OpStatusSucceeded,
-		"ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2",
-		"",
-		mock.Anything).Return(nil)
+	em.On("OperationUpdate", e, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+		return update.NamespacedOpID == "ns1:"+operationID.String() &&
+			update.Status == core.OpStatusSucceeded &&
+			update.BlockchainTXID == "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2"
+	})).Return(nil)
 
 	err := json.Unmarshal(data, &reply)
 	assert.NoError(t, err)
@@ -1318,13 +1316,11 @@ func TestHandleReceiptFailedTx(t *testing.T) {
 		"transactionId": "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2"
   }`)
 
-	em.On("OperationUpdate",
-		e,
-		"ns1:"+operationID.String(),
-		core.OpStatusFailed,
-		"ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2",
-		"",
-		mock.Anything).Return(nil)
+	em.On("OperationUpdate", e, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+		return update.NamespacedOpID == "ns1:"+operationID.String() &&
+			update.Status == core.OpStatusFailed &&
+			update.BlockchainTXID == "ce79343000e851a0c742f63a733ce19a5f8b9ce1c719b6cecd14f01bcf81fff2"
+	})).Return(nil)
 
 	err := json.Unmarshal(data, &reply)
 	assert.NoError(t, err)
diff --git a/internal/contracts/operations_test.go b/internal/contracts/operations_test.go
index ee244ad45..d37932d0b 100644
--- a/internal/contracts/operations_test.go
+++ b/internal/contracts/operations_test.go
@@ -21,7 +21,6 @@ import (
 	"testing"
 
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
-	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/pkg/core"
@@ -115,7 +114,7 @@ func TestOperationUpdateInvokeSucceed(t *testing.T) {
 		ID:   fftypes.NewUUID(),
 		Type: core.OpTypeBlockchainInvoke,
 	}
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusSucceeded,
 	}
 
@@ -137,7 +136,7 @@ func TestOperationUpdateInvokeFail(t *testing.T) {
 		ID:   fftypes.NewUUID(),
 		Type: core.OpTypeBlockchainInvoke,
 	}
-	update := &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		Status: core.OpStatusFailed,
 	}
 
diff --git a/internal/dataexchange/ffdx/ffdx_test.go b/internal/dataexchange/ffdx/ffdx_test.go
index ac7a8189f..6f47c0640 100644
--- a/internal/dataexchange/ffdx/ffdx_test.go
+++ b/internal/dataexchange/ffdx/ffdx_test.go
@@ -31,6 +31,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/wsclient"
 	"github.com/hyperledger/firefly/internal/coreconfig"
+	"github.com/hyperledger/firefly/mocks/coremocks"
 	"github.com/hyperledger/firefly/mocks/dataexchangemocks"
 	"github.com/hyperledger/firefly/mocks/wsmocks"
 	"github.com/hyperledger/firefly/pkg/core"
@@ -122,9 +123,15 @@ func TestInitMissingURL(t *testing.T) {
 	assert.Regexp(t, "FF10138", err)
 }
 
+func opAcker() func(args mock.Arguments) {
+	return func(args mock.Arguments) {
+		args[1].(*core.OperationUpdate).OnComplete()
+	}
+}
+
 func acker() func(args mock.Arguments) {
 	return func(args mock.Arguments) {
-		args[1].(dataexchange.DXEvent).Ack()
+		args[1].(*dxEvent).Ack()
 	}
 }
 
@@ -466,9 +473,6 @@ func TestBadEvents(t *testing.T) {
 	h, toServer, fromServer, _, done := newTestFFDX(t, false)
 	defer done()
 
-	mcb := &dataexchangemocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
-
 	err := h.Start()
 	assert.NoError(t, err)
 
@@ -490,7 +494,10 @@ func TestBadEvents(t *testing.T) {
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"3"}`, string(msg))
 
-	mcb.AssertExpectations(t)
+	fromServer <- `{"id":"4","type":"message-received","sender":"peer1","message":"{\"batch\":{\"namespace\":\"ns1\"}}"}`
+	msg = <-toServer
+	assert.Equal(t, `{"action":"ack","id":"4"}`, string(msg))
+
 }
 
 func TestMessageEvents(t *testing.T) {
@@ -500,48 +507,43 @@ func TestMessageEvents(t *testing.T) {
 
 	mcb := &dataexchangemocks.Callbacks{}
 	h.SetHandler("ns1", mcb)
+	ocb := &coremocks.OperationCallbacks{}
+	h.SetOperationHandler("ns1", ocb)
 
 	err := h.Start()
 	assert.NoError(t, err)
 
 	namespacedID1 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.EventID() == "1" &&
-			ev.NamespacedID() == namespacedID1 &&
-			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID1 &&
-			ev.TransferResult().Status == core.OpStatusFailed &&
-			ev.TransferResult().Error == "pop"
-	})).Run(acker()).Return(nil)
+	ocb.On("OperationUpdate", h, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+		return ev.NamespacedOpID == namespacedID1 &&
+			ev.Status == core.OpStatusFailed &&
+			ev.ErrorMessage == "pop"
+	})).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())
-	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.EventID() == "2" &&
-			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID2 &&
-			ev.TransferResult().Status == core.OpStatusSucceeded
-	})).Run(acker()).Return(nil)
+	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+		return ev.NamespacedOpID == namespacedID2 &&
+			ev.Status == core.OpStatusSucceeded
+	})).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())
-	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.EventID() == "3" &&
-			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID3 &&
-			ev.TransferResult().Status == core.OpStatusSucceeded &&
-			ev.TransferResult().Manifest == `{"manifest":true}` &&
-			ev.TransferResult().Info.String() == `{"signatures":"and stuff"}`
-	})).Run(acker()).Return(nil)
+	ocb.On("OperationUpdate", mock.Anything, 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"}`
+	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"3","type":"message-acknowledged","requestID":"` + namespacedID3 + `","info":{"signatures":"and stuff"},"manifest":"{\"manifest\":true}"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"3"}`, string(msg))
 
-	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", h, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "4" &&
 			ev.Type() == dataexchange.DXEventTypeMessageReceived &&
 			ev.MessageReceived().PeerID == "peer1"
@@ -551,6 +553,7 @@ func TestMessageEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"4","manifest":"{\"manifest\":true}"}`, string(msg))
 
 	mcb.AssertExpectations(t)
+	ocb.AssertExpectations(t)
 }
 
 func TestBlobEvents(t *testing.T) {
@@ -560,37 +563,35 @@ func TestBlobEvents(t *testing.T) {
 
 	mcb := &dataexchangemocks.Callbacks{}
 	h.SetHandler("ns1", mcb)
+	ocb := &coremocks.OperationCallbacks{}
+	h.SetOperationHandler("ns1", ocb)
 
 	err := h.Start()
 	assert.NoError(t, err)
 
 	namespacedID5 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.EventID() == "5" &&
-			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID5 &&
-			ev.TransferResult().Status == core.OpStatusFailed &&
-			ev.TransferResult().Error == "pop"
-	})).Run(acker()).Return(nil)
+	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+		return ev.NamespacedOpID == namespacedID5 &&
+			ev.Status == core.OpStatusFailed &&
+			ev.ErrorMessage == "pop"
+	})).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())
-	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.EventID() == "6" &&
-			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID6 &&
-			ev.TransferResult().Status == core.OpStatusSucceeded &&
-			ev.TransferResult().Info.String() == `{"some":"details"}`
-	})).Run(acker()).Return(nil)
+	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+		return ev.NamespacedOpID == namespacedID6 &&
+			ev.Status == core.OpStatusSucceeded &&
+			ev.Output.String() == `{"some":"details"}`
+	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"6","type":"blob-delivered","requestID":"` + namespacedID6 + `","info":{"some":"details"}}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"6"}`, string(msg))
 
 	u := fftypes.NewUUID()
 	hash := fftypes.NewRandB32()
-	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
+	mcb.On("DXEvent", h, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "9" &&
 			ev.Type() == dataexchange.DXEventTypePrivateBlobReceived &&
 			ev.PrivateBlobReceived().Hash.Equals(hash)
@@ -600,18 +601,17 @@ func TestBlobEvents(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"9"}`, string(msg))
 
 	namespacedID10 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.EventID() == "10" &&
-			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID10 &&
-			ev.TransferResult().Status == core.OpStatusSucceeded &&
-			ev.TransferResult().Info.String() == `{"signatures":"and stuff"}`
-	})).Run(acker()).Return(nil)
+	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+		return ev.NamespacedOpID == namespacedID10 &&
+			ev.Status == core.OpStatusSucceeded &&
+			ev.Output.String() == `{"signatures":"and stuff"}`
+	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"10","type":"blob-acknowledged","requestID":"` + namespacedID10 + `","info":{"signatures":"and stuff"},"manifest":"{\"manifest\":true}"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"10"}`, string(msg))
 
 	mcb.AssertExpectations(t)
+	ocb.AssertExpectations(t)
 }
 
 func TestEventsWithManifest(t *testing.T) {
@@ -629,28 +629,29 @@ func TestEventsWithManifest(t *testing.T) {
 
 	mcb := &dataexchangemocks.Callbacks{}
 	h.SetHandler("ns1", mcb)
+	ocb := &coremocks.OperationCallbacks{}
+	h.SetOperationHandler("ns1", ocb)
 
 	namespacedID1 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
-	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.EventID() == "1" &&
-			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().Status == core.OpStatusPending
-	})).Run(acker()).Return(nil)
+	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+		return ev.NamespacedOpID == namespacedID1 &&
+			ev.Status == core.OpStatusPending
+	})).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())
-	mcb.On("DXEvent", mock.Anything, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
-		return ev.EventID() == "2" &&
-			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().Status == core.OpStatusPending
-	})).Run(acker()).Return(nil)
+	ocb.On("OperationUpdate", mock.Anything, mock.MatchedBy(func(ev *core.OperationUpdate) bool {
+		return ev.NamespacedOpID == namespacedID2 &&
+			ev.Status == core.OpStatusPending
+	})).Run(opAcker()).Return(nil)
 	fromServer <- `{"id":"2","type":"blob-delivered","requestID":"` + namespacedID2 + `"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"2"}`, string(msg))
 
 	mcb.AssertExpectations(t)
+	ocb.AssertExpectations(t)
 }
 
 func TestEventLoopReceiveClosed(t *testing.T) {
diff --git a/internal/identity/identitymanager_test.go b/internal/identity/identitymanager_test.go
index a99c911a7..6937f0f03 100644
--- a/internal/identity/identitymanager_test.go
+++ b/internal/identity/identitymanager_test.go
@@ -1278,44 +1278,6 @@ func TestValidateParentTypeInvalidType(t *testing.T) {
 
 }
 
-func TestCachedVerifierLookupCaching(t *testing.T) {
-
-	ctx, im := newTestIdentityManager(t)
-
-	verifier := (&core.Verifier{
-		Namespace: "ns1",
-		VerifierRef: core.VerifierRef{
-			Value: "peer1",
-			Type:  core.VerifierTypeFFDXPeerID,
-		},
-	}).Seal()
-	mdi := im.database.(*databasemocks.Plugin)
-	mdi.On("GetVerifierByValue", ctx, verifier.Type, verifier.Namespace, verifier.Value).Return(verifier, nil).Once()
-
-	v1, err := im.CachedVerifierLookup(ctx, core.VerifierTypeFFDXPeerID, "peer1")
-	assert.NoError(t, err)
-	assert.Equal(t, verifier, v1)
-
-	v2, err := im.CachedVerifierLookup(ctx, core.VerifierTypeFFDXPeerID, "peer1")
-	assert.NoError(t, err)
-	assert.Equal(t, verifier, v2)
-
-	mdi.AssertExpectations(t)
-}
-
-func TestCachedVerifierLookupError(t *testing.T) {
-
-	ctx, im := newTestIdentityManager(t)
-
-	mdi := im.database.(*databasemocks.Plugin)
-	mdi.On("GetVerifierByValue", ctx, core.VerifierTypeFFDXPeerID, "ns1", "peer1").Return(nil, fmt.Errorf("pop"))
-
-	_, err := im.CachedVerifierLookup(ctx, core.VerifierTypeFFDXPeerID, "peer1")
-	assert.Regexp(t, "pop", err)
-
-	mdi.AssertExpectations(t)
-}
-
 func TestResolveIdentitySignerOk(t *testing.T) {
 	ctx, im := newTestIdentityManager(t)
 	mdi := im.database.(*databasemocks.Plugin)
@@ -1395,6 +1357,33 @@ func TestResolveIdentitySignerNotFound(t *testing.T) {
 	mdi.AssertExpectations(t)
 }
 
+func TestGetLocalNode(t *testing.T) {
+	ctx, im := newTestIdentityManager(t)
+	mmp := im.multiparty.(*multipartymocks.Manager)
+	mdi := im.database.(*databasemocks.Plugin)
+
+	node := &core.Identity{
+		IdentityBase: core.IdentityBase{
+			ID: fftypes.NewUUID(),
+		},
+	}
+
+	mmp.On("LocalNode").Return(multiparty.LocalNode{Name: "node1"}).Once()
+	mdi.On("GetIdentityByName", ctx, core.IdentityTypeNode, "ns1", "node1").Return(node, nil).Once()
+
+	result, err := im.GetLocalNode(ctx)
+	assert.NoError(t, err)
+	assert.Equal(t, node, result)
+
+	// second call is cached
+	result, err = im.GetLocalNode(ctx)
+	assert.NoError(t, err)
+	assert.Equal(t, node, result)
+
+	mmp.AssertExpectations(t)
+	mdi.AssertExpectations(t)
+}
+
 func TestParseKeyNormalizationConfig(t *testing.T) {
 	assert.Equal(t, KeyNormalizationBlockchainPlugin, ParseKeyNormalizationConfig("blockchain_Plugin"))
 	assert.Equal(t, KeyNormalizationNone, ParseKeyNormalizationConfig("none"))
diff --git a/internal/multiparty/manager_test.go b/internal/multiparty/manager_test.go
index 8a10f90bd..909e19278 100644
--- a/internal/multiparty/manager_test.go
+++ b/internal/multiparty/manager_test.go
@@ -81,7 +81,11 @@ func TestNewMultipartyManager(t *testing.T) {
 	mom := &operationmocks.Manager{}
 	mmi := &metricsmocks.Manager{}
 	mth := &txcommonmocks.Helper{}
-	config := Config{Contracts: []Contract{}}
+	config := Config{
+		Org:       RootOrg{Name: "org1"},
+		Node:      LocalNode{Name: "node1"},
+		Contracts: []Contract{},
+	}
 	mom.On("RegisterHandler", mock.Anything, mock.Anything, []core.OpType{
 		core.OpTypeBlockchainPinBatch,
 		core.OpTypeBlockchainNetworkAction,
@@ -92,6 +96,7 @@ func TestNewMultipartyManager(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, "MultipartyManager", nm.Name())
 	assert.Equal(t, config.Org, nm.RootOrg())
+	assert.Equal(t, config.Node, nm.LocalNode())
 }
 
 func TestInitFail(t *testing.T) {
diff --git a/internal/networkmap/register_node_test.go b/internal/networkmap/register_node_test.go
index 092ee0f49..b75e37335 100644
--- a/internal/networkmap/register_node_test.go
+++ b/internal/networkmap/register_node_test.go
@@ -20,12 +20,12 @@ import (
 	"fmt"
 	"testing"
 
-	"github.com/hyperledger/firefly-common/pkg/config"
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
-	"github.com/hyperledger/firefly/internal/coreconfig"
+	"github.com/hyperledger/firefly/internal/multiparty"
 	"github.com/hyperledger/firefly/mocks/dataexchangemocks"
 	"github.com/hyperledger/firefly/mocks/definitionsmocks"
 	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
+	"github.com/hyperledger/firefly/mocks/multipartymocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -36,16 +36,12 @@ func TestRegisterNodeOk(t *testing.T) {
 	nm, cancel := newTestNetworkmap(t)
 	defer cancel()
 
-	config.Set(coreconfig.OrgKey, "0x23456")
-	config.Set(coreconfig.OrgName, "org1")
-	config.Set(coreconfig.NodeDescription, "Node 1")
-
 	parentOrg := testOrg("org1")
+	signerRef := &core.SignerRef{Key: "0x23456"}
 
 	mim := nm.identity.(*identitymanagermocks.Manager)
 	mim.On("GetMultipartyRootOrg", nm.ctx).Return(parentOrg, nil)
 	mim.On("VerifyIdentityChain", nm.ctx, mock.AnythingOfType("*core.Identity")).Return(parentOrg, false, nil)
-	signerRef := &core.SignerRef{Key: "0x23456"}
 	mim.On("ResolveIdentitySigner", nm.ctx, parentOrg).Return(signerRef, nil)
 
 	mdx := nm.exchange.(*dataexchangemocks.Plugin)
@@ -61,6 +57,9 @@ func TestRegisterNodeOk(t *testing.T) {
 		(*core.SignerRef)(nil),
 		false).Return(nil)
 
+	mmp := nm.multiparty.(*multipartymocks.Manager)
+	mmp.On("LocalNode").Return(multiparty.LocalNode{Name: "node1"})
+
 	node, err := nm.RegisterNode(nm.ctx, false)
 	assert.NoError(t, err)
 	assert.NotNil(t, node)
@@ -68,31 +67,51 @@ func TestRegisterNodeOk(t *testing.T) {
 	mim.AssertExpectations(t)
 	mdx.AssertExpectations(t)
 	mds.AssertExpectations(t)
+	mmp.AssertExpectations(t)
 }
 
-func TestRegisterNodePeerInfoFail(t *testing.T) {
+func TestRegisterNodeMissingName(t *testing.T) {
 
 	nm, cancel := newTestNetworkmap(t)
 	defer cancel()
 
-	config.Set(coreconfig.OrgKey, "0x23456")
-	config.Set(coreconfig.OrgName, "org1")
-	config.Set(coreconfig.NodeDescription, "Node 1")
+	parentOrg := testOrg("org1")
+
+	mim := nm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetMultipartyRootOrg", nm.ctx).Return(parentOrg, nil)
+
+	mmp := nm.multiparty.(*multipartymocks.Manager)
+	mmp.On("LocalNode").Return(multiparty.LocalNode{})
+
+	_, err := nm.RegisterNode(nm.ctx, false)
+	assert.Regexp(t, "FF10216", err)
+
+	mim.AssertExpectations(t)
+	mmp.AssertExpectations(t)
+}
+
+func TestRegisterNodePeerInfoFail(t *testing.T) {
+
+	nm, cancel := newTestNetworkmap(t)
+	defer cancel()
 
 	parentOrg := testOrg("org1")
 
 	mim := nm.identity.(*identitymanagermocks.Manager)
 	mim.On("GetMultipartyRootOrg", nm.ctx).Return(parentOrg, nil)
-	mim.On("VerifyIdentityChain", nm.ctx, mock.AnythingOfType("*core.Identity")).Return(parentOrg, false, nil)
-	signerRef := &core.SignerRef{Key: "0x23456"}
-	mim.On("ResolveIdentitySigner", nm.ctx, parentOrg).Return(signerRef, nil)
 
 	mdx := nm.exchange.(*dataexchangemocks.Plugin)
 	mdx.On("GetEndpointInfo", nm.ctx).Return(fftypes.JSONObject{}, fmt.Errorf("pop"))
 
+	mmp := nm.multiparty.(*multipartymocks.Manager)
+	mmp.On("LocalNode").Return(multiparty.LocalNode{Name: "node1"})
+
 	_, err := nm.RegisterNode(nm.ctx, false)
 	assert.Regexp(t, "pop", err)
 
+	mim.AssertExpectations(t)
+	mdx.AssertExpectations(t)
+	mmp.AssertExpectations(t)
 }
 
 func TestRegisterNodeGetOwnerFail(t *testing.T) {
diff --git a/internal/operations/manager_test.go b/internal/operations/manager_test.go
index 50072a95f..95548ebc0 100644
--- a/internal/operations/manager_test.go
+++ b/internal/operations/manager_test.go
@@ -52,7 +52,7 @@ func (m *mockHandler) RunOperation(ctx context.Context, op *core.PreparedOperati
 	return m.Outputs, m.Complete, m.RunErr
 }
 
-func (m *mockHandler) OnOperationUpdate(ctx context.Context, op *core.Operation, update *OperationUpdate) error {
+func (m *mockHandler) OnOperationUpdate(ctx context.Context, op *core.Operation, update *core.OperationUpdate) error {
 	return m.UpdateErr
 }
 
diff --git a/internal/operations/operation_updater_test.go b/internal/operations/operation_updater_test.go
index e9befebc5..a271ecb8e 100644
--- a/internal/operations/operation_updater_test.go
+++ b/internal/operations/operation_updater_test.go
@@ -75,11 +75,11 @@ func TestNewOperationUpdaterNoConcurrency(t *testing.T) {
 func TestSubmitUpdateBadIDIgnored(t *testing.T) {
 	ou := newTestOperationUpdater(t)
 	ou.close()
-	ou.workQueues = []chan *OperationUpdate{
-		make(chan *OperationUpdate),
+	ou.workQueues = []chan *core.OperationUpdate{
+		make(chan *core.OperationUpdate),
 	}
 	ou.cancelFunc()
-	ou.SubmitOperationUpdate(ou.ctx, &OperationUpdate{
+	ou.SubmitOperationUpdate(ou.ctx, &core.OperationUpdate{
 		NamespacedOpID: "!!!" + fftypes.NewUUID().String(),
 	})
 }
@@ -87,11 +87,11 @@ func TestSubmitUpdateBadIDIgnored(t *testing.T) {
 func TestSubmitUpdateClosed(t *testing.T) {
 	ou := newTestOperationUpdater(t)
 	ou.close()
-	ou.workQueues = []chan *OperationUpdate{
-		make(chan *OperationUpdate),
+	ou.workQueues = []chan *core.OperationUpdate{
+		make(chan *core.OperationUpdate),
 	}
 	ou.cancelFunc()
-	ou.SubmitOperationUpdate(ou.ctx, &OperationUpdate{
+	ou.SubmitOperationUpdate(ou.ctx, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + fftypes.NewUUID().String(),
 	})
 }
@@ -109,7 +109,7 @@ func TestSubmitUpdateSyncFallbackOpNotFound(t *testing.T) {
 	mdi.On("GetOperations", customCtx, mock.Anything, mock.Anything).Return(nil, nil, nil)
 
 	complete := false
-	ou.SubmitOperationUpdate(customCtx, &OperationUpdate{
+	ou.SubmitOperationUpdate(customCtx, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + fftypes.NewUUID().String(),
 		OnComplete:     func() { complete = true },
 	})
@@ -128,7 +128,7 @@ func TestSubmitUpdateDatabaseError(t *testing.T) {
 	mdi := ou.database.(*databasemocks.Plugin)
 	mdi.On("RunAsGroup", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 
-	ou.SubmitOperationUpdate(ctx, &OperationUpdate{
+	ou.SubmitOperationUpdate(ctx, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + fftypes.NewUUID().String(),
 	})
 
@@ -140,7 +140,7 @@ func TestSubmitUpdateWrongNS(t *testing.T) {
 	defer ou.close()
 	customCtx := context.WithValue(context.Background(), "dbtx", "on this context")
 
-	ou.SubmitOperationUpdate(customCtx, &OperationUpdate{
+	ou.SubmitOperationUpdate(customCtx, &core.OperationUpdate{
 		NamespacedOpID: "ns2:" + fftypes.NewUUID().String(),
 	})
 }
@@ -177,18 +177,18 @@ func TestSubmitUpdateWorkerE2ESuccess(t *testing.T) {
 
 	om.Start()
 
-	om.SubmitOperationUpdate(&mockplug{}, &OperationUpdate{
+	om.SubmitOperationUpdate(&mockplug{}, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID1.String(),
 		Status:         core.OpStatusSucceeded,
 		BlockchainTXID: "tx12345",
 	})
-	om.SubmitOperationUpdate(&mockplug{}, &OperationUpdate{
+	om.SubmitOperationUpdate(&mockplug{}, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID2.String(),
 		Status:         core.OpStatusFailed,
 		ErrorMessage:   "err1",
 		Output:         fftypes.JSONObject{"test": true},
 	})
-	om.SubmitOperationUpdate(&mockplug{}, &OperationUpdate{
+	om.SubmitOperationUpdate(&mockplug{}, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID3.String(),
 		Status:         core.OpStatusFailed,
 		ErrorMessage:   "err2",
@@ -210,7 +210,7 @@ func TestUpdateLoopExitRetryCancelledContext(t *testing.T) {
 		ou.cancelFunc()
 	})
 
-	ou.SubmitOperationUpdate(ou.ctx, &OperationUpdate{
+	ou.SubmitOperationUpdate(ou.ctx, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + fftypes.NewUUID().String(),
 	})
 
@@ -225,7 +225,7 @@ func TestDoBatchUpdateIgnoreBadID(t *testing.T) {
 
 	ou.initQueues()
 
-	err := ou.doBatchUpdate(ou.ctx, []*OperationUpdate{
+	err := ou.doBatchUpdate(ou.ctx, []*core.OperationUpdate{
 		{NamespacedOpID: "!!Bad", Status: core.OpStatusSucceeded},
 	})
 	assert.NoError(t, err)
@@ -245,7 +245,7 @@ func TestDoBatchUpdateFailUpdate(t *testing.T) {
 
 	ou.initQueues()
 
-	err := ou.doBatchUpdate(ou.ctx, []*OperationUpdate{
+	err := ou.doBatchUpdate(ou.ctx, []*core.OperationUpdate{
 		{NamespacedOpID: "ns1:" + opID1.String(), Status: core.OpStatusSucceeded},
 	})
 	assert.Regexp(t, "pop", err)
@@ -266,7 +266,7 @@ func TestDoBatchUpdateFailGetTransactions(t *testing.T) {
 
 	ou.initQueues()
 
-	err := ou.doBatchUpdate(ou.ctx, []*OperationUpdate{
+	err := ou.doBatchUpdate(ou.ctx, []*core.OperationUpdate{
 		{NamespacedOpID: "ns1:" + opID1.String(), Status: core.OpStatusSucceeded},
 	})
 	assert.Regexp(t, "pop", err)
@@ -284,7 +284,7 @@ func TestDoBatchUpdateFailGetOperations(t *testing.T) {
 
 	ou.initQueues()
 
-	err := ou.doBatchUpdate(ou.ctx, []*OperationUpdate{
+	err := ou.doBatchUpdate(ou.ctx, []*core.OperationUpdate{
 		{NamespacedOpID: "ns1:" + opID1.String(), Status: core.OpStatusSucceeded},
 	})
 	assert.Regexp(t, "pop", err)
@@ -298,7 +298,7 @@ func TestDoUpdateIgnoreBadID(t *testing.T) {
 
 	ou.initQueues()
 
-	err := ou.doUpdate(ou.ctx, &OperationUpdate{
+	err := ou.doUpdate(ou.ctx, &core.OperationUpdate{
 		NamespacedOpID: "!!!bad", Status: core.OpStatusSucceeded, BlockchainTXID: "0x12345",
 	}, []*core.Operation{}, []*core.Transaction{})
 	assert.NoError(t, err)
@@ -316,7 +316,7 @@ func TestDoUpdateFailTransactionUpdate(t *testing.T) {
 
 	ou.initQueues()
 
-	err := ou.doUpdate(ou.ctx, &OperationUpdate{
+	err := ou.doUpdate(ou.ctx, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID1.String(), Status: core.OpStatusSucceeded, BlockchainTXID: "0x12345",
 	}, []*core.Operation{
 		{Namespace: "ns1", ID: opID1, Type: core.OpTypeBlockchainInvoke, Transaction: txID1},
@@ -338,7 +338,7 @@ func TestDoUpdateFailExternalHandler(t *testing.T) {
 
 	ou.initQueues()
 
-	err := ou.doUpdate(ou.ctx, &OperationUpdate{
+	err := ou.doUpdate(ou.ctx, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID1.String(), Status: core.OpStatusSucceeded,
 	}, []*core.Operation{
 		{Namespace: "ns1", ID: opID1, Type: core.OpTypeBlockchainInvoke, Transaction: txID1},
@@ -363,7 +363,7 @@ func TestDoUpdateVerifyBatchManifest(t *testing.T) {
 	}, nil)
 	mdi.On("ResolveOperation", mock.Anything, "ns1", opID1, core.OpStatusSucceeded, mock.Anything, fftypes.JSONObject(nil)).Return(nil)
 
-	err := ou.doUpdate(ou.ctx, &OperationUpdate{
+	err := ou.doUpdate(ou.ctx, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID1.String(),
 		Status:         core.OpStatusSucceeded,
 		VerifyManifest: true,
@@ -397,7 +397,7 @@ func TestDoUpdateVerifyBatchManifestQuery(t *testing.T) {
 	mdi := ou.database.(*databasemocks.Plugin)
 	mdi.On("GetBatchByID", mock.Anything, "ns1", batchID).Return(nil, fmt.Errorf("pop"))
 
-	err := ou.doUpdate(ou.ctx, &OperationUpdate{
+	err := ou.doUpdate(ou.ctx, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID1.String(),
 		Status:         core.OpStatusSucceeded,
 		VerifyManifest: true,
@@ -434,7 +434,7 @@ func TestDoUpdateVerifyBatchManifestFail(t *testing.T) {
 	}, nil)
 	mdi.On("ResolveOperation", mock.Anything, "ns1", opID1, core.OpStatusFailed, mock.Anything, fftypes.JSONObject(nil)).Return(nil)
 
-	err := ou.doUpdate(ou.ctx, &OperationUpdate{
+	err := ou.doUpdate(ou.ctx, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID1.String(),
 		Status:         core.OpStatusSucceeded,
 		VerifyManifest: true,
@@ -468,7 +468,7 @@ func TestDoUpdateVerifyBlobManifestFail(t *testing.T) {
 	mdi := ou.database.(*databasemocks.Plugin)
 	mdi.On("ResolveOperation", mock.Anything, "ns1", opID1, core.OpStatusFailed, mock.Anything, fftypes.JSONObject(nil)).Return(nil)
 
-	err := ou.doUpdate(ou.ctx, &OperationUpdate{
+	err := ou.doUpdate(ou.ctx, &core.OperationUpdate{
 		NamespacedOpID: "ns1:" + opID1.String(),
 		Status:         core.OpStatusSucceeded,
 		VerifyManifest: true,
diff --git a/internal/orchestrator/bound_callbacks_test.go b/internal/orchestrator/bound_callbacks_test.go
index 6079b6ea9..2387a893a 100644
--- a/internal/orchestrator/bound_callbacks_test.go
+++ b/internal/orchestrator/bound_callbacks_test.go
@@ -17,19 +17,15 @@
 package orchestrator
 
 import (
-	"context"
 	"fmt"
 	"testing"
 
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
-	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
-	"github.com/hyperledger/firefly/mocks/dataexchangemocks"
 	"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/hyperledger/firefly/pkg/dataexchange"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 )
@@ -37,24 +33,24 @@ import (
 func TestBoundCallbacks(t *testing.T) {
 	mei := &eventmocks.EventManager{}
 	mbi := &blockchainmocks.Plugin{}
-	mdx := &dataexchangemocks.Plugin{}
 	mss := &sharedstoragemocks.Plugin{}
 	mom := &operationmocks.Manager{}
-	bc := boundCallbacks{dx: mdx, ei: mei, ss: mss, om: mom}
+	bc := boundCallbacks{ei: mei, ss: mss, om: mom}
 
 	info := fftypes.JSONObject{"hello": "world"}
 	hash := fftypes.NewRandB32()
 	opID := fftypes.NewUUID()
 	nsOpID := "ns1:" + opID.String()
 
-	mom.On("SubmitOperationUpdate", mock.Anything, &operations.OperationUpdate{
+	update := &core.OperationUpdate{
 		NamespacedOpID: nsOpID,
 		Status:         core.OpStatusFailed,
 		BlockchainTXID: "0xffffeeee",
 		ErrorMessage:   "error info",
 		Output:         info,
-	}).Return().Once()
-	bc.OperationUpdate(mbi, nsOpID, core.OpStatusFailed, "0xffffeeee", "error info", info)
+	}
+	mom.On("SubmitOperationUpdate", mock.Anything, update).Return().Once()
+	bc.OperationUpdate(mbi, update)
 
 	mei.On("SharedStorageBatchDownloaded", mss, "payload1", []byte(`{}`)).Return(nil, fmt.Errorf("pop"))
 	_, err := bc.SharedStorageBatchDownloaded("payload1", []byte(`{}`))
@@ -65,87 +61,6 @@ func TestBoundCallbacks(t *testing.T) {
 
 	mei.AssertExpectations(t)
 	mbi.AssertExpectations(t)
-	mdx.AssertExpectations(t)
-	mss.AssertExpectations(t)
-	mom.AssertExpectations(t)
-}
-
-func TestBoundCallbacksDXEvent(t *testing.T) {
-	mei := &eventmocks.EventManager{}
-	mdx := &dataexchangemocks.Plugin{}
-	mss := &sharedstoragemocks.Plugin{}
-	mom := &operationmocks.Manager{}
-	bc := boundCallbacks{dx: mdx, ei: mei, ss: mss, om: mom}
-
-	ctx := context.Background()
-	info := fftypes.JSONObject{"hello": "world"}
-	opID := fftypes.NewUUID()
-	nsOpID := "ns1:" + opID.String()
-
-	mdx.On("Capabilities").Return(&dataexchange.Capabilities{
-		Manifest: true,
-	})
-
-	event1 := &dataexchangemocks.DXEvent{}
-	mei.On("DXEvent", mdx, event1).Return().Once()
-	event1.On("Type").Return(dataexchange.DXEventTypeMessageReceived).Once()
-	bc.DXEvent(ctx, event1)
-	event1.AssertExpectations(t)
-
-	event2 := &dataexchangemocks.DXEvent{}
-	event2.On("Type").Return(dataexchange.DXEventTypeTransferResult).Once()
-	event2.On("TransferResult").Return(&dataexchange.TransferResult{
-		TrackingID: opID.String(),
-		Status:     core.OpStatusSucceeded,
-		TransportStatusUpdate: core.TransportStatusUpdate{
-			Info:     info,
-			Manifest: "Sally",
-		},
-	})
-	event2.On("NamespacedID").Return(nsOpID)
-	event2.On("Ack").Return()
-	mom.On("SubmitOperationUpdate", mock.Anything, mock.MatchedBy(func(update *operations.OperationUpdate) bool {
-		if update.NamespacedOpID == nsOpID &&
-			update.Status == core.OpStatusSucceeded &&
-			update.VerifyManifest &&
-			update.DXManifest == "Sally" &&
-			update.DXHash == "" {
-			update.OnComplete()
-			return true
-		}
-		return false
-	})).Return().Once()
-	bc.DXEvent(ctx, event2)
-	event2.AssertExpectations(t)
-
-	event3 := &dataexchangemocks.DXEvent{}
-	event3.On("Type").Return(dataexchange.DXEventTypeTransferResult).Once()
-	event3.On("TransferResult").Return(&dataexchange.TransferResult{
-		TrackingID: opID.String(),
-		Status:     core.OpStatusSucceeded,
-		TransportStatusUpdate: core.TransportStatusUpdate{
-			Info: info,
-			Hash: "hash1",
-		},
-	})
-	event3.On("NamespacedID").Return(nsOpID)
-	event3.On("Ack").Return()
-	mom.On("SubmitOperationUpdate", mock.Anything, mock.MatchedBy(func(update *operations.OperationUpdate) bool {
-		if update.NamespacedOpID == nsOpID &&
-			update.Status == core.OpStatusSucceeded &&
-			update.VerifyManifest &&
-			update.DXManifest == "" &&
-			update.DXHash == "hash1" {
-			update.OnComplete()
-			return true
-		}
-		return false
-	})).Return().Once()
-	bc.DXEvent(ctx, event3)
-	event3.AssertExpectations(t)
-
-	mei.AssertExpectations(t)
-	mdx.AssertExpectations(t)
 	mss.AssertExpectations(t)
 	mom.AssertExpectations(t)
 }
diff --git a/internal/orchestrator/orchestrator_test.go b/internal/orchestrator/orchestrator_test.go
index 6474dabe2..23692daab 100644
--- a/internal/orchestrator/orchestrator_test.go
+++ b/internal/orchestrator/orchestrator_test.go
@@ -209,10 +209,11 @@ func TestInitOK(t *testing.T) {
 	or.mbi.On("SetOperationHandler", "ns", mock.Anything).Return()
 	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return([]*core.Identity{{}}, nil, nil)
 	or.mdx.On("SetHandler", "ns2", mock.Anything).Return()
+	or.mdx.On("SetOperationHandler", "ns", mock.Anything).Return()
 	or.mdx.On("SetNodes", mock.Anything).Return()
 	or.mps.On("SetHandler", "ns", mock.Anything).Return()
 	or.mti.On("SetHandler", "ns", mock.Anything).Return(nil)
-	or.mti.On("SetOperationHandler", "ns", mock.Anything).Return(nil)
+	or.mti.On("SetOperationHandler", "ns", mock.Anything).Return()
 	or.mmp.On("ConfigureContract", mock.Anything, mock.Anything).Return(nil)
 	err := or.Init(or.ctx, or.cancelCtx)
 	assert.NoError(t, err)
diff --git a/internal/orchestrator/status_test.go b/internal/orchestrator/status_test.go
index 79003917a..da5848339 100644
--- a/internal/orchestrator/status_test.go
+++ b/internal/orchestrator/status_test.go
@@ -81,7 +81,6 @@ func TestGetStatusRegistered(t *testing.T) {
 
 	coreconfig.Reset()
 	config.Set(coreconfig.NamespacesDefault, "default")
-	config.Set(coreconfig.NodeName, "node1")
 
 	orgID := fftypes.NewUUID()
 	nodeID := fftypes.NewUUID()
@@ -109,6 +108,7 @@ func TestGetStatusRegistered(t *testing.T) {
 	}, nil, nil)
 
 	or.config.Multiparty.Org.Name = "org1"
+	or.config.Multiparty.Node.Name = "node1"
 
 	or.mem.On("GetPlugins").Return(mockEventPlugins)
 
@@ -146,7 +146,6 @@ func TestGetStatusVerifierLookupFail(t *testing.T) {
 
 	coreconfig.Reset()
 	config.Set(coreconfig.NamespacesDefault, "default")
-	config.Set(coreconfig.NodeName, "node1")
 
 	orgID := fftypes.NewUUID()
 
@@ -173,7 +172,6 @@ func TestGetStatusWrongNodeOwner(t *testing.T) {
 
 	coreconfig.Reset()
 	config.Set(coreconfig.NamespacesDefault, "default")
-	config.Set(coreconfig.NodeName, "node1")
 
 	orgID := fftypes.NewUUID()
 	nodeID := fftypes.NewUUID()
@@ -201,6 +199,7 @@ func TestGetStatusWrongNodeOwner(t *testing.T) {
 	}, nil, nil)
 
 	or.config.Multiparty.Org.Name = "org1"
+	or.config.Multiparty.Node.Name = "node1"
 
 	or.mem.On("GetPlugins").Return(mockEventPlugins)
 
@@ -226,11 +225,11 @@ func TestGetStatusUnregistered(t *testing.T) {
 
 	coreconfig.Reset()
 	config.Set(coreconfig.NamespacesDefault, "default")
-	config.Set(coreconfig.NodeName, "node1")
 
 	or.mim.On("GetMultipartyRootOrg", or.ctx).Return(nil, fmt.Errorf("pop"))
 
 	or.config.Multiparty.Org.Name = "org1"
+	or.config.Multiparty.Node.Name = "node1"
 
 	or.mem.On("GetPlugins").Return(mockEventPlugins)
 
@@ -255,7 +254,6 @@ func TestGetStatusOrgOnlyRegistered(t *testing.T) {
 
 	coreconfig.Reset()
 	config.Set(coreconfig.NamespacesDefault, "default")
-	config.Set(coreconfig.NodeName, "node1")
 
 	orgID := fftypes.NewUUID()
 
@@ -276,6 +274,7 @@ func TestGetStatusOrgOnlyRegistered(t *testing.T) {
 	}, nil, nil)
 
 	or.config.Multiparty.Org.Name = "org1"
+	or.config.Multiparty.Node.Name = "node1"
 
 	or.mem.On("GetPlugins").Return(mockEventPlugins)
 
@@ -309,7 +308,6 @@ func TestGetStatusNodeError(t *testing.T) {
 
 	coreconfig.Reset()
 	config.Set(coreconfig.NamespacesDefault, "default")
-	config.Set(coreconfig.NodeName, "node1")
 
 	orgID := fftypes.NewUUID()
 
@@ -329,6 +327,8 @@ func TestGetStatusNodeError(t *testing.T) {
 		}},
 	}, nil, nil)
 
+	or.config.Multiparty.Node.Name = "node1"
+
 	or.mem.On("GetPlugins").Return(mockEventPlugins)
 
 	_, err := or.GetStatus(or.ctx)
diff --git a/internal/privatemessaging/message_test.go b/internal/privatemessaging/message_test.go
index 612044185..cafe5d2ea 100644
--- a/internal/privatemessaging/message_test.go
+++ b/internal/privatemessaging/message_test.go
@@ -82,6 +82,7 @@ func TestSendConfirmMessageE2EOk(t *testing.T) {
 	localNode := newTestNode("node1", intermediateOrg)
 	mim.On("ResolveInputSigningIdentity", pm.ctx, mock.Anything).Return(nil)
 	mim.On("GetMultipartyRootOrg", pm.ctx).Return(intermediateOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(localNode, nil)
 	mim.On("CachedIdentityLookupMustExist", pm.ctx, "org1").Return(intermediateOrg, false, nil)
 	mim.On("CachedIdentityLookupByID", pm.ctx, rootOrg.ID).Return(rootOrg, nil)
 
@@ -224,6 +225,7 @@ func TestResolveAndSendBadInlineData(t *testing.T) {
 	localNode := newTestNode("node1", localOrg)
 	mim.On("ResolveInputSigningIdentity", pm.ctx, mock.Anything).Return(nil)
 	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(localNode, nil)
 	mim.On("ResolveInputSigningIdentity", pm.ctx, mock.Anything).Run(func(args mock.Arguments) {
 		identity := args[2].(*core.SignerRef)
 		identity.Author = "localorg"
@@ -340,6 +342,7 @@ func TestMessagePrepare(t *testing.T) {
 	localNode := newTestNode("node1", localOrg)
 	mim.On("ResolveInputSigningIdentity", pm.ctx, mock.Anything).Return(nil)
 	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(localNode, nil)
 	mim.On("ResolveInputSigningIdentity", pm.ctx, mock.Anything).Run(func(args mock.Arguments) {
 		identity := args[1].(*core.SignerRef)
 		identity.Author = "localorg"
diff --git a/internal/privatemessaging/privatemessaging_test.go b/internal/privatemessaging/privatemessaging_test.go
index 9d6bfedee..8b1786475 100644
--- a/internal/privatemessaging/privatemessaging_test.go
+++ b/internal/privatemessaging/privatemessaging_test.go
@@ -42,7 +42,6 @@ import (
 
 func newTestPrivateMessagingCommon(t *testing.T, metricsEnabled bool) (*privateMessaging, func()) {
 	coreconfig.Reset()
-	config.Set(coreconfig.NodeName, "node1")
 	config.Set(coreconfig.GroupCacheTTL, "1m")
 	config.Set(coreconfig.GroupCacheSize, "1m")
 
diff --git a/internal/privatemessaging/recipients_test.go b/internal/privatemessaging/recipients_test.go
index 0bb457d39..cbf12685f 100644
--- a/internal/privatemessaging/recipients_test.go
+++ b/internal/privatemessaging/recipients_test.go
@@ -43,13 +43,13 @@ func TestResolveMemberListNewGroupE2E(t *testing.T) {
 	var dataID *fftypes.UUID
 	mdi := pm.database.(*databasemocks.Plugin)
 	mdi.On("GetIdentities", pm.ctx, "ns1", mock.Anything).Return([]*core.Identity{remoteNode}, nil, nil).Once()
-	mdi.On("GetIdentities", pm.ctx, "ns1", mock.Anything).Return([]*core.Identity{localNode}, nil, nil).Once()
 	mdi.On("GetGroupByHash", pm.ctx, "ns1", mock.Anything, mock.Anything).Return(nil, nil).Once()
 	mdi.On("UpsertGroup", pm.ctx, mock.Anything, database.UpsertOptimizationNew).Return(nil)
 
 	mim := pm.identity.(*identitymanagermocks.Manager)
 	mim.On("CachedIdentityLookupMustExist", pm.ctx, "remoteorg").Return(remoteOrg, false, nil)
 	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(localNode, nil)
 	ud := mdi.On("UpsertData", pm.ctx, mock.Anything, database.UpsertOptimizationNew).Return(nil)
 	ud.RunFn = func(a mock.Arguments) {
 		data := a[1].(*core.Data)
@@ -99,7 +99,9 @@ func TestResolveMemberListNewGroupE2E(t *testing.T) {
 		},
 	})
 	assert.NoError(t, err)
+
 	mdi.AssertExpectations(t)
+	mim.AssertExpectations(t)
 
 }
 
@@ -116,7 +118,8 @@ func TestResolveMemberListExistingGroup(t *testing.T) {
 	mdi.On("GetGroupByHash", pm.ctx, "ns1", mock.Anything, mock.Anything).Return(&core.Group{Hash: fftypes.NewRandB32()}, nil, nil).Once()
 	mim := pm.identity.(*identitymanagermocks.Manager)
 	mim.On("CachedIdentityLookupMustExist", pm.ctx, "org1").Return(localOrg, false, nil)
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localNode, nil)
+	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(localNode, nil)
 
 	err := pm.resolveRecipientList(pm.ctx, &core.MessageInOut{
 		Message: core.Message{
@@ -149,7 +152,8 @@ func TestResolveMemberListLookupFail(t *testing.T) {
 
 	mim := pm.identity.(*identitymanagermocks.Manager)
 	mim.On("CachedIdentityLookupMustExist", pm.ctx, "org1").Return(nil, true, fmt.Errorf("pop"))
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localNode, nil)
+	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(localNode, nil)
 
 	err := pm.resolveRecipientList(pm.ctx, &core.MessageInOut{
 		Message: core.Message{
@@ -184,7 +188,8 @@ func TestResolveMemberListGetGroupsFail(t *testing.T) {
 	mdi.On("GetGroupByHash", pm.ctx, "ns1", mock.Anything, mock.Anything).Return(nil, fmt.Errorf("pop"))
 	mim := pm.identity.(*identitymanagermocks.Manager)
 	mim.On("CachedIdentityLookupMustExist", pm.ctx, "org1").Return(localOrg, false, nil)
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localNode, nil)
+	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(localNode, nil)
 
 	err := pm.resolveRecipientList(pm.ctx, &core.MessageInOut{
 		Message: core.Message{
@@ -202,6 +207,7 @@ func TestResolveMemberListGetGroupsFail(t *testing.T) {
 		},
 	})
 	assert.EqualError(t, err, "pop")
+
 	mdi.AssertExpectations(t)
 	mim.AssertExpectations(t)
 
@@ -236,23 +242,16 @@ func TestResolveMemberListLocalOrgUnregistered(t *testing.T) {
 
 }
 
-func TestResolveMemberListMissingLocalMemberLookupFailed(t *testing.T) {
+func TestResolveMemberListLocalMemberLookupFailed(t *testing.T) {
 
 	pm, cancel := newTestPrivateMessaging(t)
 	defer cancel()
 
 	localOrg := newTestOrg("localorg")
-	remoteOrg := newTestOrg("remoteorg")
-	localNode := newTestNode("node1", localOrg)
-	remoteNode := newTestNode("node2", remoteOrg)
-
-	mdi := pm.database.(*databasemocks.Plugin)
-	mdi.On("GetIdentities", pm.ctx, "ns1", mock.Anything).Return([]*core.Identity{remoteNode}, nil, nil).Once()
-	mdi.On("GetIdentities", pm.ctx, "ns1", mock.Anything).Return([]*core.Identity{localNode}, nil, fmt.Errorf("pop")).Once()
 
 	mim := pm.identity.(*identitymanagermocks.Manager)
-	mim.On("CachedIdentityLookupMustExist", pm.ctx, "org1").Return(localOrg, false, nil)
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localNode, nil)
+	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(nil, fmt.Errorf("pop"))
 
 	err := pm.resolveRecipientList(pm.ctx, &core.MessageInOut{
 		Message: core.Message{
@@ -270,7 +269,7 @@ func TestResolveMemberListMissingLocalMemberLookupFailed(t *testing.T) {
 		},
 	})
 	assert.Regexp(t, "pop", err)
-	mdi.AssertExpectations(t)
+
 	mim.AssertExpectations(t)
 
 }
@@ -288,7 +287,8 @@ func TestResolveMemberListNodeNotFound(t *testing.T) {
 
 	mim := pm.identity.(*identitymanagermocks.Manager)
 	mim.On("CachedIdentityLookupMustExist", pm.ctx, "org1").Return(localOrg, false, nil)
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localNode, nil)
+	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(localNode, nil)
 
 	err := pm.resolveRecipientList(pm.ctx, &core.MessageInOut{
 		Message: core.Message{
@@ -327,6 +327,7 @@ func TestResolveMemberNodeOwnedParentOrg(t *testing.T) {
 	mdi.On("GetGroupByHash", pm.ctx, "ns1", mock.Anything, mock.Anything).Return(&core.Group{Hash: fftypes.NewRandB32()}, nil, nil).Once()
 	mim := pm.identity.(*identitymanagermocks.Manager)
 	mim.On("GetMultipartyRootOrg", pm.ctx).Return(parentOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(localNode, nil)
 	mim.On("CachedIdentityLookupMustExist", pm.ctx, "org1").Return(childOrg, false, nil)
 	mim.On("CachedIdentityLookupByID", pm.ctx, parentOrg.ID).Return(parentOrg, nil)
 
@@ -408,36 +409,3 @@ func TestResolveReceipientListEmptyList(t *testing.T) {
 	err := pm.resolveRecipientList(pm.ctx, &core.MessageInOut{})
 	assert.Regexp(t, "FF00115", err)
 }
-
-func TestResolveLocalNodeCached(t *testing.T) {
-	pm, cancel := newTestPrivateMessaging(t)
-	defer cancel()
-
-	pm.localNodeID = fftypes.NewUUID()
-
-	ni, err := pm.resolveLocalNode(pm.ctx, newTestOrg("localorg"))
-	assert.NoError(t, err)
-	assert.Equal(t, pm.localNodeID, ni)
-}
-
-func TestResolveLocalNodeNotFound(t *testing.T) {
-	pm, cancel := newTestPrivateMessaging(t)
-	defer cancel()
-
-	mdi := pm.database.(*databasemocks.Plugin)
-	mdi.On("GetIdentities", pm.ctx, "ns1", mock.Anything).Return([]*core.Identity{}, nil, nil)
-
-	_, err := pm.resolveLocalNode(pm.ctx, newTestOrg("localorg"))
-	assert.Regexp(t, "FF10225", err)
-}
-
-func TestResolveLocalNodeNotError(t *testing.T) {
-	pm, cancel := newTestPrivateMessaging(t)
-	defer cancel()
-
-	mdi := pm.database.(*databasemocks.Plugin)
-	mdi.On("GetIdentities", pm.ctx, "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
-
-	_, err := pm.resolveLocalNode(pm.ctx, newTestOrg("localorg"))
-	assert.EqualError(t, err, "pop")
-}
diff --git a/internal/tokens/fftokens/fftokens_test.go b/internal/tokens/fftokens/fftokens_test.go
index 658c6572a..bb4891f9f 100644
--- a/internal/tokens/fftokens/fftokens_test.go
+++ b/internal/tokens/fftokens/fftokens_test.go
@@ -828,7 +828,9 @@ func TestReceiptEvents(t *testing.T) {
 	opID := fftypes.NewUUID()
 
 	// receipt: bad ID - passed through
-	mcb.On("OperationUpdate", h, "wrong", core.OpStatusFailed, "", "", mock.Anything).Return(nil).Once()
+	mcb.On("OperationUpdate", h, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+		return update.NamespacedOpID == "wrong " && update.Status == core.OpStatusFailed
+	})).Return(nil).Once()
 	fromServer <- fftypes.JSONObject{
 		"id":    "3",
 		"event": "receipt",
@@ -836,7 +838,11 @@ func TestReceiptEvents(t *testing.T) {
 	}.String()
 
 	// receipt: success
-	mcb.On("OperationUpdate", h, "ns1:"+opID.String(), core.OpStatusSucceeded, "0xffffeeee", "", mock.Anything).Return(nil).Once()
+	mcb.On("OperationUpdate", h, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+		return update.NamespacedOpID == "ns1:"+opID.String() &&
+			update.Status == core.OpStatusSucceeded &&
+			update.BlockchainTXID == "0xffffeeee"
+	})).Return(nil).Once()
 	fromServer <- fftypes.JSONObject{
 		"id":    "4",
 		"event": "receipt",
@@ -848,7 +854,11 @@ func TestReceiptEvents(t *testing.T) {
 	}.String()
 
 	// receipt: failure
-	mcb.On("OperationUpdate", h, "ns1:"+opID.String(), core.OpStatusFailed, "0xffffeeee", "", mock.Anything).Return(nil).Once()
+	mcb.On("OperationUpdate", h, mock.MatchedBy(func(update *core.OperationUpdate) bool {
+		return update.NamespacedOpID == "ns1:"+opID.String() &&
+			update.Status == core.OpStatusFailed &&
+			update.BlockchainTXID == "0xffffeeee"
+	})).Return(nil).Once()
 	fromServer <- fftypes.JSONObject{
 		"id":    "5",
 		"event": "receipt",