Repository: hyperledger/firefly
Pull request #926 (2022-08-04 14:56:10)

Destroy sysmessaging package 💥
It was created to stash random interfaces just to avoid circular dependencies - but all of these interfaces can be relocated now without introducing a circle.


Patch:
diff --git a/Makefile b/Makefile
index 03fedaf65..f684c8613 100644
--- a/Makefile
+++ b/Makefile
@@ -53,9 +53,7 @@ $(eval $(call makemock, pkg/tokens,                Plugin,             tokenmock
 $(eval $(call makemock, pkg/tokens,                Callbacks,          tokenmocks))
 $(eval $(call makemock, internal/txcommon,         Helper,             txcommonmocks))
 $(eval $(call makemock, internal/identity,         Manager,            identitymanagermocks))
-$(eval $(call makemock, internal/sysmessaging,     SystemEvents,       sysmessagingmocks))
-$(eval $(call makemock, internal/sysmessaging,     MessageSender,      sysmessagingmocks))
-$(eval $(call makemock, internal/sysmessaging,     LocalNodeInfo,      sysmessagingmocks))
+$(eval $(call makemock, internal/syncasync,        Sender,             syncasyncmocks))
 $(eval $(call makemock, internal/syncasync,        Bridge,             syncasyncmocks))
 $(eval $(call makemock, internal/data,             Manager,            datamocks))
 $(eval $(call makemock, internal/batch,            Manager,            batchmocks))
@@ -66,6 +64,7 @@ $(eval $(call makemock, internal/shareddownload,   Callbacks,          shareddow
 $(eval $(call makemock, internal/definitions,      Handler,            definitionsmocks))
 $(eval $(call makemock, internal/definitions,      Sender,             definitionsmocks))
 $(eval $(call makemock, internal/events,           EventManager,       eventmocks))
+$(eval $(call makemock, internal/events/system,    EventInterface,     systemeventmocks))
 $(eval $(call makemock, internal/namespace,        Manager,            namespacemocks))
 $(eval $(call makemock, internal/networkmap,       Manager,            networkmapmocks))
 $(eval $(call makemock, internal/assets,           Manager,            assetmocks))
diff --git a/docs/reference/config.md b/docs/reference/config.md
index 72c388127..e1009783d 100644
--- a/docs/reference/config.md
+++ b/docs/reference/config.md
@@ -30,7 +30,7 @@ nav_order: 2
 
 |Key|Description|Type|Default Value|
 |---|-----------|----|-------------|
-|keynormalization|Mechanism to normalize keys before using them. Valid options are `blockchain_plugin` - use blockchain plugin (default) or `none` - do not attempt normalization (deprecated - use namespaces.predefined[].asset.manager.keynormalization)|`string`|`<nil>`
+|keyNormalization|Mechanism to normalize keys before using them. Valid options are `blockchain_plugin` - use blockchain plugin (default) or `none` - do not attempt normalization|`string`|`<nil>`
 
 ## batch.cache
 
@@ -132,10 +132,10 @@ nav_order: 2
 |batchTimeout|How long Ethconnect should wait for new events to arrive and fill a batch, before sending the batch to FireFly core. Only applies when automatically creating a new event stream|[`time.Duration`](https://pkg.go.dev/time#Duration)|`500`
 |connectionTimeout|The maximum amount of time that a connection is allowed to remain with no data transmitted|[`time.Duration`](https://pkg.go.dev/time#Duration)|`30s`
 |expectContinueTimeout|See [ExpectContinueTimeout in the Go docs](https://pkg.go.dev/net/http#Transport)|[`time.Duration`](https://pkg.go.dev/time#Duration)|`1s`
-|fromBlock|The first event this FireFly instance should listen to from the BatchPin smart contract. Default=0. Only affects initial creation of the event stream (deprecated - use namespaces.predefined[].multiparty.contract[].location.firstEvent)|Address `string`|`0`
+|fromBlock|The first event this FireFly instance should listen to from the BatchPin smart contract. Default=0. Only affects initial creation of the event stream (deprecated - use namespace.predefined[].multiparty.contract[].location.firstEvent)|Address `string`|`0`
 |headers|Adds custom headers to HTTP requests|`map[string]string`|`<nil>`
 |idleTimeout|The max duration to hold a HTTP keepalive connection between calls|[`time.Duration`](https://pkg.go.dev/time#Duration)|`475ms`
-|instance|The Ethereum address of the FireFly BatchPin smart contract that has been deployed to the blockchain (deprecated - use namespaces.predefined[].multiparty.contract[].location.address)|Address `string`|`<nil>`
+|instance|The Ethereum address of the FireFly BatchPin smart contract that has been deployed to the blockchain (deprecated - use namespace.predefined[].multiparty.contract[].location.address)|Address `string`|`<nil>`
 |maxIdleConns|The max number of idle connections to hold pooled|`int`|`100`
 |prefixLong|The prefix that will be used for Ethconnect specific HTTP headers when FireFly makes requests to Ethconnect|`string`|`firefly`
 |prefixShort|The prefix that will be used for Ethconnect specific query parameters when FireFly makes requests to Ethconnect|`string`|`fly`
@@ -217,8 +217,8 @@ nav_order: 2
 |---|-----------|----|-------------|
 |batchSize|The number of events Fabconnect should batch together for delivery to FireFly core. Only applies when automatically creating a new event stream|`int`|`50`
 |batchTimeout|The maximum amount of time to wait for a batch to complete|[`time.Duration`](https://pkg.go.dev/time#Duration)|`500`
-|chaincode|The name of the Fabric chaincode that FireFly will use for BatchPin transactions (deprecated - use namespaces.predefined[].multiparty.contract[].location.chaincode)|`string`|`<nil>`
-|channel|The Fabric channel that FireFly will use for BatchPin transactions (deprecated - use namespaces.predefined[].multiparty.contract[].location.channel)|`string`|`<nil>`
+|chaincode|The name of the Fabric chaincode that FireFly will use for BatchPin transactions (deprecated - use namespace.predefined[].multiparty.contract[].location.chaincode)|`string`|`<nil>`
+|channel|The Fabric channel that FireFly will use for BatchPin transactions (deprecated - use namespace.predefined[].multiparty.contract[].location.channel)|`string`|`<nil>`
 |connectionTimeout|The maximum amount of time that a connection is allowed to remain with no data transmitted|[`time.Duration`](https://pkg.go.dev/time#Duration)|`30s`
 |expectContinueTimeout|See [ExpectContinueTimeout in the Go docs](https://pkg.go.dev/net/http#Transport)|[`time.Duration`](https://pkg.go.dev/time#Duration)|`1s`
 |headers|Adds custom headers to HTTP requests|`map[string]string`|`<nil>`
@@ -622,12 +622,6 @@ nav_order: 2
 |plugins|The list of plugins for this namespace|`string`|`<nil>`
 |remotename|The namespace name to be sent in plugin calls, if it differs from namespace name|`string`|`<nil>`
 
-## namespaces.predefined[].asset.manager
-
-|Key|Description|Type|Default Value|
-|---|-----------|----|-------------|
-|keynormalization|Mechanism to normalize keys before using them. Valid options are `blockchain_plugin` - use blockchain plugin (default) or `none` - do not attempt normalization|`string`|`<nil>`
-
 ## namespaces.predefined[].multiparty
 
 |Key|Description|Type|Default Value|
diff --git a/docs/tutorials/broadcast_data.md b/docs/tutorials/broadcast_data.md
index 2e90b5a26..890859d71 100644
--- a/docs/tutorials/broadcast_data.md
+++ b/docs/tutorials/broadcast_data.md
@@ -198,7 +198,7 @@ At this point the data has not be shared with anyone else in the network
 
 Just include a reference to the `id` returned from the upload.
 
-`POST` `/api/v1/namespaces/default/messages/broadcast`
+`POST` `/api/v1/namespaces/default/broadcast/message`
 
 ```json
 {
diff --git a/internal/assets/manager.go b/internal/assets/manager.go
index fc05916fe..b4cbfc371 100644
--- a/internal/assets/manager.go
+++ b/internal/assets/manager.go
@@ -19,16 +19,17 @@ package assets
 import (
 	"context"
 
+	"github.com/hyperledger/firefly-common/pkg/config"
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly/internal/broadcast"
+	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/identity"
 	"github.com/hyperledger/firefly/internal/metrics"
 	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/internal/privatemessaging"
 	"github.com/hyperledger/firefly/internal/syncasync"
-	"github.com/hyperledger/firefly/internal/sysmessaging"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -51,14 +52,14 @@ type Manager interface {
 	GetTokenTransfers(ctx context.Context, filter database.AndFilter) ([]*core.TokenTransfer, *database.FilterResult, error)
 	GetTokenTransferByID(ctx context.Context, id string) (*core.TokenTransfer, error)
 
-	NewTransfer(transfer *core.TokenTransferInput) sysmessaging.MessageSender
+	NewTransfer(transfer *core.TokenTransferInput) syncasync.Sender
 	MintTokens(ctx context.Context, transfer *core.TokenTransferInput, waitConfirm bool) (*core.TokenTransfer, error)
 	BurnTokens(ctx context.Context, transfer *core.TokenTransferInput, waitConfirm bool) (*core.TokenTransfer, error)
 	TransferTokens(ctx context.Context, transfer *core.TokenTransferInput, waitConfirm bool) (*core.TokenTransfer, error)
 
 	GetTokenConnectors(ctx context.Context) []*core.TokenConnector
 
-	NewApproval(approve *core.TokenApprovalInput) sysmessaging.MessageSender
+	NewApproval(approve *core.TokenApprovalInput) syncasync.Sender
 	TokenApproval(ctx context.Context, approval *core.TokenApprovalInput, waitConfirm bool) (*core.TokenApproval, error)
 	GetTokenApprovals(ctx context.Context, filter database.AndFilter) ([]*core.TokenApproval, *database.FilterResult, error)
 
@@ -82,7 +83,7 @@ type assetManager struct {
 	keyNormalization int
 }
 
-func NewAssetManager(ctx context.Context, ns, keyNormalization string, di database.Plugin, ti map[string]tokens.Plugin, im identity.Manager, sa syncasync.Bridge, bm broadcast.Manager, pm privatemessaging.Manager, mm metrics.Manager, om operations.Manager, txHelper txcommon.Helper) (Manager, error) {
+func NewAssetManager(ctx context.Context, ns string, di database.Plugin, ti map[string]tokens.Plugin, im identity.Manager, sa syncasync.Bridge, bm broadcast.Manager, pm privatemessaging.Manager, mm metrics.Manager, om operations.Manager, txHelper txcommon.Helper) (Manager, error) {
 	if di == nil || im == nil || sa == nil || ti == nil || mm == nil || om == nil {
 		return nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "AssetManager")
 	}
@@ -96,7 +97,7 @@ func NewAssetManager(ctx context.Context, ns, keyNormalization string, di databa
 		broadcast:        bm,
 		messaging:        pm,
 		tokens:           ti,
-		keyNormalization: identity.ParseKeyNormalizationConfig(keyNormalization),
+		keyNormalization: identity.ParseKeyNormalizationConfig(config.GetString(coreconfig.AssetManagerKeyNormalization)),
 		metrics:          mm,
 		operations:       om,
 	}
diff --git a/internal/assets/token_approval.go b/internal/assets/token_approval.go
index 5d41a80f9..6b5502282 100644
--- a/internal/assets/token_approval.go
+++ b/internal/assets/token_approval.go
@@ -22,7 +22,7 @@ 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/sysmessaging"
+	"github.com/hyperledger/firefly/internal/syncasync"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -53,7 +53,7 @@ func (s *approveSender) setDefaults() {
 	s.approval.LocalID = fftypes.NewUUID()
 }
 
-func (am *assetManager) NewApproval(approval *core.TokenApprovalInput) sysmessaging.MessageSender {
+func (am *assetManager) NewApproval(approval *core.TokenApprovalInput) syncasync.Sender {
 	sender := &approveSender{
 		mgr:      am,
 		approval: approval,
diff --git a/internal/assets/token_transfer.go b/internal/assets/token_transfer.go
index a0b17d065..211f59adb 100644
--- a/internal/assets/token_transfer.go
+++ b/internal/assets/token_transfer.go
@@ -22,7 +22,7 @@ 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/sysmessaging"
+	"github.com/hyperledger/firefly/internal/syncasync"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -40,7 +40,7 @@ func (am *assetManager) GetTokenTransferByID(ctx context.Context, id string) (*c
 	return am.database.GetTokenTransferByID(ctx, am.namespace, transferID)
 }
 
-func (am *assetManager) NewTransfer(transfer *core.TokenTransferInput) sysmessaging.MessageSender {
+func (am *assetManager) NewTransfer(transfer *core.TokenTransferInput) syncasync.Sender {
 	sender := &transferSender{
 		mgr:      am,
 		transfer: transfer,
@@ -53,7 +53,7 @@ type transferSender struct {
 	mgr       *assetManager
 	transfer  *core.TokenTransferInput
 	resolved  bool
-	msgSender sysmessaging.MessageSender
+	msgSender syncasync.Sender
 }
 
 // sendMethod is the specific operation requested of the transferSender.
@@ -260,7 +260,7 @@ func (s *transferSender) sendInternal(ctx context.Context, method sendMethod) (e
 	return err
 }
 
-func (s *transferSender) buildTransferMessage(ctx context.Context, in *core.MessageInOut) (sysmessaging.MessageSender, error) {
+func (s *transferSender) buildTransferMessage(ctx context.Context, in *core.MessageInOut) (syncasync.Sender, error) {
 	allowedTypes := []fftypes.FFEnum{
 		core.MessageTypeTransferBroadcast,
 		core.MessageTypeTransferPrivate,
diff --git a/internal/batch/batch_manager.go b/internal/batch/batch_manager.go
index 3c022dd06..214045a99 100644
--- a/internal/batch/batch_manager.go
+++ b/internal/batch/batch_manager.go
@@ -30,14 +30,14 @@ import (
 	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/data"
-	"github.com/hyperledger/firefly/internal/sysmessaging"
+	"github.com/hyperledger/firefly/internal/identity"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 )
 
-func NewBatchManager(ctx context.Context, ns string, ni sysmessaging.LocalNodeInfo, di database.Plugin, dm data.Manager, txHelper txcommon.Helper) (Manager, error) {
-	if di == nil || dm == nil {
+func NewBatchManager(ctx context.Context, ns string, di database.Plugin, dm data.Manager, im identity.Manager, txHelper txcommon.Helper) (Manager, error) {
+	if di == nil || dm == nil || im == nil {
 		return nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "BatchManager")
 	}
 	pCtx, cancelCtx := context.WithCancel(log.WithLogField(ctx, "role", "batchmgr"))
@@ -46,7 +46,7 @@ func NewBatchManager(ctx context.Context, ns string, ni sysmessaging.LocalNodeIn
 		ctx:                        pCtx,
 		cancelCtx:                  cancelCtx,
 		namespace:                  ns,
-		ni:                         ni,
+		identity:                   im,
 		database:                   di,
 		data:                       dm,
 		txHelper:                   txHelper,
@@ -94,7 +94,7 @@ type batchManager struct {
 	ctx                        context.Context
 	cancelCtx                  func()
 	namespace                  string
-	ni                         sysmessaging.LocalNodeInfo
+	identity                   identity.Manager
 	database                   database.Plugin
 	data                       data.Manager
 	txHelper                   txcommon.Helper
diff --git a/internal/batch/batch_processor.go b/internal/batch/batch_processor.go
index 0653815bf..294114001 100644
--- a/internal/batch/batch_processor.go
+++ b/internal/batch/batch_processor.go
@@ -31,7 +31,6 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/retry"
 	"github.com/hyperledger/firefly/internal/data"
 	"github.com/hyperledger/firefly/internal/operations"
-	"github.com/hyperledger/firefly/internal/sysmessaging"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -76,7 +75,6 @@ type FlushStatus struct {
 type batchProcessor struct {
 	ctx                context.Context
 	bm                 *batchManager
-	ni                 sysmessaging.LocalNodeInfo
 	data               data.Manager
 	database           database.Plugin
 	txHelper           txcommon.Helper
@@ -116,7 +114,6 @@ func newBatchProcessor(bm *batchManager, conf *batchProcessorConf, baseRetryConf
 		ctx:       pCtx,
 		cancelCtx: cancelCtx,
 		bm:        bm,
-		ni:        bm.ni,
 		database:  bm.database,
 		data:      bm.data,
 		txHelper:  txHelper,
@@ -382,10 +379,13 @@ func (bp *batchProcessor) initFlushState(id *fftypes.UUID, flushWork []*batchWor
 				SignerRef: bp.conf.signer,
 				Group:     bp.conf.group,
 				Created:   fftypes.Now(),
-				Node:      bp.ni.GetNodeUUID(bp.ctx),
 			},
 		},
 	}
+	localNode, err := bp.bm.identity.GetLocalNode(bp.ctx)
+	if err == nil && localNode != nil {
+		state.Persisted.BatchHeader.Node = localNode.ID
+	}
 	for _, w := range flushWork {
 		if w.msg != nil {
 			w.msg.BatchID = id
diff --git a/internal/broadcast/manager.go b/internal/broadcast/manager.go
index 46be34e9e..38d7696f6 100644
--- a/internal/broadcast/manager.go
+++ b/internal/broadcast/manager.go
@@ -32,7 +32,6 @@ import (
 	"github.com/hyperledger/firefly/internal/multiparty"
 	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/internal/syncasync"
-	"github.com/hyperledger/firefly/internal/sysmessaging"
 	"github.com/hyperledger/firefly/pkg/blockchain"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -45,7 +44,7 @@ const broadcastDispatcherName = "pinned_broadcast"
 type Manager interface {
 	core.Named
 
-	NewBroadcast(in *core.MessageInOut) sysmessaging.MessageSender
+	NewBroadcast(in *core.MessageInOut) syncasync.Sender
 	BroadcastMessage(ctx context.Context, in *core.MessageInOut, waitConfirm bool) (out *core.Message, err error)
 	Start() error
 	WaitStop()
diff --git a/internal/broadcast/message.go b/internal/broadcast/message.go
index 9ab074fa4..b993b6164 100644
--- a/internal/broadcast/message.go
+++ b/internal/broadcast/message.go
@@ -24,11 +24,11 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/data"
-	"github.com/hyperledger/firefly/internal/sysmessaging"
+	"github.com/hyperledger/firefly/internal/syncasync"
 	"github.com/hyperledger/firefly/pkg/core"
 )
 
-func (bm *broadcastManager) NewBroadcast(in *core.MessageInOut) sysmessaging.MessageSender {
+func (bm *broadcastManager) NewBroadcast(in *core.MessageInOut) syncasync.Sender {
 	broadcast := &broadcastSender{
 		mgr: bm,
 		msg: &data.NewMessage{
diff --git a/internal/coreconfig/coreconfig.go b/internal/coreconfig/coreconfig.go
index e3b3498fd..6e04aa28d 100644
--- a/internal/coreconfig/coreconfig.go
+++ b/internal/coreconfig/coreconfig.go
@@ -41,8 +41,6 @@ const (
 	NamespacePlugins = "plugins"
 	// NamespaceDefaultKey is the default signing key for blockchain transactions within this namespace
 	NamespaceDefaultKey = "defaultkey"
-	// NamespaceAssetKeyNormalization mechanism to normalize keys before using them. Valid options: "blockchain_plugin" - use blockchain plugin (default), "none" - do not attempt normalization
-	NamespaceAssetKeyNormalization = "asset.manager.keynormalization"
 	// NamespaceMultiparty contains the multiparty configuration for a namespace
 	NamespaceMultiparty = "multiparty"
 	// NamespaceMultipartyEnabled specifies if multi-party mode is enabled for a namespace
@@ -289,7 +287,7 @@ var (
 	// TransactionCacheTTL
 	TransactionCacheTTL = ffc("transaction.cache.ttl")
 	// AssetManagerKeyNormalization mechanism to normalize keys before using them. Valid options: "blockchain_plugin" - use blockchain plugin (default), "none" - do not attempt normalization
-	AssetManagerKeyNormalization = ffc("asset.manager.keynormalization")
+	AssetManagerKeyNormalization = ffc("asset.manager.keyNormalization")
 	// UIEnabled set to false to disable the UI (default is true, so UI will be enabled if ui.path is valid)
 	UIEnabled = ffc("ui.enabled")
 	// UIPath the path on which to serve the UI
diff --git a/internal/coremsgs/en_config_descriptions.go b/internal/coremsgs/en_config_descriptions.go
index c9c062fbd..cd26b5283 100644
--- a/internal/coremsgs/en_config_descriptions.go
+++ b/internal/coremsgs/en_config_descriptions.go
@@ -41,7 +41,7 @@ var (
 	ConfigAPIDefaultFilterLimit        = ffc("config.api.defaultFilterLimit", "The maximum number of rows to return if no limit is specified on an API request", i18n.IntType)
 	ConfigAPIMaxFilterLimit            = ffc("config.api.maxFilterLimit", "The largest value of `limit` that an HTTP client can specify in a request", i18n.IntType)
 	ConfigAPIRequestMaxTimeout         = ffc("config.api.requestMaxTimeout", "The maximum amount of time that an HTTP client can specify in a `Request-Timeout` header to keep a specific request open", i18n.TimeDurationType)
-	ConfigAssetManagerKeyNormalization = ffc("config.asset.manager.keynormalization", "Mechanism to normalize keys before using them. Valid options are `blockchain_plugin` - use blockchain plugin (default) or `none` - do not attempt normalization (deprecated - use namespaces.predefined[].asset.manager.keynormalization)", i18n.StringType)
+	ConfigAssetManagerKeyNormalization = ffc("config.asset.manager.keyNormalization", "Mechanism to normalize keys before using them. Valid options are `blockchain_plugin` - use blockchain plugin (default) or `none` - do not attempt normalization", i18n.StringType)
 
 	ConfigBatchManagerMinimumPollDelay = ffc("config.batch.manager.minimumPollDelay", "The minimum time the batch manager waits between polls on the DB - to prevent thrashing", i18n.TimeDurationType)
 	ConfigBatchManagerPollTimeout      = ffc("config.batch.manager.pollTimeout", "How long to wait without any notifications of new messages before doing a page query", i18n.TimeDurationType)
@@ -70,8 +70,8 @@ var (
 
 	ConfigBlockchainEthereumEthconnectBatchSize    = ffc("config.blockchain.ethereum.ethconnect.batchSize", "The number of events Ethconnect should batch together for delivery to FireFly core. Only applies when automatically creating a new event stream", i18n.IntType)
 	ConfigBlockchainEthereumEthconnectBatchTimeout = ffc("config.blockchain.ethereum.ethconnect.batchTimeout", "How long Ethconnect should wait for new events to arrive and fill a batch, before sending the batch to FireFly core. Only applies when automatically creating a new event stream", i18n.TimeDurationType)
-	ConfigBlockchainEthereumEthconnectInstance     = ffc("config.blockchain.ethereum.ethconnect.instance", "The Ethereum address of the FireFly BatchPin smart contract that has been deployed to the blockchain (deprecated - use namespaces.predefined[].multiparty.contract[].location.address)", "Address "+i18n.StringType)
-	ConfigBlockchainEthereumEthconnectFromBlock    = ffc("config.blockchain.ethereum.ethconnect.fromBlock", "The first event this FireFly instance should listen to from the BatchPin smart contract. Default=0. Only affects initial creation of the event stream (deprecated - use namespaces.predefined[].multiparty.contract[].location.firstEvent)", "Address "+i18n.StringType)
+	ConfigBlockchainEthereumEthconnectInstance     = ffc("config.blockchain.ethereum.ethconnect.instance", "The Ethereum address of the FireFly BatchPin smart contract that has been deployed to the blockchain (deprecated - use namespace.predefined[].multiparty.contract[].location.address)", "Address "+i18n.StringType)
+	ConfigBlockchainEthereumEthconnectFromBlock    = ffc("config.blockchain.ethereum.ethconnect.fromBlock", "The first event this FireFly instance should listen to from the BatchPin smart contract. Default=0. Only affects initial creation of the event stream (deprecated - use namespace.predefined[].multiparty.contract[].location.firstEvent)", "Address "+i18n.StringType)
 	ConfigBlockchainEthereumEthconnectPrefixLong   = ffc("config.blockchain.ethereum.ethconnect.prefixLong", "The prefix that will be used for Ethconnect specific HTTP headers when FireFly makes requests to Ethconnect", i18n.StringType)
 	ConfigBlockchainEthereumEthconnectPrefixShort  = ffc("config.blockchain.ethereum.ethconnect.prefixShort", "The prefix that will be used for Ethconnect specific query parameters when FireFly makes requests to Ethconnect", i18n.StringType)
 	ConfigBlockchainEthereumEthconnectTopic        = ffc("config.blockchain.ethereum.ethconnect.topic", "The websocket listen topic that the node should register on, which is important if there are multiple nodes using a single ethconnect", i18n.StringType)
@@ -83,8 +83,8 @@ var (
 
 	ConfigBlockchainFabricFabconnectBatchSize    = ffc("config.blockchain.fabric.fabconnect.batchSize", "The number of events Fabconnect should batch together for delivery to FireFly core. Only applies when automatically creating a new event stream", i18n.IntType)
 	ConfigBlockchainFabricFabconnectBatchTimeout = ffc("config.blockchain.fabric.fabconnect.batchTimeout", "The maximum amount of time to wait for a batch to complete", i18n.TimeDurationType)
-	ConfigBlockchainFabricFabconnectChaincode    = ffc("config.blockchain.fabric.fabconnect.chaincode", "The name of the Fabric chaincode that FireFly will use for BatchPin transactions (deprecated - use namespaces.predefined[].multiparty.contract[].location.chaincode)", i18n.StringType)
-	ConfigBlockchainFabricFabconnectChannel      = ffc("config.blockchain.fabric.fabconnect.channel", "The Fabric channel that FireFly will use for BatchPin transactions (deprecated - use namespaces.predefined[].multiparty.contract[].location.channel)", i18n.StringType)
+	ConfigBlockchainFabricFabconnectChaincode    = ffc("config.blockchain.fabric.fabconnect.chaincode", "The name of the Fabric chaincode that FireFly will use for BatchPin transactions (deprecated - use namespace.predefined[].multiparty.contract[].location.chaincode)", i18n.StringType)
+	ConfigBlockchainFabricFabconnectChannel      = ffc("config.blockchain.fabric.fabconnect.channel", "The Fabric channel that FireFly will use for BatchPin transactions (deprecated - use namespace.predefined[].multiparty.contract[].location.channel)", i18n.StringType)
 	ConfigBlockchainFabricFabconnectPrefixLong   = ffc("config.blockchain.fabric.fabconnect.prefixLong", "The prefix that will be used for Fabconnect specific HTTP headers when FireFly makes requests to Fabconnect", i18n.StringType)
 	ConfigBlockchainFabricFabconnectPrefixShort  = ffc("config.blockchain.fabric.fabconnect.prefixShort", "The prefix that will be used for Fabconnect specific query parameters when FireFly makes requests to Fabconnect", i18n.StringType)
 	ConfigBlockchainFabricFabconnectSigner       = ffc("config.blockchain.fabric.fabconnect.signer", "The Fabric signing key to use when submitting transactions to Fabconnect", i18n.StringType)
@@ -257,7 +257,6 @@ var (
 	ConfigNamespacesPredefinedPlugins            = ffc("config.namespaces.predefined[].plugins", "The list of plugins for this namespace", i18n.StringType)
 	ConfigNamespacesPredefinedRemoteName         = ffc("config.namespaces.predefined[].remotename", "The namespace name to be sent in plugin calls, if it differs from namespace name", i18n.StringType)
 	ConfigNamespacesPredefinedDefaultKey         = ffc("config.namespaces.predefined[].defaultkey", "A default signing key for blockchain transactions within this namespace", i18n.StringType)
-	ConfigNamespacesPredefinedKeyNormalization   = ffc("config.namespaces.predefined[].asset.manager.keynormalization", "Mechanism to normalize keys before using them. Valid options are `blockchain_plugin` - use blockchain plugin (default) or `none` - do not attempt normalization", i18n.StringType)
 	ConfigNamespacesMultipartyEnabled            = ffc("config.namespaces.predefined[].multiparty.enabled", "Enables multi-party mode for this namespace (defaults to true if an org name or key is configured, either here or at the root level)", i18n.BooleanType)
 	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)
diff --git a/internal/events/event_manager.go b/internal/events/event_manager.go
index b22bd8f6d..b6bec83ed 100644
--- a/internal/events/event_manager.go
+++ b/internal/events/event_manager.go
@@ -41,7 +41,6 @@ import (
 	"github.com/hyperledger/firefly/internal/multiparty"
 	"github.com/hyperledger/firefly/internal/privatemessaging"
 	"github.com/hyperledger/firefly/internal/shareddownload"
-	"github.com/hyperledger/firefly/internal/sysmessaging"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/blockchain"
 	"github.com/hyperledger/firefly/pkg/core"
@@ -84,13 +83,12 @@ type EventManager interface {
 	GetPlugins() []*core.NamespaceStatusPlugin
 
 	// Internal events
-	sysmessaging.SystemEvents
+	system.EventInterface
 }
 
 type eventManager struct {
 	ctx                   context.Context
 	namespace             core.NamespaceRef
-	ni                    sysmessaging.LocalNodeInfo
 	database              database.Plugin
 	txHelper              txcommon.Helper
 	identity              identity.Manager
@@ -115,8 +113,8 @@ type eventManager struct {
 	multiparty            multiparty.Manager // optional
 }
 
-func NewEventManager(ctx context.Context, ns core.NamespaceRef, ni sysmessaging.LocalNodeInfo, di database.Plugin, bi blockchain.Plugin, im identity.Manager, dh definitions.Handler, dm data.Manager, ds definitions.Sender, bm broadcast.Manager, pm privatemessaging.Manager, am assets.Manager, sd shareddownload.Manager, mm metrics.Manager, txHelper txcommon.Helper, transports map[string]events.Plugin, mp multiparty.Manager) (EventManager, error) {
-	if ni == nil || di == nil || im == nil || dh == nil || dm == nil || ds == nil || am == nil {
+func NewEventManager(ctx context.Context, ns core.NamespaceRef, di database.Plugin, bi blockchain.Plugin, im identity.Manager, dh definitions.Handler, dm data.Manager, ds definitions.Sender, bm broadcast.Manager, pm privatemessaging.Manager, am assets.Manager, sd shareddownload.Manager, mm metrics.Manager, txHelper txcommon.Helper, transports map[string]events.Plugin, mp multiparty.Manager) (EventManager, error) {
+	if di == nil || im == nil || dh == nil || dm == nil || ds == nil || am == nil {
 		return nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "EventManager")
 	}
 	newPinNotifier := newEventNotifier(ctx, "pins")
@@ -124,7 +122,6 @@ func NewEventManager(ctx context.Context, ns core.NamespaceRef, ni sysmessaging.
 	em := &eventManager{
 		ctx:            log.WithLogField(ctx, "role", "event-manager"),
 		namespace:      ns,
-		ni:             ni,
 		database:       di,
 		txHelper:       txHelper,
 		identity:       im,
diff --git a/internal/events/persist_batch.go b/internal/events/persist_batch.go
index ccdaca474..8697fb869 100644
--- a/internal/events/persist_batch.go
+++ b/internal/events/persist_batch.go
@@ -213,11 +213,11 @@ func (em *eventManager) validateBatchMessage(ctx context.Context, batch *core.Ba
 }
 
 func (em *eventManager) sentByUs(ctx context.Context, batch *core.Batch) bool {
-	localNode := em.ni.GetNodeUUID(ctx)
-	if batch.Node == nil {
+	localNode, err := em.identity.GetLocalNode(ctx)
+	if localNode == nil || err != nil {
 		// This is from a node that hasn't yet completed registration, so we can't optimize
 		return false
-	} else if batch.Node.Equals(localNode) {
+	} else if batch.Node.Equals(localNode.ID) {
 		// We sent the batch, so we should already have all the messages and data locally
 		return true
 	}
diff --git a/internal/events/system/events.go b/internal/events/system/events.go
index 50f241d8d..b476e04ba 100644
--- a/internal/events/system/events.go
+++ b/internal/events/system/events.go
@@ -42,6 +42,10 @@ type Events struct {
 	readAhead    uint16
 }
 
+type EventInterface interface {
+	AddSystemEventListener(ns string, el EventListener) error
+}
+
 type EventListener func(event *core.EventDelivery) error
 
 func (se *Events) Name() string { return SystemEventsTransport }
diff --git a/internal/identity/identitymanager.go b/internal/identity/identitymanager.go
index 5c862b108..38dd73318 100644
--- a/internal/identity/identitymanager.go
+++ b/internal/identity/identitymanager.go
@@ -135,7 +135,7 @@ func (im *identityManager) NormalizeSigningKey(ctx context.Context, inputKey str
 	}
 	// If the caller is not confident that the blockchain plugin/connector should be used to resolve,
 	// for example it might be a different blockchain (Eth vs Fabric etc.), or it has its own
-	// verification/management of keys, it should set `namespaces.predefined[].asset.manager.keynormalization: "none"` in the config.
+	// verification/management of keys, it should set `assets.keyNormalization: "none"` in the config.
 	if keyNormalizationMode != KeyNormalizationBlockchainPlugin {
 		return inputKey, nil
 	}
diff --git a/internal/namespace/config.go b/internal/namespace/config.go
index c8d57abcd..1a8909464 100644
--- a/internal/namespace/config.go
+++ b/internal/namespace/config.go
@@ -39,7 +39,6 @@ func InitConfig(withDefaults bool) {
 	namespacePredefined.AddKnownKey(coreconfig.NamespaceRemoteName)
 	namespacePredefined.AddKnownKey(coreconfig.NamespacePlugins)
 	namespacePredefined.AddKnownKey(coreconfig.NamespaceDefaultKey)
-	namespacePredefined.AddKnownKey(coreconfig.NamespaceAssetKeyNormalization)
 
 	multipartyConf := namespacePredefined.SubSection(coreconfig.NamespaceMultiparty)
 	multipartyConf.AddKnownKey(coreconfig.NamespaceMultipartyEnabled)
@@ -56,6 +55,5 @@ func InitConfig(withDefaults bool) {
 	if withDefaults {
 		namespaceConfig.AddKnownKey(NamespacePredefined+".0."+coreconfig.NamespaceName, "default")
 		namespaceConfig.AddKnownKey(NamespacePredefined+".0."+coreconfig.NamespaceDescription, "Default predefined namespace")
-		namespaceConfig.AddKnownKey(NamespacePredefined+".0."+coreconfig.NamespaceAssetKeyNormalization, "blockchain_plugin")
 	}
 }
diff --git a/internal/namespace/manager.go b/internal/namespace/manager.go
index 2dca5d5a0..07e22cad9 100644
--- a/internal/namespace/manager.go
+++ b/internal/namespace/manager.go
@@ -791,11 +791,6 @@ func (nm *namespaceManager) loadNamespace(ctx context.Context, name string, inde
 		remoteName = name
 	}
 
-	keyNormalization := conf.GetString(coreconfig.NamespaceAssetKeyNormalization)
-	if keyNormalization == "" {
-		keyNormalization = config.GetString(coreconfig.AssetManagerKeyNormalization)
-	}
-
 	multipartyConf := conf.SubSection(coreconfig.NamespaceMultiparty)
 	// If any multiparty org information is configured (here or at the root), assume multiparty mode by default
 	orgName := multipartyConf.GetString(coreconfig.NamespaceMultipartyOrgName)
@@ -864,7 +859,6 @@ func (nm *namespaceManager) loadNamespace(ctx context.Context, name string, inde
 	config := orchestrator.Config{
 		DefaultKey:       conf.GetString(coreconfig.NamespaceDefaultKey),
 		TokenRemoteNames: nm.tokenRemoteNames,
-		KeyNormalization: keyNormalization,
 	}
 	if multipartyEnabled.(bool) {
 		contractsConf := multipartyConf.SubArray(coreconfig.NamespaceMultipartyContract)
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index 21ed04f27..78c6553e1 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -170,7 +170,6 @@ type Plugins struct {
 
 type Config struct {
 	DefaultKey       string
-	KeyNormalization string
 	Multiparty       multiparty.Config
 	TokenRemoteNames map[string]string
 }
@@ -197,7 +196,6 @@ type orchestrator struct {
 	assets         assets.Manager
 	bc             boundCallbacks
 	contracts      contracts.Manager
-	node           *fftypes.UUID
 	metrics        metrics.Manager
 	operations     operations.Manager
 	txHelper       txcommon.Helper
@@ -423,7 +421,7 @@ func (or *orchestrator) initManagers(ctx context.Context) (err error) {
 
 	if or.config.Multiparty.Enabled {
 		if or.batch == nil {
-			or.batch, err = batch.NewBatchManager(ctx, or.namespace.LocalName, or, or.database(), or.data, or.txHelper)
+			or.batch, err = batch.NewBatchManager(ctx, or.namespace.LocalName, or.database(), or.data, or.identity, or.txHelper)
 			if err != nil {
 				return err
 			}
@@ -450,7 +448,7 @@ func (or *orchestrator) initManagers(ctx context.Context) (err error) {
 	}
 
 	if or.assets == nil {
-		or.assets, err = assets.NewAssetManager(ctx, or.namespace.LocalName, or.config.KeyNormalization, or.database(), or.tokens(), or.identity, or.syncasync, or.broadcast, or.messaging, or.metrics, or.operations, or.txHelper)
+		or.assets, err = assets.NewAssetManager(ctx, or.namespace.LocalName, or.database(), or.tokens(), or.identity, or.syncasync, or.broadcast, or.messaging, or.metrics, or.operations, or.txHelper)
 		if err != nil {
 			return err
 		}
@@ -495,7 +493,7 @@ func (or *orchestrator) initComponents(ctx context.Context) (err error) {
 	}
 
 	if or.events == nil {
-		or.events, err = events.NewEventManager(ctx, or.namespace.Ref(), or, or.database(), or.blockchain(), or.identity, or.defhandler, or.data, or.defsender, or.broadcast, or.messaging, or.assets, or.sharedDownload, or.metrics, or.txHelper, or.plugins.Events, or.multiparty)
+		or.events, err = events.NewEventManager(ctx, or.namespace.Ref(), or.database(), or.blockchain(), or.identity, or.defhandler, or.data, or.defsender, or.broadcast, or.messaging, or.assets, or.sharedDownload, or.metrics, or.txHelper, or.plugins.Events, or.multiparty)
 		if err != nil {
 			return err
 		}
diff --git a/internal/orchestrator/status.go b/internal/orchestrator/status.go
index 86b7ee1ca..7b00b3cc0 100644
--- a/internal/orchestrator/status.go
+++ b/internal/orchestrator/status.go
@@ -20,7 +20,6 @@ import (
 	"context"
 	"fmt"
 
-	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -79,23 +78,6 @@ func (or *orchestrator) getPlugins() core.NamespaceStatusPlugins {
 	}
 }
 
-func (or *orchestrator) GetNodeUUID(ctx context.Context) (node *fftypes.UUID) {
-	if or.node != nil {
-		return or.node
-	}
-	status, err := or.GetStatus(ctx)
-	if err != nil {
-		log.L(or.ctx).Warnf("Failed to query local node UUID: %s", err)
-		return nil
-	}
-	if status.Node.Registered {
-		or.node = status.Node.ID
-	} else {
-		log.L(or.ctx).Infof("Node not yet registered")
-	}
-	return or.node
-}
-
 func (or *orchestrator) GetStatus(ctx context.Context) (status *core.NamespaceStatus, err error) {
 
 	status = &core.NamespaceStatus{
diff --git a/internal/privatemessaging/message.go b/internal/privatemessaging/message.go
index 08b5434b0..28a2d9367 100644
--- a/internal/privatemessaging/message.go
+++ b/internal/privatemessaging/message.go
@@ -24,11 +24,11 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/data"
-	"github.com/hyperledger/firefly/internal/sysmessaging"
+	"github.com/hyperledger/firefly/internal/syncasync"
 	"github.com/hyperledger/firefly/pkg/core"
 )
 
-func (pm *privateMessaging) NewMessage(in *core.MessageInOut) sysmessaging.MessageSender {
+func (pm *privateMessaging) NewMessage(in *core.MessageInOut) syncasync.Sender {
 	message := &messageSender{
 		mgr: pm,
 		msg: &data.NewMessage{
diff --git a/internal/privatemessaging/privatemessaging.go b/internal/privatemessaging/privatemessaging.go
index 322c27304..5986ba5f2 100644
--- a/internal/privatemessaging/privatemessaging.go
+++ b/internal/privatemessaging/privatemessaging.go
@@ -34,7 +34,6 @@ import (
 	"github.com/hyperledger/firefly/internal/multiparty"
 	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/internal/syncasync"
-	"github.com/hyperledger/firefly/internal/sysmessaging"
 	"github.com/hyperledger/firefly/pkg/blockchain"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -49,7 +48,7 @@ type Manager interface {
 	core.Named
 	GroupManager
 
-	NewMessage(msg *core.MessageInOut) sysmessaging.MessageSender
+	NewMessage(msg *core.MessageInOut) syncasync.Sender
 	SendMessage(ctx context.Context, in *core.MessageInOut, waitConfirm bool) (out *core.Message, err error)
 	RequestReply(ctx context.Context, request *core.MessageInOut) (reply *core.MessageInOut, err error)
 
diff --git a/internal/syncasync/sync_async_bridge.go b/internal/syncasync/sync_async_bridge.go
index 8c5b9ecd6..5b534ea8e 100644
--- a/internal/syncasync/sync_async_bridge.go
+++ b/internal/syncasync/sync_async_bridge.go
@@ -28,7 +28,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/data"
-	"github.com/hyperledger/firefly/internal/sysmessaging"
+	"github.com/hyperledger/firefly/internal/events/system"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 )
@@ -36,29 +36,36 @@ import (
 // Bridge translates synchronous (HTTP API) calls, into asynchronously sending a
 // message and blocking until a correlating response is received, or we hit a timeout.
 type Bridge interface {
-	// Init is required as there's a bi-directional relationship between sysmessaging and syncasync bridge
-	Init(sysevents sysmessaging.SystemEvents)
+	// Init is required as there's a bi-directional relationship between event manager and syncasync bridge
+	Init(sysevents system.EventInterface)
 
 	// The following "WaitFor*" methods all wait for a particular type of event callback, and block until it is received.
 	// To use them, invoke the appropriate method, and pass a "send" callback that is expected to trigger the relevant event.
 
 	// WaitForReply waits for a reply to the message with the supplied ID
-	WaitForReply(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.MessageInOut, error)
+	WaitForReply(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.MessageInOut, error)
 	// WaitForMessage waits for a message with the supplied ID
-	WaitForMessage(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.Message, error)
+	WaitForMessage(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.Message, error)
 	// WaitForIdentity waits for an identity with the supplied ID
-	WaitForIdentity(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.Identity, error)
+	WaitForIdentity(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.Identity, error)
 	// WaitForTokenPool waits for a token pool with the supplied ID
-	WaitForTokenPool(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.TokenPool, error)
+	WaitForTokenPool(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.TokenPool, error)
 	// WaitForTokenTransfer waits for a token transfer with the supplied ID
-	WaitForTokenTransfer(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.TokenTransfer, error)
+	WaitForTokenTransfer(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.TokenTransfer, error)
 	// WaitForTokenTransfer waits for a token approval with the supplied ID
-	WaitForTokenApproval(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.TokenApproval, error)
+	WaitForTokenApproval(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.TokenApproval, error)
 	// WaitForInvokeOperation waits for an operation with the supplied ID
-	WaitForInvokeOperation(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.Operation, error)
+	WaitForInvokeOperation(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.Operation, error)
 }
 
-type RequestSender func(ctx context.Context) error
+// Sender interface may be implemented by other types that wish to provide generic sync/async capabilities
+type Sender interface {
+	Prepare(ctx context.Context) error
+	Send(ctx context.Context) error
+	SendAndWait(ctx context.Context) error
+}
+
+type SendFunction func(ctx context.Context) error
 
 type requestType int
 
@@ -92,7 +99,7 @@ type syncAsyncBridge struct {
 	namespace   string
 	database    database.Plugin
 	data        data.Manager
-	sysevents   sysmessaging.SystemEvents
+	sysevents   system.EventInterface
 	inflightMux sync.Mutex
 	inflight    inflightRequestMap
 }
@@ -108,7 +115,7 @@ func NewSyncAsyncBridge(ctx context.Context, ns string, di database.Plugin, dm d
 	return sa
 }
 
-func (sa *syncAsyncBridge) Init(sysevents sysmessaging.SystemEvents) {
+func (sa *syncAsyncBridge) Init(sysevents system.EventInterface) {
 	sa.sysevents = sysevents
 }
 
@@ -565,7 +572,7 @@ func (sa *syncAsyncBridge) resolveFailedOperation(inflight *inflightRequest, typ
 	inflight.response <- inflightResponse{err: fmt.Errorf(op.Error)}
 }
 
-func (sa *syncAsyncBridge) sendAndWait(ctx context.Context, ns string, id *fftypes.UUID, reqType requestType, send RequestSender) (interface{}, error) {
+func (sa *syncAsyncBridge) sendAndWait(ctx context.Context, ns string, id *fftypes.UUID, reqType requestType, send SendFunction) (interface{}, error) {
 	inflight, err := sa.addInFlight(ns, id, reqType)
 	if err != nil {
 		return nil, err
@@ -595,7 +602,7 @@ func (sa *syncAsyncBridge) sendAndWait(ctx context.Context, ns string, id *fftyp
 	}
 }
 
-func (sa *syncAsyncBridge) WaitForReply(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.MessageInOut, error) {
+func (sa *syncAsyncBridge) WaitForReply(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.MessageInOut, error) {
 	reply, err := sa.sendAndWait(ctx, sa.namespace, id, messageReply, send)
 	if err != nil {
 		return nil, err
@@ -603,7 +610,7 @@ func (sa *syncAsyncBridge) WaitForReply(ctx context.Context, id *fftypes.UUID, s
 	return reply.(*core.MessageInOut), err
 }
 
-func (sa *syncAsyncBridge) WaitForMessage(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.Message, error) {
+func (sa *syncAsyncBridge) WaitForMessage(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.Message, error) {
 	reply, err := sa.sendAndWait(ctx, sa.namespace, id, messageConfirm, send)
 	if err != nil {
 		return nil, err
@@ -611,7 +618,7 @@ func (sa *syncAsyncBridge) WaitForMessage(ctx context.Context, id *fftypes.UUID,
 	return reply.(*core.Message), err
 }
 
-func (sa *syncAsyncBridge) WaitForIdentity(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.Identity, error) {
+func (sa *syncAsyncBridge) WaitForIdentity(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.Identity, error) {
 	reply, err := sa.sendAndWait(ctx, sa.namespace, id, identityConfirm, send)
 	if err != nil {
 		return nil, err
@@ -619,7 +626,7 @@ func (sa *syncAsyncBridge) WaitForIdentity(ctx context.Context, id *fftypes.UUID
 	return reply.(*core.Identity), err
 }
 
-func (sa *syncAsyncBridge) WaitForTokenPool(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.TokenPool, error) {
+func (sa *syncAsyncBridge) WaitForTokenPool(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.TokenPool, error) {
 	reply, err := sa.sendAndWait(ctx, sa.namespace, id, tokenPoolConfirm, send)
 	if err != nil {
 		return nil, err
@@ -627,7 +634,7 @@ func (sa *syncAsyncBridge) WaitForTokenPool(ctx context.Context, id *fftypes.UUI
 	return reply.(*core.TokenPool), err
 }
 
-func (sa *syncAsyncBridge) WaitForTokenTransfer(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.TokenTransfer, error) {
+func (sa *syncAsyncBridge) WaitForTokenTransfer(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.TokenTransfer, error) {
 	reply, err := sa.sendAndWait(ctx, sa.namespace, id, tokenTransferConfirm, send)
 	if err != nil {
 		return nil, err
@@ -635,7 +642,7 @@ func (sa *syncAsyncBridge) WaitForTokenTransfer(ctx context.Context, id *fftypes
 	return reply.(*core.TokenTransfer), err
 }
 
-func (sa *syncAsyncBridge) WaitForTokenApproval(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.TokenApproval, error) {
+func (sa *syncAsyncBridge) WaitForTokenApproval(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.TokenApproval, error) {
 	reply, err := sa.sendAndWait(ctx, sa.namespace, id, tokenApproveConfirm, send)
 	if err != nil {
 		return nil, err
@@ -643,7 +650,7 @@ func (sa *syncAsyncBridge) WaitForTokenApproval(ctx context.Context, id *fftypes
 	return reply.(*core.TokenApproval), err
 }
 
-func (sa *syncAsyncBridge) WaitForInvokeOperation(ctx context.Context, id *fftypes.UUID, send RequestSender) (*core.Operation, error) {
+func (sa *syncAsyncBridge) WaitForInvokeOperation(ctx context.Context, id *fftypes.UUID, send SendFunction) (*core.Operation, error) {
 	reply, err := sa.sendAndWait(ctx, sa.namespace, id, invokeOperationConfirm, send)
 	if err != nil {
 		return nil, err
diff --git a/internal/sysmessaging/localnodeinfo.go b/internal/sysmessaging/localnodeinfo.go
deleted file mode 100644
index 805dab9ca..000000000
--- a/internal/sysmessaging/localnodeinfo.go
+++ /dev/null
@@ -1,29 +0,0 @@
-// Copyright © 2022 Kaleido, Inc.
-//
-// SPDX-License-Identifier: Apache-2.0
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package sysmessaging
-
-import (
-	"context"
-
-	"github.com/hyperledger/firefly-common/pkg/fftypes"
-)
-
-// LocalNodeInfo provides an interface to query the local node info
-type LocalNodeInfo interface {
-	// GetNodeUUID returns the local node UUID in a namespace, or nil if the node is not yet registered. It is cached for fast access
-	GetNodeUUID(ctx context.Context) *fftypes.UUID
-}
diff --git a/internal/sysmessaging/message_sender.go b/internal/sysmessaging/message_sender.go
deleted file mode 100644
index 1bd83c20c..000000000
--- a/internal/sysmessaging/message_sender.go
+++ /dev/null
@@ -1,25 +0,0 @@
-// Copyright © 2021 Kaleido, Inc.
-//
-// SPDX-License-Identifier: Apache-2.0
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package sysmessaging
-
-import "context"
-
-type MessageSender interface {
-	Prepare(ctx context.Context) error
-	Send(ctx context.Context) error
-	SendAndWait(ctx context.Context) error
-}
diff --git a/internal/sysmessaging/sysevents.go b/internal/sysmessaging/sysevents.go
deleted file mode 100644
index a4e235de0..000000000
--- a/internal/sysmessaging/sysevents.go
+++ /dev/null
@@ -1,24 +0,0 @@
-// Copyright © 2021 Kaleido, Inc.
-//
-// SPDX-License-Identifier: Apache-2.0
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package sysmessaging
-
-import "github.com/hyperledger/firefly/internal/events/system"
-
-// SystemEvents specifies the internal interface, without creating a cycle
-type SystemEvents interface {
-	AddSystemEventListener(ns string, el system.EventListener) error
-}
diff --git a/mocks/assetmocks/manager.go b/mocks/assetmocks/manager.go
index 79dbefcb0..1d535c6e9 100644
--- a/mocks/assetmocks/manager.go
+++ b/mocks/assetmocks/manager.go
@@ -12,7 +12,7 @@ import (
 
 	mock "github.com/stretchr/testify/mock"
 
-	sysmessaging "github.com/hyperledger/firefly/internal/sysmessaging"
+	syncasync "github.com/hyperledger/firefly/internal/syncasync"
 )
 
 // Manager is an autogenerated mock type for the Manager type
@@ -395,15 +395,15 @@ func (_m *Manager) Name() string {
 }
 
 // NewApproval provides a mock function with given fields: approve
-func (_m *Manager) NewApproval(approve *core.TokenApprovalInput) sysmessaging.MessageSender {
+func (_m *Manager) NewApproval(approve *core.TokenApprovalInput) syncasync.Sender {
 	ret := _m.Called(approve)
 
-	var r0 sysmessaging.MessageSender
-	if rf, ok := ret.Get(0).(func(*core.TokenApprovalInput) sysmessaging.MessageSender); ok {
+	var r0 syncasync.Sender
+	if rf, ok := ret.Get(0).(func(*core.TokenApprovalInput) syncasync.Sender); ok {
 		r0 = rf(approve)
 	} else {
 		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(sysmessaging.MessageSender)
+			r0 = ret.Get(0).(syncasync.Sender)
 		}
 	}
 
@@ -411,15 +411,15 @@ func (_m *Manager) NewApproval(approve *core.TokenApprovalInput) sysmessaging.Me
 }
 
 // NewTransfer provides a mock function with given fields: transfer
-func (_m *Manager) NewTransfer(transfer *core.TokenTransferInput) sysmessaging.MessageSender {
+func (_m *Manager) NewTransfer(transfer *core.TokenTransferInput) syncasync.Sender {
 	ret := _m.Called(transfer)
 
-	var r0 sysmessaging.MessageSender
-	if rf, ok := ret.Get(0).(func(*core.TokenTransferInput) sysmessaging.MessageSender); ok {
+	var r0 syncasync.Sender
+	if rf, ok := ret.Get(0).(func(*core.TokenTransferInput) syncasync.Sender); ok {
 		r0 = rf(transfer)
 	} else {
 		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(sysmessaging.MessageSender)
+			r0 = ret.Get(0).(syncasync.Sender)
 		}
 	}
 
diff --git a/mocks/broadcastmocks/manager.go b/mocks/broadcastmocks/manager.go
index 4c778b984..062d2a2ae 100644
--- a/mocks/broadcastmocks/manager.go
+++ b/mocks/broadcastmocks/manager.go
@@ -10,7 +10,7 @@ import (
 
 	mock "github.com/stretchr/testify/mock"
 
-	sysmessaging "github.com/hyperledger/firefly/internal/sysmessaging"
+	syncasync "github.com/hyperledger/firefly/internal/syncasync"
 )
 
 // Manager is an autogenerated mock type for the Manager type
@@ -56,15 +56,15 @@ func (_m *Manager) Name() string {
 }
 
 // NewBroadcast provides a mock function with given fields: in
-func (_m *Manager) NewBroadcast(in *core.MessageInOut) sysmessaging.MessageSender {
+func (_m *Manager) NewBroadcast(in *core.MessageInOut) syncasync.Sender {
 	ret := _m.Called(in)
 
-	var r0 sysmessaging.MessageSender
-	if rf, ok := ret.Get(0).(func(*core.MessageInOut) sysmessaging.MessageSender); ok {
+	var r0 syncasync.Sender
+	if rf, ok := ret.Get(0).(func(*core.MessageInOut) syncasync.Sender); ok {
 		r0 = rf(in)
 	} else {
 		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(sysmessaging.MessageSender)
+			r0 = ret.Get(0).(syncasync.Sender)
 		}
 	}
 
diff --git a/mocks/privatemessagingmocks/manager.go b/mocks/privatemessagingmocks/manager.go
index 15eba3c57..e4a320990 100644
--- a/mocks/privatemessagingmocks/manager.go
+++ b/mocks/privatemessagingmocks/manager.go
@@ -12,7 +12,7 @@ import (
 
 	mock "github.com/stretchr/testify/mock"
 
-	sysmessaging "github.com/hyperledger/firefly/internal/sysmessaging"
+	syncasync "github.com/hyperledger/firefly/internal/syncasync"
 )
 
 // Manager is an autogenerated mock type for the Manager type
@@ -111,15 +111,15 @@ func (_m *Manager) Name() string {
 }
 
 // NewMessage provides a mock function with given fields: msg
-func (_m *Manager) NewMessage(msg *core.MessageInOut) sysmessaging.MessageSender {
+func (_m *Manager) NewMessage(msg *core.MessageInOut) syncasync.Sender {
 	ret := _m.Called(msg)
 
-	var r0 sysmessaging.MessageSender
-	if rf, ok := ret.Get(0).(func(*core.MessageInOut) sysmessaging.MessageSender); ok {
+	var r0 syncasync.Sender
+	if rf, ok := ret.Get(0).(func(*core.MessageInOut) syncasync.Sender); ok {
 		r0 = rf(msg)
 	} else {
 		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(sysmessaging.MessageSender)
+			r0 = ret.Get(0).(syncasync.Sender)
 		}
 	}
 
diff --git a/mocks/syncasyncmocks/bridge.go b/mocks/syncasyncmocks/bridge.go
index 4ec2e8d1d..08b12900f 100644
--- a/mocks/syncasyncmocks/bridge.go
+++ b/mocks/syncasyncmocks/bridge.go
@@ -12,7 +12,7 @@ import (
 
 	syncasync "github.com/hyperledger/firefly/internal/syncasync"
 
-	sysmessaging "github.com/hyperledger/firefly/internal/sysmessaging"
+	system "github.com/hyperledger/firefly/internal/events/system"
 )
 
 // Bridge is an autogenerated mock type for the Bridge type
@@ -21,16 +21,16 @@ type Bridge struct {
 }
 
 // Init provides a mock function with given fields: sysevents
-func (_m *Bridge) Init(sysevents sysmessaging.SystemEvents) {
+func (_m *Bridge) Init(sysevents system.EventInterface) {
 	_m.Called(sysevents)
 }
 
 // WaitForIdentity provides a mock function with given fields: ctx, id, send
-func (_m *Bridge) WaitForIdentity(ctx context.Context, id *fftypes.UUID, send syncasync.RequestSender) (*core.Identity, error) {
+func (_m *Bridge) WaitForIdentity(ctx context.Context, id *fftypes.UUID, send syncasync.SendFunction) (*core.Identity, error) {
 	ret := _m.Called(ctx, id, send)
 
 	var r0 *core.Identity
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) *core.Identity); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) *core.Identity); ok {
 		r0 = rf(ctx, id, send)
 	} else {
 		if ret.Get(0) != nil {
@@ -39,7 +39,7 @@ func (_m *Bridge) WaitForIdentity(ctx context.Context, id *fftypes.UUID, send sy
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) error); ok {
+	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) error); ok {
 		r1 = rf(ctx, id, send)
 	} else {
 		r1 = ret.Error(1)
@@ -49,11 +49,11 @@ func (_m *Bridge) WaitForIdentity(ctx context.Context, id *fftypes.UUID, send sy
 }
 
 // WaitForInvokeOperation provides a mock function with given fields: ctx, id, send
-func (_m *Bridge) WaitForInvokeOperation(ctx context.Context, id *fftypes.UUID, send syncasync.RequestSender) (*core.Operation, error) {
+func (_m *Bridge) WaitForInvokeOperation(ctx context.Context, id *fftypes.UUID, send syncasync.SendFunction) (*core.Operation, error) {
 	ret := _m.Called(ctx, id, send)
 
 	var r0 *core.Operation
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) *core.Operation); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) *core.Operation); ok {
 		r0 = rf(ctx, id, send)
 	} else {
 		if ret.Get(0) != nil {
@@ -62,7 +62,7 @@ func (_m *Bridge) WaitForInvokeOperation(ctx context.Context, id *fftypes.UUID,
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) error); ok {
+	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) error); ok {
 		r1 = rf(ctx, id, send)
 	} else {
 		r1 = ret.Error(1)
@@ -72,11 +72,11 @@ func (_m *Bridge) WaitForInvokeOperation(ctx context.Context, id *fftypes.UUID,
 }
 
 // WaitForMessage provides a mock function with given fields: ctx, id, send
-func (_m *Bridge) WaitForMessage(ctx context.Context, id *fftypes.UUID, send syncasync.RequestSender) (*core.Message, error) {
+func (_m *Bridge) WaitForMessage(ctx context.Context, id *fftypes.UUID, send syncasync.SendFunction) (*core.Message, error) {
 	ret := _m.Called(ctx, id, send)
 
 	var r0 *core.Message
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) *core.Message); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) *core.Message); ok {
 		r0 = rf(ctx, id, send)
 	} else {
 		if ret.Get(0) != nil {
@@ -85,7 +85,7 @@ func (_m *Bridge) WaitForMessage(ctx context.Context, id *fftypes.UUID, send syn
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) error); ok {
+	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) error); ok {
 		r1 = rf(ctx, id, send)
 	} else {
 		r1 = ret.Error(1)
@@ -95,11 +95,11 @@ func (_m *Bridge) WaitForMessage(ctx context.Context, id *fftypes.UUID, send syn
 }
 
 // WaitForReply provides a mock function with given fields: ctx, id, send
-func (_m *Bridge) WaitForReply(ctx context.Context, id *fftypes.UUID, send syncasync.RequestSender) (*core.MessageInOut, error) {
+func (_m *Bridge) WaitForReply(ctx context.Context, id *fftypes.UUID, send syncasync.SendFunction) (*core.MessageInOut, error) {
 	ret := _m.Called(ctx, id, send)
 
 	var r0 *core.MessageInOut
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) *core.MessageInOut); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) *core.MessageInOut); ok {
 		r0 = rf(ctx, id, send)
 	} else {
 		if ret.Get(0) != nil {
@@ -108,7 +108,7 @@ func (_m *Bridge) WaitForReply(ctx context.Context, id *fftypes.UUID, send synca
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) error); ok {
+	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) error); ok {
 		r1 = rf(ctx, id, send)
 	} else {
 		r1 = ret.Error(1)
@@ -118,11 +118,11 @@ func (_m *Bridge) WaitForReply(ctx context.Context, id *fftypes.UUID, send synca
 }
 
 // WaitForTokenApproval provides a mock function with given fields: ctx, id, send
-func (_m *Bridge) WaitForTokenApproval(ctx context.Context, id *fftypes.UUID, send syncasync.RequestSender) (*core.TokenApproval, error) {
+func (_m *Bridge) WaitForTokenApproval(ctx context.Context, id *fftypes.UUID, send syncasync.SendFunction) (*core.TokenApproval, error) {
 	ret := _m.Called(ctx, id, send)
 
 	var r0 *core.TokenApproval
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) *core.TokenApproval); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) *core.TokenApproval); ok {
 		r0 = rf(ctx, id, send)
 	} else {
 		if ret.Get(0) != nil {
@@ -131,7 +131,7 @@ func (_m *Bridge) WaitForTokenApproval(ctx context.Context, id *fftypes.UUID, se
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) error); ok {
+	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) error); ok {
 		r1 = rf(ctx, id, send)
 	} else {
 		r1 = ret.Error(1)
@@ -141,11 +141,11 @@ func (_m *Bridge) WaitForTokenApproval(ctx context.Context, id *fftypes.UUID, se
 }
 
 // WaitForTokenPool provides a mock function with given fields: ctx, id, send
-func (_m *Bridge) WaitForTokenPool(ctx context.Context, id *fftypes.UUID, send syncasync.RequestSender) (*core.TokenPool, error) {
+func (_m *Bridge) WaitForTokenPool(ctx context.Context, id *fftypes.UUID, send syncasync.SendFunction) (*core.TokenPool, error) {
 	ret := _m.Called(ctx, id, send)
 
 	var r0 *core.TokenPool
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) *core.TokenPool); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) *core.TokenPool); ok {
 		r0 = rf(ctx, id, send)
 	} else {
 		if ret.Get(0) != nil {
@@ -154,7 +154,7 @@ func (_m *Bridge) WaitForTokenPool(ctx context.Context, id *fftypes.UUID, send s
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) error); ok {
+	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) error); ok {
 		r1 = rf(ctx, id, send)
 	} else {
 		r1 = ret.Error(1)
@@ -164,11 +164,11 @@ func (_m *Bridge) WaitForTokenPool(ctx context.Context, id *fftypes.UUID, send s
 }
 
 // WaitForTokenTransfer provides a mock function with given fields: ctx, id, send
-func (_m *Bridge) WaitForTokenTransfer(ctx context.Context, id *fftypes.UUID, send syncasync.RequestSender) (*core.TokenTransfer, error) {
+func (_m *Bridge) WaitForTokenTransfer(ctx context.Context, id *fftypes.UUID, send syncasync.SendFunction) (*core.TokenTransfer, error) {
 	ret := _m.Called(ctx, id, send)
 
 	var r0 *core.TokenTransfer
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) *core.TokenTransfer); ok {
+	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) *core.TokenTransfer); ok {
 		r0 = rf(ctx, id, send)
 	} else {
 		if ret.Get(0) != nil {
@@ -177,7 +177,7 @@ func (_m *Bridge) WaitForTokenTransfer(ctx context.Context, id *fftypes.UUID, se
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.RequestSender) error); ok {
+	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, syncasync.SendFunction) error); ok {
 		r1 = rf(ctx, id, send)
 	} else {
 		r1 = ret.Error(1)
diff --git a/mocks/sysmessagingmocks/message_sender.go b/mocks/syncasyncmocks/sender.go
similarity index 71%
rename from mocks/sysmessagingmocks/message_sender.go
rename to mocks/syncasyncmocks/sender.go
index bc300a98f..caf2c553d 100644
--- a/mocks/sysmessagingmocks/message_sender.go
+++ b/mocks/syncasyncmocks/sender.go
@@ -1,6 +1,6 @@
 // Code generated by mockery v1.0.0. DO NOT EDIT.
 
-package sysmessagingmocks
+package syncasyncmocks
 
 import (
 	context "context"
@@ -8,13 +8,13 @@ import (
 	mock "github.com/stretchr/testify/mock"
 )
 
-// MessageSender is an autogenerated mock type for the MessageSender type
-type MessageSender struct {
+// Sender is an autogenerated mock type for the Sender type
+type Sender struct {
 	mock.Mock
 }
 
 // Prepare provides a mock function with given fields: ctx
-func (_m *MessageSender) Prepare(ctx context.Context) error {
+func (_m *Sender) Prepare(ctx context.Context) error {
 	ret := _m.Called(ctx)
 
 	var r0 error
@@ -28,7 +28,7 @@ func (_m *MessageSender) Prepare(ctx context.Context) error {
 }
 
 // Send provides a mock function with given fields: ctx
-func (_m *MessageSender) Send(ctx context.Context) error {
+func (_m *Sender) Send(ctx context.Context) error {
 	ret := _m.Called(ctx)
 
 	var r0 error
@@ -42,7 +42,7 @@ func (_m *MessageSender) Send(ctx context.Context) error {
 }
 
 // SendAndWait provides a mock function with given fields: ctx
-func (_m *MessageSender) SendAndWait(ctx context.Context) error {
+func (_m *Sender) SendAndWait(ctx context.Context) error {
 	ret := _m.Called(ctx)
 
 	var r0 error
diff --git a/mocks/sysmessagingmocks/local_node_info.go b/mocks/sysmessagingmocks/local_node_info.go
deleted file mode 100644
index b9c9d758e..000000000
--- a/mocks/sysmessagingmocks/local_node_info.go
+++ /dev/null
@@ -1,31 +0,0 @@
-// Code generated by mockery v1.0.0. DO NOT EDIT.
-
-package sysmessagingmocks
-
-import (
-	context "context"
-
-	fftypes "github.com/hyperledger/firefly-common/pkg/fftypes"
-	mock "github.com/stretchr/testify/mock"
-)
-
-// LocalNodeInfo is an autogenerated mock type for the LocalNodeInfo type
-type LocalNodeInfo struct {
-	mock.Mock
-}
-
-// GetNodeUUID provides a mock function with given fields: ctx
-func (_m *LocalNodeInfo) GetNodeUUID(ctx context.Context) *fftypes.UUID {
-	ret := _m.Called(ctx)
-
-	var r0 *fftypes.UUID
-	if rf, ok := ret.Get(0).(func(context.Context) *fftypes.UUID); ok {
-		r0 = rf(ctx)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*fftypes.UUID)
-		}
-	}
-
-	return r0
-}
diff --git a/mocks/sysmessagingmocks/system_events.go b/mocks/systemeventmocks/event_interface.go
similarity index 66%
rename from mocks/sysmessagingmocks/system_events.go
rename to mocks/systemeventmocks/event_interface.go
index 640983e49..5a8152e2f 100644
--- a/mocks/sysmessagingmocks/system_events.go
+++ b/mocks/systemeventmocks/event_interface.go
@@ -1,20 +1,19 @@
 // Code generated by mockery v1.0.0. DO NOT EDIT.
 
-package sysmessagingmocks
+package systemeventmocks
 
 import (
-	mock "github.com/stretchr/testify/mock"
-
 	system "github.com/hyperledger/firefly/internal/events/system"
+	mock "github.com/stretchr/testify/mock"
 )
 
-// SystemEvents is an autogenerated mock type for the SystemEvents type
-type SystemEvents struct {
+// EventInterface is an autogenerated mock type for the EventInterface type
+type EventInterface struct {
 	mock.Mock
 }
 
 // AddSystemEventListener provides a mock function with given fields: ns, el
-func (_m *SystemEvents) AddSystemEventListener(ns string, el system.EventListener) error {
+func (_m *EventInterface) AddSystemEventListener(ns string, el system.EventListener) error {
 	ret := _m.Called(ns, el)
 
 	var r0 error


Test patch:
diff --git a/internal/assets/manager_test.go b/internal/assets/manager_test.go
index 851cf54ff..6cae98b09 100644
--- a/internal/assets/manager_test.go
+++ b/internal/assets/manager_test.go
@@ -63,7 +63,7 @@ func newTestAssetsCommon(t *testing.T, metrics bool) (*assetManager, func()) {
 	mom.On("RegisterHandler", mock.Anything, mock.Anything, mock.Anything)
 	mti.On("Name").Return("ut").Maybe()
 	ctx, cancel := context.WithCancel(context.Background())
-	a, err := NewAssetManager(ctx, "ns1", "blockchain_plugin", mdi, map[string]tokens.Plugin{"magic-tokens": mti}, mim, msa, mbm, mpm, mm, mom, txHelper)
+	a, err := NewAssetManager(ctx, "ns1", mdi, map[string]tokens.Plugin{"magic-tokens": mti}, mim, msa, mbm, mpm, mm, mom, txHelper)
 	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything).Maybe()
 	rag.RunFn = func(a mock.Arguments) {
 		rag.ReturnArguments = mock.Arguments{a[1].(func(context.Context) error)(a[0].(context.Context))}
@@ -75,7 +75,7 @@ func newTestAssetsCommon(t *testing.T, metrics bool) (*assetManager, func()) {
 }
 
 func TestInitFail(t *testing.T) {
-	_, err := NewAssetManager(context.Background(), "", "", nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	_, err := NewAssetManager(context.Background(), "", nil, nil, nil, nil, nil, nil, nil, nil, nil)
 	assert.Regexp(t, "FF10128", err)
 }
 
diff --git a/internal/assets/token_approval_test.go b/internal/assets/token_approval_test.go
index 763b436ba..52b274489 100644
--- a/internal/assets/token_approval_test.go
+++ b/internal/assets/token_approval_test.go
@@ -449,7 +449,7 @@ func TestTokenApprovalConfirm(t *testing.T) {
 
 	msa.On("WaitForTokenApproval", context.Background(), mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(context.Background())
 		}).
 		Return(&core.TokenApproval{}, nil)
diff --git a/internal/assets/token_pool_test.go b/internal/assets/token_pool_test.go
index d8e0a567e..91ef84979 100644
--- a/internal/assets/token_pool_test.go
+++ b/internal/assets/token_pool_test.go
@@ -368,7 +368,7 @@ func TestCreateTokenPoolConfirm(t *testing.T) {
 	mdi.On("InsertOperation", context.Background(), mock.Anything).Return(nil)
 	msa.On("WaitForTokenPool", context.Background(), mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(context.Background())
 		}).
 		Return(nil, nil)
diff --git a/internal/assets/token_transfer_test.go b/internal/assets/token_transfer_test.go
index 5f9ae6aa4..43d306484 100644
--- a/internal/assets/token_transfer_test.go
+++ b/internal/assets/token_transfer_test.go
@@ -28,7 +28,6 @@ import (
 	"github.com/hyperledger/firefly/mocks/operationmocks"
 	"github.com/hyperledger/firefly/mocks/privatemessagingmocks"
 	"github.com/hyperledger/firefly/mocks/syncasyncmocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
 	"github.com/hyperledger/firefly/mocks/txcommonmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -412,7 +411,7 @@ func TestMintTokensConfirm(t *testing.T) {
 	mdi.On("InsertOperation", context.Background(), mock.Anything).Return(nil)
 	msa.On("WaitForTokenTransfer", context.Background(), mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(context.Background())
 		}).
 		Return(&core.TokenTransfer{}, nil)
@@ -519,7 +518,7 @@ func TestBurnTokensConfirm(t *testing.T) {
 	mdi.On("InsertOperation", context.Background(), mock.Anything).Return(nil)
 	msa.On("WaitForTokenTransfer", context.Background(), mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(context.Background())
 		}).
 		Return(&core.TokenTransfer{}, nil)
@@ -724,7 +723,7 @@ func TestTransferTokensWithBroadcastMessage(t *testing.T) {
 	mdi := am.database.(*databasemocks.Plugin)
 	mim := am.identity.(*identitymanagermocks.Manager)
 	mbm := am.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	mth := am.txHelper.(*txcommonmocks.Helper)
 	mom := am.operations.(*operationmocks.Manager)
 	mim.On("NormalizeSigningKey", context.Background(), "", identity.KeyNormalizationBlockchainPlugin).Return("0x12345", nil)
@@ -820,7 +819,7 @@ func TestTransferTokensWithBroadcastMessageSendFail(t *testing.T) {
 	mdi := am.database.(*databasemocks.Plugin)
 	mim := am.identity.(*identitymanagermocks.Manager)
 	mbm := am.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	mth := am.txHelper.(*txcommonmocks.Helper)
 	mom := am.operations.(*operationmocks.Manager)
 	mim.On("NormalizeSigningKey", context.Background(), "", identity.KeyNormalizationBlockchainPlugin).Return("0x12345", nil)
@@ -865,7 +864,7 @@ func TestTransferTokensWithBroadcastPrepareFail(t *testing.T) {
 	}
 
 	mbm := am.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	mbm.On("NewBroadcast", transfer.Message).Return(mms)
 	mms.On("Prepare", context.Background()).Return(fmt.Errorf("pop"))
 
@@ -912,7 +911,7 @@ func TestTransferTokensWithPrivateMessage(t *testing.T) {
 	mdi := am.database.(*databasemocks.Plugin)
 	mim := am.identity.(*identitymanagermocks.Manager)
 	mpm := am.messaging.(*privatemessagingmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	mth := am.txHelper.(*txcommonmocks.Helper)
 	mom := am.operations.(*operationmocks.Manager)
 	mim.On("NormalizeSigningKey", context.Background(), "", identity.KeyNormalizationBlockchainPlugin).Return("0x12345", nil)
@@ -1031,7 +1030,7 @@ func TestTransferTokensConfirm(t *testing.T) {
 	mdi.On("InsertOperation", context.Background(), mock.Anything).Return(nil)
 	msa.On("WaitForTokenTransfer", context.Background(), mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(context.Background())
 		}).
 		Return(&core.TokenTransfer{}, nil)
@@ -1085,7 +1084,7 @@ func TestTransferTokensWithBroadcastConfirm(t *testing.T) {
 	mdi := am.database.(*databasemocks.Plugin)
 	mim := am.identity.(*identitymanagermocks.Manager)
 	mbm := am.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	msa := am.syncasync.(*syncasyncmocks.Bridge)
 	mth := am.txHelper.(*txcommonmocks.Helper)
 	mom := am.operations.(*operationmocks.Manager)
@@ -1098,13 +1097,13 @@ func TestTransferTokensWithBroadcastConfirm(t *testing.T) {
 	mms.On("Send", context.Background()).Return(nil)
 	msa.On("WaitForMessage", context.Background(), mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(context.Background())
 		}).
 		Return(&core.Message{}, nil)
 	msa.On("WaitForTokenTransfer", context.Background(), mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(context.Background())
 		}).
 		Return(&transfer.TokenTransfer, nil)
diff --git a/internal/batch/batch_manager_test.go b/internal/batch/batch_manager_test.go
index 4440af911..382fffc2b 100644
--- a/internal/batch/batch_manager_test.go
+++ b/internal/batch/batch_manager_test.go
@@ -30,7 +30,7 @@ import (
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
+	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/stretchr/testify/assert"
@@ -46,9 +46,9 @@ func testConfigReset() {
 func newTestBatchManager(t *testing.T) (*batchManager, func()) {
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	bm, err := NewBatchManager(context.Background(), "ns1", mni, mdi, mdm, txHelper)
+	bm, err := NewBatchManager(context.Background(), "ns1", mdi, mdm, mim, txHelper)
 	assert.NoError(t, err)
 	return bm.(*batchManager), bm.(*batchManager).cancelCtx
 }
@@ -58,9 +58,10 @@ func TestE2EDispatchBroadcast(t *testing.T) {
 
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	mni.On("GetNodeUUID", mock.Anything).Return(fftypes.NewUUID())
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
+
 	readyForDispatch := make(chan bool)
 	waitForDispatch := make(chan *DispatchState)
 	handler := func(ctx context.Context, state *DispatchState) error {
@@ -89,7 +90,7 @@ func TestE2EDispatchBroadcast(t *testing.T) {
 		return nil
 	}
 	ctx, cancel := context.WithCancel(context.Background())
-	bmi, _ := NewBatchManager(ctx, "ns1", mni, mdi, mdm, txHelper)
+	bmi, _ := NewBatchManager(ctx, "ns1", mdi, mdm, mim, txHelper)
 	bm := bmi.(*batchManager)
 	bm.readOffset = 1000
 
@@ -174,9 +175,10 @@ func TestE2EDispatchPrivateUnpinned(t *testing.T) {
 
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	mni.On("GetNodeUUID", mock.Anything).Return(fftypes.NewUUID())
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
+
 	readyForDispatch := make(chan bool)
 	waitForDispatch := make(chan *DispatchState)
 	var groupID fftypes.Bytes32
@@ -208,7 +210,7 @@ func TestE2EDispatchPrivateUnpinned(t *testing.T) {
 		return nil
 	}
 	ctx, cancel := context.WithCancel(context.Background())
-	bmi, _ := NewBatchManager(ctx, "ns1", mni, mdi, mdm, txHelper)
+	bmi, _ := NewBatchManager(ctx, "ns1", mdi, mdm, mim, txHelper)
 	bm := bmi.(*batchManager)
 
 	bm.RegisterDispatcher("utdispatcher", core.TransactionTypeBatchPin, []core.MessageType{core.MessageTypePrivate}, handler, DispatcherOptions{
@@ -285,10 +287,10 @@ func TestDispatchUnknownType(t *testing.T) {
 
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
 	ctx, cancel := context.WithCancel(context.Background())
-	bmi, _ := NewBatchManager(ctx, "ns1", mni, mdi, mdm, txHelper)
+	bmi, _ := NewBatchManager(ctx, "ns1", mdi, mdm, mim, txHelper)
 	bm := bmi.(*batchManager)
 
 	msg := &core.Message{
@@ -316,9 +318,9 @@ func TestGetInvalidBatchTypeMsg(t *testing.T) {
 
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	bm, _ := NewBatchManager(context.Background(), "ns1", mni, mdi, mdm, txHelper)
+	bm, _ := NewBatchManager(context.Background(), "ns1", mdi, mdm, mim, txHelper)
 	defer bm.Close()
 	_, err := bm.(*batchManager).getProcessor(core.BatchTypeBroadcast, "wrong", nil, &core.SignerRef{})
 	assert.Regexp(t, "FF10126", err)
@@ -327,10 +329,10 @@ func TestGetInvalidBatchTypeMsg(t *testing.T) {
 func TestMessageSequencerCancelledContext(t *testing.T) {
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).Return(nil, fmt.Errorf("pop")).Once()
-	bm, _ := NewBatchManager(context.Background(), "ns1", mni, mdi, mdm, txHelper)
+	bm, _ := NewBatchManager(context.Background(), "ns1", mdi, mdm, mim, txHelper)
 	defer bm.Close()
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel()
@@ -342,9 +344,9 @@ func TestMessageSequencerCancelledContext(t *testing.T) {
 func TestMessageSequencerMissingMessageData(t *testing.T) {
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	bm, _ := NewBatchManager(context.Background(), "ns1", mni, mdi, mdm, txHelper)
+	bm, _ := NewBatchManager(context.Background(), "ns1", mdi, mdm, mim, txHelper)
 	bm.RegisterDispatcher("utdispatcher", core.TransactionTypeNone, []core.MessageType{core.MessageTypeBroadcast},
 		func(c context.Context, state *DispatchState) error {
 			return nil
@@ -384,11 +386,11 @@ func TestMessageSequencerMissingMessageData(t *testing.T) {
 func TestMessageSequencerUpdateMessagesFail(t *testing.T) {
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	mni.On("GetNodeUUID", mock.Anything).Return(fftypes.NewUUID())
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
 	ctx, cancelCtx := context.WithCancel(context.Background())
-	bm, _ := NewBatchManager(ctx, "ns1", mni, mdi, mdm, txHelper)
+	bm, _ := NewBatchManager(ctx, "ns1", mdi, mdm, mim, txHelper)
 	bm.RegisterDispatcher("utdispatcher", core.TransactionTypeBatchPin, []core.MessageType{core.MessageTypeBroadcast},
 		func(c context.Context, state *DispatchState) error {
 			return nil
@@ -439,11 +441,11 @@ func TestMessageSequencerUpdateMessagesFail(t *testing.T) {
 func TestMessageSequencerDispatchFail(t *testing.T) {
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
-	mni.On("GetNodeUUID", mock.Anything).Return(fftypes.NewUUID())
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
 	ctx, cancelCtx := context.WithCancel(context.Background())
-	bm, _ := NewBatchManager(ctx, "ns1", mni, mdi, mdm, txHelper)
+	bm, _ := NewBatchManager(ctx, "ns1", mdi, mdm, mim, txHelper)
 	bm.RegisterDispatcher("utdispatcher", core.TransactionTypeBatchPin, []core.MessageType{core.MessageTypeBroadcast},
 		func(c context.Context, state *DispatchState) error {
 			cancelCtx()
@@ -479,11 +481,11 @@ func TestMessageSequencerDispatchFail(t *testing.T) {
 func TestMessageSequencerUpdateBatchFail(t *testing.T) {
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
-	mni.On("GetNodeUUID", mock.Anything).Return(fftypes.NewUUID())
+	mim := &identitymanagermocks.Manager{}
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	bm, _ := NewBatchManager(ctx, "ns1", mni, mdi, mdm, txHelper)
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
+	bm, _ := NewBatchManager(ctx, "ns1", mdi, mdm, mim, txHelper)
 	bm.RegisterDispatcher("utdispatcher", core.TransactionTypeBatchPin, []core.MessageType{core.MessageTypeBroadcast},
 		func(c context.Context, state *DispatchState) error {
 			return nil
@@ -564,9 +566,9 @@ func TestRewindForNewMessage(t *testing.T) {
 func TestAssembleMessageDataNilData(t *testing.T) {
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	bm, _ := NewBatchManager(context.Background(), "ns1", mni, mdi, mdm, txHelper)
+	bm, _ := NewBatchManager(context.Background(), "ns1", mdi, mdm, mim, txHelper)
 	bm.Close()
 	mdm.On("GetMessageWithDataCached", mock.Anything, mock.Anything).Return(nil, nil, false, nil)
 	_, _, err := bm.(*batchManager).assembleMessageData(fftypes.NewUUID())
@@ -576,9 +578,9 @@ func TestAssembleMessageDataNilData(t *testing.T) {
 func TestGetMessageDataFail(t *testing.T) {
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	bm, _ := NewBatchManager(context.Background(), "ns1", mni, mdi, mdm, txHelper)
+	bm, _ := NewBatchManager(context.Background(), "ns1", mdi, mdm, mim, txHelper)
 	mdm.On("GetMessageWithDataCached", mock.Anything, mock.Anything).Return(nil, nil, false, fmt.Errorf("pop"))
 	bm.Close()
 	_, _, err := bm.(*batchManager).assembleMessageData(fftypes.NewUUID())
@@ -589,9 +591,9 @@ func TestGetMessageDataFail(t *testing.T) {
 func TestGetMessageNotFound(t *testing.T) {
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
+	mim := &identitymanagermocks.Manager{}
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	bm, _ := NewBatchManager(context.Background(), "ns1", mni, mdi, mdm, txHelper)
+	bm, _ := NewBatchManager(context.Background(), "ns1", mdi, mdm, mim, txHelper)
 	mdm.On("GetMessageWithDataCached", mock.Anything, mock.Anything).Return(nil, nil, false, nil)
 	bm.Close()
 	_, _, err := bm.(*batchManager).assembleMessageData(fftypes.NewUUID())
diff --git a/internal/batch/batch_processor_test.go b/internal/batch/batch_processor_test.go
index 4463db68f..8729c5dc7 100644
--- a/internal/batch/batch_processor_test.go
+++ b/internal/batch/batch_processor_test.go
@@ -29,7 +29,7 @@ import (
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
+	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
 	"github.com/hyperledger/firefly/mocks/txcommonmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
@@ -39,10 +39,8 @@ import (
 func newTestBatchProcessor(t *testing.T, dispatch DispatchHandler) (func(), *databasemocks.Plugin, *batchProcessor) {
 	bm, cancel := newTestBatchManager(t)
 	mdi := bm.database.(*databasemocks.Plugin)
-	mni := bm.ni.(*sysmessagingmocks.LocalNodeInfo)
 	mdm := bm.data.(*datamocks.Manager)
 	txHelper := txcommon.NewTransactionHelper("ns1", mdi, mdm)
-	mni.On("GetNodeUUID", mock.Anything).Return(fftypes.NewUUID()).Maybe()
 	bp := newBatchProcessor(bm, &batchProcessorConf{
 		txType:   core.TransactionTypeBatchPin,
 		signer:   core.SignerRef{Author: "did:firefly:org/abcd", Key: "0x12345"},
@@ -90,6 +88,9 @@ func TestUnfilledBatch(t *testing.T) {
 	mdm := bp.data.(*datamocks.Manager)
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 
+	mim := bp.bm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
+
 	// Dispatch the work
 	go func() {
 		for i := 0; i < 5; i++ {
@@ -112,6 +113,7 @@ func TestUnfilledBatch(t *testing.T) {
 	mdm.AssertExpectations(t)
 	mdi.AssertExpectations(t)
 	mth.AssertExpectations(t)
+	mim.AssertExpectations(t)
 }
 
 func TestBatchSizeOverflow(t *testing.T) {
@@ -135,6 +137,9 @@ func TestBatchSizeOverflow(t *testing.T) {
 	mdm := bp.data.(*datamocks.Manager)
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 
+	mim := bp.bm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
+
 	// Dispatch the work
 	msgIDs := []*fftypes.UUID{fftypes.NewUUID(), fftypes.NewUUID()}
 	go func() {
@@ -161,6 +166,7 @@ func TestBatchSizeOverflow(t *testing.T) {
 	mdi.AssertExpectations(t)
 	mth.AssertExpectations(t)
 	mdm.AssertExpectations(t)
+	mim.AssertExpectations(t)
 }
 
 func TestCloseToUnblockDispatch(t *testing.T) {
@@ -192,6 +198,9 @@ func TestCloseToUnblockUpsertBatch(t *testing.T) {
 		}).
 		Return(nil, fmt.Errorf("pop"))
 
+	mim := bp.bm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
+
 	// Generate the work
 	msgid := fftypes.NewUUID()
 	go func() {
@@ -207,6 +216,9 @@ func TestCloseToUnblockUpsertBatch(t *testing.T) {
 	// Close to unblock
 	bp.cancelCtx()
 	<-bp.done
+
+	mth.AssertExpectations(t)
+	mim.AssertExpectations(t)
 }
 
 func TestInsertNewNonceFail(t *testing.T) {
@@ -412,6 +424,9 @@ func TestMarkMessageDispatchedUnpinnedOK(t *testing.T) {
 	mdm := bp.data.(*datamocks.Manager)
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 
+	mim := bp.bm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
+
 	// Dispatch the work
 	go func() {
 		for i := 0; i < 5; i++ {
@@ -434,6 +449,7 @@ func TestMarkMessageDispatchedUnpinnedOK(t *testing.T) {
 	mdi.AssertExpectations(t)
 	mdm.AssertExpectations(t)
 	mth.AssertExpectations(t)
+	mim.AssertExpectations(t)
 }
 
 func TestMaskContextsRetryAfterPinsAssigned(t *testing.T) {
@@ -463,6 +479,9 @@ func TestMaskContextsRetryAfterPinsAssigned(t *testing.T) {
 	mth := bp.txHelper.(*txcommonmocks.Helper)
 	mth.On("SubmitNewTransaction", mock.Anything, core.TransactionTypeBatchPin).Return(fftypes.NewUUID(), nil)
 
+	mim := bp.bm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
+
 	groupID := fftypes.NewRandB32()
 	msg1 := &core.Message{
 		Header: core.MessageHeader{
@@ -494,6 +513,9 @@ func TestMaskContextsRetryAfterPinsAssigned(t *testing.T) {
 	<-bp.done
 
 	mdi.AssertExpectations(t)
+	mdm.AssertExpectations(t)
+	mim.AssertExpectations(t)
+	mth.AssertExpectations(t)
 }
 
 func TestMaskContextsUpdateMessageFail(t *testing.T) {
@@ -512,6 +534,9 @@ func TestMaskContextsUpdateMessageFail(t *testing.T) {
 	mdi.On("InsertNonce", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("UpdateMessage", mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("pop")).Once()
 
+	mim := bp.bm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
+
 	msg := &core.Message{
 		Header: core.MessageHeader{
 			ID:     fftypes.NewUUID(),
@@ -529,6 +554,7 @@ func TestMaskContextsUpdateMessageFail(t *testing.T) {
 	<-bp.done
 
 	mdi.AssertExpectations(t)
+	mim.AssertExpectations(t)
 }
 
 func TestBigBatchEstimate(t *testing.T) {
diff --git a/internal/broadcast/message_test.go b/internal/broadcast/message_test.go
index 954005e98..80d2b4533 100644
--- a/internal/broadcast/message_test.go
+++ b/internal/broadcast/message_test.go
@@ -113,7 +113,7 @@ func TestBroadcastMessageWaitConfirmOk(t *testing.T) {
 	}
 	msa.On("WaitForMessage", ctx, mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(ctx)
 		}).
 		Return(replyMsg, nil)
diff --git a/internal/contracts/manager_test.go b/internal/contracts/manager_test.go
index 1c03888eb..13c1e2285 100644
--- a/internal/contracts/manager_test.go
+++ b/internal/contracts/manager_test.go
@@ -1338,7 +1338,7 @@ func TestInvokeContractConfirm(t *testing.T) {
 	})).Return(nil, nil)
 	msa.On("WaitForInvokeOperation", mock.Anything, mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(context.Background())
 		}).
 		Return(&core.Operation{}, nil)
diff --git a/internal/definitions/sender_contracts_test.go b/internal/definitions/sender_contracts_test.go
index e393222cf..f109b52f9 100644
--- a/internal/definitions/sender_contracts_test.go
+++ b/internal/definitions/sender_contracts_test.go
@@ -25,7 +25,7 @@ import (
 	"github.com/hyperledger/firefly/mocks/broadcastmocks"
 	"github.com/hyperledger/firefly/mocks/contractmocks"
 	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
+	"github.com/hyperledger/firefly/mocks/syncasyncmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -84,7 +84,7 @@ func TestDefineFFIOk(t *testing.T) {
 	mim.On("ResolveInputSigningIdentity", context.Background(), mock.Anything).Return(nil)
 
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
 	mms.On("Send", context.Background()).Return(nil)
 
@@ -160,7 +160,7 @@ func TestDefineContractAPIOk(t *testing.T) {
 	mim.On("ResolveInputSigningIdentity", context.Background(), mock.Anything).Return(nil)
 
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
 	mms.On("Send", context.Background()).Return(nil)
 
diff --git a/internal/definitions/sender_datatype_test.go b/internal/definitions/sender_datatype_test.go
index bca20dfa6..d3d2a7f0d 100644
--- a/internal/definitions/sender_datatype_test.go
+++ b/internal/definitions/sender_datatype_test.go
@@ -25,7 +25,7 @@ import (
 	"github.com/hyperledger/firefly/mocks/broadcastmocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
 	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
+	"github.com/hyperledger/firefly/mocks/syncasyncmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -89,7 +89,7 @@ func TestBroadcastOk(t *testing.T) {
 	mdm := ds.data.(*datamocks.Manager)
 	mim := ds.identity.(*identitymanagermocks.Manager)
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 
 	mim.On("ResolveInputSigningIdentity", mock.Anything, mock.Anything).Return(nil)
 	mdm.On("CheckDatatype", mock.Anything, mock.Anything).Return(nil)
diff --git a/internal/definitions/sender_identity_test.go b/internal/definitions/sender_identity_test.go
index a239a2c9b..1f32ff4af 100644
--- a/internal/definitions/sender_identity_test.go
+++ b/internal/definitions/sender_identity_test.go
@@ -23,7 +23,7 @@ import (
 	"github.com/hyperledger/firefly/internal/identity"
 	"github.com/hyperledger/firefly/mocks/broadcastmocks"
 	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
+	"github.com/hyperledger/firefly/mocks/syncasyncmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -35,7 +35,7 @@ func TestClaimIdentity(t *testing.T) {
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 
 	mim.On("NormalizeSigningKey", mock.Anything, "0x1234", identity.KeyNormalizationBlockchainPlugin).Return("", nil)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
@@ -61,7 +61,7 @@ func TestClaimIdentityFail(t *testing.T) {
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 
 	mim.On("NormalizeSigningKey", mock.Anything, "0x1234", identity.KeyNormalizationBlockchainPlugin).Return("", nil)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
@@ -107,8 +107,8 @@ func TestClaimIdentityChild(t *testing.T) {
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms1 := &sysmessagingmocks.MessageSender{}
-	mms2 := &sysmessagingmocks.MessageSender{}
+	mms1 := &syncasyncmocks.Sender{}
+	mms2 := &syncasyncmocks.Sender{}
 
 	mim.On("NormalizeSigningKey", mock.Anything, "0x1234", identity.KeyNormalizationBlockchainPlugin).Return("", nil)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms1).Once()
@@ -141,8 +141,8 @@ func TestClaimIdentityChildFail(t *testing.T) {
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms1 := &sysmessagingmocks.MessageSender{}
-	mms2 := &sysmessagingmocks.MessageSender{}
+	mms1 := &syncasyncmocks.Sender{}
+	mms2 := &syncasyncmocks.Sender{}
 
 	mim.On("NormalizeSigningKey", mock.Anything, "0x1234", identity.KeyNormalizationBlockchainPlugin).Return("", nil)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms1).Once()
@@ -195,7 +195,7 @@ func TestUpdateIdentity(t *testing.T) {
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
 	mms.On("Send", mock.Anything).Return(nil)
diff --git a/internal/definitions/sender_test.go b/internal/definitions/sender_test.go
index 5ba920a12..d0def75b6 100644
--- a/internal/definitions/sender_test.go
+++ b/internal/definitions/sender_test.go
@@ -29,7 +29,7 @@ import (
 	"github.com/hyperledger/firefly/mocks/dataexchangemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
 	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
+	"github.com/hyperledger/firefly/mocks/syncasyncmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -70,7 +70,7 @@ func TestCreateDefinitionConfirm(t *testing.T) {
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 
 	mim.On("ResolveInputSigningIdentity", mock.Anything, mock.Anything).Return(nil)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
@@ -91,7 +91,7 @@ func TestCreateDatatypeDefinitionAsNodeConfirm(t *testing.T) {
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 
 	mim.On("ResolveInputSigningIdentity", mock.Anything, mock.Anything).Return(nil)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
diff --git a/internal/definitions/sender_tokenpool_test.go b/internal/definitions/sender_tokenpool_test.go
index 878f2e9a8..dc40392de 100644
--- a/internal/definitions/sender_tokenpool_test.go
+++ b/internal/definitions/sender_tokenpool_test.go
@@ -26,7 +26,7 @@ import (
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
 	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
+	"github.com/hyperledger/firefly/mocks/syncasyncmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -88,7 +88,7 @@ func TestDefineTokenPoolOk(t *testing.T) {
 	mdm := ds.data.(*datamocks.Manager)
 	mim := ds.identity.(*identitymanagermocks.Manager)
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 
 	pool := &core.TokenPoolAnnouncement{
 		Pool: &core.TokenPool{
@@ -151,7 +151,7 @@ func TestDefineTokenPoolBadName(t *testing.T) {
 
 	mim := ds.identity.(*identitymanagermocks.Manager)
 	mbm := ds.broadcast.(*broadcastmocks.Manager)
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 
 	pool := &core.TokenPoolAnnouncement{
 		Pool: &core.TokenPool{
diff --git a/internal/events/batch_pin_complete_test.go b/internal/events/batch_pin_complete_test.go
index fc7117c86..95019601f 100644
--- a/internal/events/batch_pin_complete_test.go
+++ b/internal/events/batch_pin_complete_test.go
@@ -620,6 +620,9 @@ func TestPersistBatchGoodDataUpsertOptimizeFail(t *testing.T) {
 	mdi.On("InsertDataArray", mock.Anything, mock.Anything).Return(fmt.Errorf("optimzation miss"))
 	mdi.On("UpsertData", mock.Anything, mock.Anything, database.UpsertOptimizationExisting).Return(fmt.Errorf("pop"))
 
+	mim := em.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	bp, valid, err := em.persistBatch(context.Background(), batch)
 	assert.Nil(t, bp)
 	assert.False(t, valid)
@@ -638,6 +641,9 @@ func TestPersistBatchGoodDataMessageFail(t *testing.T) {
 	mdi.On("InsertMessages", mock.Anything, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("optimzation miss"))
 	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationExisting, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("pop"))
 
+	mim := em.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	bp, valid, err := em.persistBatch(context.Background(), batch)
 	assert.False(t, valid)
 	assert.Nil(t, bp)
diff --git a/internal/events/dx_callbacks_test.go b/internal/events/dx_callbacks_test.go
index ad45bbd46..3a259c646 100644
--- a/internal/events/dx_callbacks_test.go
+++ b/internal/events/dx_callbacks_test.go
@@ -152,6 +152,8 @@ func TestPinnedReceiveOK(t *testing.T) {
 		Value: "peer1",
 	}).Return(node1, nil)
 	mim.On("CachedIdentityLookupMustExist", em.ctx, "signingOrg").Return(org1, false, nil)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
 	mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil, nil)
 	mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
@@ -580,6 +582,8 @@ func TestMessageReceiveMessagePersistMessageFail(t *testing.T) {
 		Value: "peer1",
 	}).Return(node1, nil)
 	mim.On("CachedIdentityLookupMustExist", em.ctx, "signingOrg").Return(org1, false, nil)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
 	mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil)
 	mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("optimization fail"))
@@ -616,6 +620,8 @@ func TestMessageReceiveMessagePersistDataFail(t *testing.T) {
 		Value: "peer1",
 	}).Return(node1, nil)
 	mim.On("CachedIdentityLookupMustExist", em.ctx, "signingOrg").Return(org1, false, nil)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
 	mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(fmt.Errorf("optimization miss"))
 	mdi.On("UpsertData", em.ctx, mock.Anything, database.UpsertOptimizationExisting).Return(fmt.Errorf("pop"))
@@ -651,6 +657,8 @@ func TestMessageReceiveUnpinnedBatchOk(t *testing.T) {
 		Value: "peer1",
 	}).Return(node1, nil)
 	mim.On("CachedIdentityLookupMustExist", em.ctx, "signingOrg").Return(org1, false, nil)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
 	mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil)
 	mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
@@ -692,6 +700,8 @@ func TestMessageReceiveUnpinnedBatchConfirmMessagesFail(t *testing.T) {
 		Value: "peer1",
 	}).Return(node1, nil)
 	mim.On("CachedIdentityLookupMustExist", em.ctx, "signingOrg").Return(org1, false, nil)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
 	mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil)
 	mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
@@ -733,6 +743,8 @@ func TestMessageReceiveUnpinnedBatchPersistEventFail(t *testing.T) {
 		Value: "peer1",
 	}).Return(node1, nil)
 	mim.On("CachedIdentityLookupMustExist", em.ctx, "signingOrg").Return(org1, false, nil)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
 	mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil)
 	mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
diff --git a/internal/events/event_dispatcher_test.go b/internal/events/event_dispatcher_test.go
index bce1a9158..22fb9b736 100644
--- a/internal/events/event_dispatcher_test.go
+++ b/internal/events/event_dispatcher_test.go
@@ -32,7 +32,7 @@ import (
 	"github.com/hyperledger/firefly/mocks/datamocks"
 	"github.com/hyperledger/firefly/mocks/eventsmocks"
 	"github.com/hyperledger/firefly/mocks/privatemessagingmocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
+	"github.com/hyperledger/firefly/mocks/syncasyncmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/events"
@@ -1031,7 +1031,7 @@ func TestEventDispatcherWithReply(t *testing.T) {
 		Namespace: "ns1",
 	}
 
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	mbm := ed.broadcast.(*broadcastmocks.Manager)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
 	mms.On("Send", mock.Anything).Return(nil)
diff --git a/internal/events/event_manager_test.go b/internal/events/event_manager_test.go
index 9ecd65a0a..5d816d41e 100644
--- a/internal/events/event_manager_test.go
+++ b/internal/events/event_manager_test.go
@@ -38,7 +38,6 @@ import (
 	"github.com/hyperledger/firefly/mocks/multipartymocks"
 	"github.com/hyperledger/firefly/mocks/privatemessagingmocks"
 	"github.com/hyperledger/firefly/mocks/shareddownloadmocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
 	"github.com/hyperledger/firefly/mocks/txcommonmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -49,6 +48,11 @@ import (
 )
 
 var testNodeID = fftypes.NewUUID()
+var testNode = &core.Identity{
+	IdentityBase: core.IdentityBase{
+		ID: testNodeID,
+	},
+}
 
 func newTestEventManager(t *testing.T) (*eventManager, func()) {
 	return newTestEventManagerCommon(t, false, false)
@@ -78,7 +82,6 @@ func newTestEventManagerCommon(t *testing.T, metrics, dbconcurrency bool) (*even
 	mbm := &broadcastmocks.Manager{}
 	mpm := &privatemessagingmocks.Manager{}
 	mam := &assetmocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
 	mdd := &shareddownloadmocks.Manager{}
 	mmi := &metricsmocks.Manager{}
 	mev := &eventsmocks.Plugin{}
@@ -89,14 +92,13 @@ func newTestEventManagerCommon(t *testing.T, metrics, dbconcurrency bool) (*even
 	if metrics {
 		mmi.On("TransferConfirmed", mock.Anything)
 	}
-	mni.On("GetNodeUUID", mock.Anything).Return(testNodeID).Maybe()
 	met.On("Name").Return("ut").Maybe()
 	mbi.On("VerifierType").Return(core.VerifierTypeEthAddress).Maybe()
 	mdi.On("Capabilities").Return(&database.Capabilities{Concurrency: dbconcurrency}).Maybe()
 	mev.On("SetHandler", "ns1", mock.Anything).Return(nil).Maybe()
 	mev.On("ValidateOptions", mock.Anything).Return(nil).Maybe()
 	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
-	emi, err := NewEventManager(ctx, ns, mni, mdi, mbi, mim, msh, mdm, mds, mbm, mpm, mam, mdd, mmi, txHelper, events, mmp)
+	emi, err := NewEventManager(ctx, ns, mdi, mbi, mim, msh, mdm, mds, mbm, mpm, mam, mdd, mmi, txHelper, events, mmp)
 	em := emi.(*eventManager)
 	em.txHelper = &txcommonmocks.Helper{}
 	mockRunAsGroupPassthrough(mdi)
@@ -131,7 +133,7 @@ func TestStartStop(t *testing.T) {
 }
 
 func TestStartStopBadDependencies(t *testing.T) {
-	_, err := NewEventManager(context.Background(), core.NamespaceRef{}, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	_, err := NewEventManager(context.Background(), core.NamespaceRef{}, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil, nil)
 	assert.Regexp(t, "FF10128", err)
 
 }
@@ -147,7 +149,6 @@ func TestStartStopEventListenerFail(t *testing.T) {
 	mds := &definitionsmocks.Sender{}
 	mbm := &broadcastmocks.Manager{}
 	mpm := &privatemessagingmocks.Manager{}
-	mni := &sysmessagingmocks.LocalNodeInfo{}
 	mam := &assetmocks.Manager{}
 	msd := &shareddownloadmocks.Manager{}
 	mm := &metricsmocks.Manager{}
@@ -159,7 +160,7 @@ func TestStartStopEventListenerFail(t *testing.T) {
 	mbi.On("VerifierType").Return(core.VerifierTypeEthAddress)
 	mev.On("SetHandler", "ns1", mock.Anything).Return(fmt.Errorf("pop"))
 	ns := core.NamespaceRef{LocalName: "ns1", RemoteName: "ns1"}
-	_, err := NewEventManager(context.Background(), ns, mni, mdi, mbi, mim, msh, mdm, mds, mbm, mpm, mam, msd, mm, txHelper, events, mmp)
+	_, err := NewEventManager(context.Background(), ns, mdi, mbi, mim, msh, mdm, mds, mbm, mpm, mam, msd, mm, txHelper, events, mmp)
 	assert.EqualError(t, err, "pop")
 }
 
diff --git a/internal/events/persist_batch_test.go b/internal/events/persist_batch_test.go
index d1bfb1a6e..89c9ecebe 100644
--- a/internal/events/persist_batch_test.go
+++ b/internal/events/persist_batch_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
+	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/stretchr/testify/assert"
@@ -158,11 +159,15 @@ func TestPersistBatchContentSendByUsOK(t *testing.T) {
 	mdm := em.data.(*datamocks.Manager)
 	mdm.On("GetMessageWithDataCached", em.ctx, batch.Payload.Messages[0].Header.ID).Return(batch.Payload.Messages[0], batch.Payload.Data, true, nil)
 
+	mim := em.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	ok, err := em.persistBatchContent(em.ctx, batch, []*messageAndData{})
 	assert.NoError(t, err)
 	assert.True(t, ok)
 
 	mdm.AssertExpectations(t)
+	mim.AssertExpectations(t)
 }
 
 func TestPersistBatchContentSentByNil(t *testing.T) {
@@ -180,11 +185,15 @@ func TestPersistBatchContentSentByNil(t *testing.T) {
 		args[2].(database.PostCompletionHook)()
 	})
 
+	mim := em.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(nil, nil)
+
 	ok, err := em.persistBatchContent(em.ctx, batch, []*messageAndData{})
 	assert.NoError(t, err)
 	assert.True(t, ok)
 
 	mdi.AssertExpectations(t)
+	mim.AssertExpectations(t)
 
 }
 
@@ -206,12 +215,16 @@ func TestPersistBatchContentSentByUsNotFoundFallback(t *testing.T) {
 		args[2].(database.PostCompletionHook)()
 	})
 
+	mim := em.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	ok, err := em.persistBatchContent(em.ctx, batch, []*messageAndData{})
 	assert.NoError(t, err)
 	assert.True(t, ok)
 
 	mdm.AssertExpectations(t)
 	mdi.AssertExpectations(t)
+	mim.AssertExpectations(t)
 
 }
 
@@ -236,12 +249,16 @@ func TestPersistBatchContentSentByUsFoundMismatch(t *testing.T) {
 	mdi.On("InsertMessages", mock.Anything, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("optimization miss"))
 	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationExisting, mock.AnythingOfType("database.PostCompletionHook")).Return(database.HashMismatch)
 
+	mim := em.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	ok, err := em.persistBatchContent(em.ctx, batch, []*messageAndData{})
 	assert.NoError(t, err)
 	assert.False(t, ok)
 
 	mdm.AssertExpectations(t)
 	mdi.AssertExpectations(t)
+	mim.AssertExpectations(t)
 
 }
 
@@ -267,12 +284,16 @@ func TestPersistBatchContentInsertMessagesFail(t *testing.T) {
 	}
 	mdm.On("UpdateMessageCache", msgData.message, msgData.data).Return()
 
+	mim := em.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	ok, err := em.persistBatchContent(em.ctx, batch, []*messageAndData{msgData})
 	assert.NoError(t, err)
 	assert.True(t, ok)
 
 	mdm.AssertExpectations(t)
 	mdi.AssertExpectations(t)
+	mim.AssertExpectations(t)
 }
 
 func TestPersistBatchContentSentByUsFoundError(t *testing.T) {
@@ -287,11 +308,15 @@ func TestPersistBatchContentSentByUsFoundError(t *testing.T) {
 	mdm := em.data.(*datamocks.Manager)
 	mdm.On("GetMessageWithDataCached", em.ctx, batch.Payload.Messages[0].Header.ID).Return(nil, nil, false, fmt.Errorf("pop"))
 
+	mim := em.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	ok, err := em.persistBatchContent(em.ctx, batch, []*messageAndData{})
 	assert.Regexp(t, "pop", err)
 	assert.False(t, ok)
 
 	mdm.AssertExpectations(t)
+	mim.AssertExpectations(t)
 
 }
 
@@ -307,12 +332,15 @@ func TestPersistBatchContentDataHashMismatch(t *testing.T) {
 	mdi.On("InsertDataArray", mock.Anything, mock.Anything).Return(fmt.Errorf("optimization miss"))
 	mdi.On("UpsertData", mock.Anything, mock.Anything, database.UpsertOptimizationExisting).Return(database.HashMismatch)
 
+	mim := em.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	ok, err := em.persistBatchContent(em.ctx, batch, []*messageAndData{})
 	assert.NoError(t, err)
 	assert.False(t, ok)
 
 	mdi.AssertExpectations(t)
-
+	mim.AssertExpectations(t)
 }
 
 func TestPersistBatchContentDataMissingBlobRef(t *testing.T) {
diff --git a/internal/events/reply_sender_test.go b/internal/events/reply_sender_test.go
index 46ee368ee..ccb4ba3a4 100644
--- a/internal/events/reply_sender_test.go
+++ b/internal/events/reply_sender_test.go
@@ -24,7 +24,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly/mocks/broadcastmocks"
 	"github.com/hyperledger/firefly/mocks/privatemessagingmocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
+	"github.com/hyperledger/firefly/mocks/syncasyncmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/mock"
 )
@@ -35,7 +35,7 @@ func TestSendReplyBroadcastFail(t *testing.T) {
 	})
 	defer cancel()
 
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	mbm := ed.broadcast.(*broadcastmocks.Manager)
 	mbm.On("NewBroadcast", mock.Anything).Return(mms)
 	mms.On("Send", context.Background()).Return(fmt.Errorf("pop"))
@@ -55,7 +55,7 @@ func TestSendReplyPrivateFail(t *testing.T) {
 	})
 	defer cancel()
 
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	mpm := ed.messaging.(*privatemessagingmocks.Manager)
 	mpm.On("NewMessage", mock.Anything).Return(mms)
 	mms.On("Send", context.Background()).Return(fmt.Errorf("pop"))
@@ -87,7 +87,7 @@ func TestSendReplyPrivateOk(t *testing.T) {
 		},
 	}
 
-	mms := &sysmessagingmocks.MessageSender{}
+	mms := &syncasyncmocks.Sender{}
 	mpm := ed.messaging.(*privatemessagingmocks.Manager)
 	mpm.On("NewMessage", mock.Anything).Return(mms)
 	mms.On("Send", context.Background()).Return(nil)
diff --git a/internal/events/ss_callbacks_test.go b/internal/events/ss_callbacks_test.go
index c04a61379..27eb0890c 100644
--- a/internal/events/ss_callbacks_test.go
+++ b/internal/events/ss_callbacks_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
+	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
 	"github.com/hyperledger/firefly/mocks/sharedstoragemocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -51,6 +52,9 @@ func TestSharedStorageBatchDownloadedOk(t *testing.T) {
 	mdm := em.data.(*datamocks.Manager)
 	mdm.On("UpdateMessageCache", mock.Anything, mock.Anything).Return()
 
+	mim := em.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+
 	bid, err := em.SharedStorageBatchDownloaded(mss, "payload1", b)
 	assert.NoError(t, err)
 	assert.Equal(t, batch.ID, bid)
@@ -61,6 +65,7 @@ func TestSharedStorageBatchDownloadedOk(t *testing.T) {
 	mdi.AssertExpectations(t)
 	mss.AssertExpectations(t)
 	mdm.AssertExpectations(t)
+	mim.AssertExpectations(t)
 
 }
 
diff --git a/internal/networkmap/register_identity_test.go b/internal/networkmap/register_identity_test.go
index b026bfbf9..fe2a5b9f1 100644
--- a/internal/networkmap/register_identity_test.go
+++ b/internal/networkmap/register_identity_test.go
@@ -86,7 +86,7 @@ func TestRegisterIdentityOrgWithParentWaitConfirmOk(t *testing.T) {
 		ctx := args[0].(context.Context)
 		id := args[1].(*fftypes.UUID)
 		assert.NotNil(t, id)
-		cb := args[2].(syncasync.RequestSender)
+		cb := args[2].(syncasync.SendFunction)
 		err := cb(ctx)
 		assert.NoError(t, err)
 	}).Return(nil, nil)
diff --git a/internal/orchestrator/status_test.go b/internal/orchestrator/status_test.go
index da5848339..607e0729a 100644
--- a/internal/orchestrator/status_test.go
+++ b/internal/orchestrator/status_test.go
@@ -135,9 +135,6 @@ func TestGetStatusRegistered(t *testing.T) {
 	assert.ElementsMatch(t, pluginsResult.SharedStorage, status.Plugins.SharedStorage)
 	assert.ElementsMatch(t, pluginsResult.Tokens, status.Plugins.Tokens)
 
-	assert.True(t, or.GetNodeUUID(or.ctx).Equals(nodeID))
-	assert.True(t, or.GetNodeUUID(or.ctx).Equals(nodeID)) // cached
-
 }
 
 func TestGetStatusVerifierLookupFail(t *testing.T) {
@@ -244,8 +241,6 @@ func TestGetStatusUnregistered(t *testing.T) {
 	assert.Equal(t, "node1", status.Node.Name)
 	assert.False(t, status.Node.Registered)
 
-	assert.Nil(t, or.GetNodeUUID(or.ctx))
-
 }
 
 func TestGetStatusOrgOnlyRegistered(t *testing.T) {
@@ -298,8 +293,6 @@ func TestGetStatusOrgOnlyRegistered(t *testing.T) {
 	assert.ElementsMatch(t, pluginsResult.Events, status.Plugins.Events)
 	assert.ElementsMatch(t, pluginsResult.SharedStorage, status.Plugins.SharedStorage)
 	assert.ElementsMatch(t, pluginsResult.Tokens, status.Plugins.Tokens)
-
-	assert.Nil(t, or.GetNodeUUID(or.ctx))
 }
 
 func TestGetStatusNodeError(t *testing.T) {
@@ -334,6 +327,4 @@ func TestGetStatusNodeError(t *testing.T) {
 	_, err := or.GetStatus(or.ctx)
 	assert.EqualError(t, err, "pop")
 
-	assert.Nil(t, or.GetNodeUUID(or.ctx))
-
 }
diff --git a/internal/privatemessaging/message_test.go b/internal/privatemessaging/message_test.go
index cafe5d2ea..950ba9ee8 100644
--- a/internal/privatemessaging/message_test.go
+++ b/internal/privatemessaging/message_test.go
@@ -103,7 +103,7 @@ func TestSendConfirmMessageE2EOk(t *testing.T) {
 	msa := pm.syncasync.(*syncasyncmocks.Bridge)
 	msa.On("WaitForMessage", pm.ctx, mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(pm.ctx)
 		}).
 		Return(retMsg, nil).Once()
@@ -601,7 +601,7 @@ func TestRequestReplySuccess(t *testing.T) {
 	msa := pm.syncasync.(*syncasyncmocks.Bridge)
 	msa.On("WaitForReply", pm.ctx, mock.Anything, mock.Anything).
 		Run(func(args mock.Arguments) {
-			send := args[2].(syncasync.RequestSender)
+			send := args[2].(syncasync.SendFunction)
 			send(pm.ctx)
 		}).
 		Return(nil, nil)
diff --git a/internal/syncasync/sync_async_bridge_test.go b/internal/syncasync/sync_async_bridge_test.go
index cc6dc1954..a623a42e5 100644
--- a/internal/syncasync/sync_async_bridge_test.go
+++ b/internal/syncasync/sync_async_bridge_test.go
@@ -25,7 +25,7 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
-	"github.com/hyperledger/firefly/mocks/sysmessagingmocks"
+	"github.com/hyperledger/firefly/mocks/systemeventmocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -35,7 +35,7 @@ func newTestSyncAsyncBridge(t *testing.T) (*syncAsyncBridge, func()) {
 	ctx, cancel := context.WithCancel(context.Background())
 	mdi := &databasemocks.Plugin{}
 	mdm := &datamocks.Manager{}
-	mse := &sysmessagingmocks.SystemEvents{}
+	mse := &systemeventmocks.EventInterface{}
 	sa := NewSyncAsyncBridge(ctx, "ns1", mdi, mdm)
 	sa.Init(mse)
 	return sa.(*syncAsyncBridge), cancel
@@ -50,7 +50,7 @@ func TestRequestReplyOk(t *testing.T) {
 	replyID := fftypes.NewUUID()
 	dataID := fftypes.NewUUID()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -104,7 +104,7 @@ func TestAwaitConfirmationOk(t *testing.T) {
 	requestID := fftypes.NewUUID()
 	dataID := fftypes.NewUUID()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -152,7 +152,7 @@ func TestAwaitConfirmationRejected(t *testing.T) {
 	requestID := fftypes.NewUUID()
 	dataID := fftypes.NewUUID()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -195,7 +195,7 @@ func TestRequestReplyTimeout(t *testing.T) {
 	sa, cancel := newTestSyncAsyncBridge(t)
 	cancel()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	_, err := sa.WaitForReply(sa.ctx, fftypes.NewUUID(), func(ctx context.Context) error {
@@ -209,7 +209,7 @@ func TestRequestSetupSystemListenerFail(t *testing.T) {
 	sa, cancel := newTestSyncAsyncBridge(t)
 	defer cancel()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(fmt.Errorf("pop"))
 
 	_, err := sa.WaitForReply(sa.ctx, fftypes.NewUUID(), func(ctx context.Context) error {
@@ -224,7 +224,7 @@ func TestEventCallbackNotInflight(t *testing.T) {
 	sa, cancel := newTestSyncAsyncBridge(t)
 	defer cancel()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	err := sa.eventCallback(&core.EventDelivery{
@@ -762,7 +762,7 @@ func TestAwaitTokenPoolConfirmation(t *testing.T) {
 
 	requestID := fftypes.NewUUID()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -802,7 +802,7 @@ func TestAwaitTokenPoolConfirmationSendFail(t *testing.T) {
 	sa, cancel := newTestSyncAsyncBridge(t)
 	defer cancel()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	_, err := sa.WaitForTokenPool(sa.ctx, fftypes.NewUUID(), func(ctx context.Context) error {
@@ -837,7 +837,7 @@ func TestAwaitTokenPoolConfirmationRejected(t *testing.T) {
 		},
 	}
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -870,7 +870,7 @@ func TestAwaitTokenTransferConfirmation(t *testing.T) {
 
 	requestID := fftypes.NewUUID()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -912,7 +912,7 @@ func TestAwaitTokenApprovalConfirmation(t *testing.T) {
 
 	requestID := fftypes.NewUUID()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -952,7 +952,7 @@ func TestAwaitTokenApprovalConfirmationSendFail(t *testing.T) {
 	sa, cancel := newTestSyncAsyncBridge(t)
 	defer cancel()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	_, err := sa.WaitForTokenApproval(sa.ctx, fftypes.NewUUID(), func(ctx context.Context) error {
@@ -966,7 +966,7 @@ func TestAwaitTokenTransferConfirmationSendFail(t *testing.T) {
 	sa, cancel := newTestSyncAsyncBridge(t)
 	defer cancel()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	_, err := sa.WaitForTokenTransfer(sa.ctx, fftypes.NewUUID(), func(ctx context.Context) error {
@@ -989,7 +989,7 @@ func TestAwaitFailedTokenPool(t *testing.T) {
 		Error: "pop",
 	}
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -1028,7 +1028,7 @@ func TestAwaitFailedTokenTransfer(t *testing.T) {
 		Error: "pop",
 	}
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -1067,7 +1067,7 @@ func TestAwaitFailedTokenApproval(t *testing.T) {
 		Error: "pop",
 	}
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -1380,7 +1380,7 @@ func TestAwaitIdentityConfirmed(t *testing.T) {
 		},
 	}
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -1412,7 +1412,7 @@ func TestAwaitIdentityFail(t *testing.T) {
 
 	requestID := fftypes.NewUUID()
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	_, err := sa.WaitForIdentity(sa.ctx, requestID, func(ctx context.Context) error {
@@ -1432,7 +1432,7 @@ func TestAwaitInvokeOpSucceeded(t *testing.T) {
 		Status: core.OpStatusSucceeded,
 	}
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -1471,7 +1471,7 @@ func TestAwaitInvokeOpSucceededLookupFail(t *testing.T) {
 		},
 	}
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -1502,7 +1502,7 @@ func TestAwaitInvokeOpFailed(t *testing.T) {
 		Error:  "pop",
 	}
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
@@ -1540,7 +1540,7 @@ func TestAwaitInvokeOpFailedLookupFail(t *testing.T) {
 		},
 	}
 
-	mse := sa.sysevents.(*sysmessagingmocks.SystemEvents)
+	mse := sa.sysevents.(*systemeventmocks.EventInterface)
 	mse.On("AddSystemEventListener", "ns1", mock.Anything).Return(nil)
 
 	mdi := sa.database.(*databasemocks.Plugin)
diff --git a/test/e2e/common.go b/test/e2e/common.go
deleted file mode 100644
index d975a44dc..000000000
--- a/test/e2e/common.go
+++ /dev/null
@@ -1,70 +0,0 @@
-// Copyright © 2022 Kaleido, Inc.
-//
-// SPDX-License-Identifier: Apache-2.0
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package e2e
-
-import (
-	"crypto/rand"
-	"fmt"
-	"io/ioutil"
-	"os"
-	"testing"
-
-	"github.com/go-resty/resty/v2"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"gopkg.in/yaml.v2"
-)
-
-func ReadConfig(t *testing.T, configFile string) map[string]interface{} {
-	yfile, err := ioutil.ReadFile(configFile)
-	assert.NoError(t, err)
-	data := make(map[string]interface{})
-	err = yaml.Unmarshal(yfile, &data)
-	assert.NoError(t, err)
-	return data
-}
-
-func WriteConfig(t *testing.T, configFile string, data map[string]interface{}) {
-	out, err := yaml.Marshal(data)
-	assert.NoError(t, err)
-	f, err := os.Create(configFile)
-	assert.NoError(t, err)
-	_, err = f.Write(out)
-	assert.NoError(t, err)
-	f.Close()
-}
-
-func AddNamespace(data map[string]interface{}, ns map[string]interface{}) {
-	namespaces := data["namespaces"].(map[interface{}]interface{})
-	predefined := namespaces["predefined"].([]interface{})
-	namespaces["predefined"] = append(predefined, ns)
-}
-
-func ResetFireFly(t *testing.T, client *resty.Client) {
-	resp, err := client.R().
-		SetBody(map[string]interface{}{}).
-		Post("/reset")
-	require.NoError(t, err)
-	assert.Equal(t, 204, resp.StatusCode())
-}
-
-func RandomName(t *testing.T) string {
-	b := make([]byte, 5)
-	_, err := rand.Read(b)
-	assert.NoError(t, err)
-	return fmt.Sprintf("e2e_%x", b)
-}
diff --git a/test/e2e/gateway/common.go b/test/e2e/gateway/common.go
index 9b707a288..8f1fef2b9 100644
--- a/test/e2e/gateway/common.go
+++ b/test/e2e/gateway/common.go
@@ -17,17 +17,18 @@
 package gateway
 
 import (
+	"crypto/rand"
 	"encoding/base64"
 	"fmt"
 	"net/http"
 	"os"
-	"path/filepath"
 	"testing"
 	"time"
 
 	"github.com/gorilla/websocket"
 	"github.com/hyperledger/firefly/test/e2e"
 	"github.com/hyperledger/firefly/test/e2e/client"
+	"github.com/stretchr/testify/assert"
 )
 
 const (
@@ -43,9 +44,6 @@ type testState struct {
 	client1              *client.FireFlyClient
 	unregisteredAccounts []interface{}
 	namespace            string
-	stackName            string
-	adminHost1           string
-	configFile1          string
 }
 
 func (m *testState) T() *testing.T {
@@ -61,10 +59,6 @@ func (m *testState) Done() func() {
 }
 
 func beforeE2ETest(t *testing.T) *testState {
-	stackDir := os.Getenv("STACK_DIR")
-	if stackDir == "" {
-		t.Fatal("STACK_DIR must be set")
-	}
 	stack := e2e.ReadStack(t)
 	stackState := e2e.ReadStackState(t)
 
@@ -80,9 +74,7 @@ func beforeE2ETest(t *testing.T) *testState {
 		member0WithPort = fmt.Sprintf(":%d", stack.Members[0].ExposedFireflyPort)
 	}
 
-	base1 := fmt.Sprintf("%s://%s%s", httpProtocolClient1, stack.Members[0].FireflyHostname, member0WithPort)
-	admin1 := fmt.Sprintf("%s://%s:%d", httpProtocolClient1, stack.Members[0].FireflyHostname, stack.Members[0].ExposedAdminPort)
-
+	baseURL := fmt.Sprintf("%s://%s%s", httpProtocolClient1, stack.Members[0].FireflyHostname, member0WithPort)
 	namespace := os.Getenv("NAMESPACE")
 	if namespace == "" {
 		namespace = "default"
@@ -90,13 +82,10 @@ func beforeE2ETest(t *testing.T) *testState {
 
 	ts := &testState{
 		t:                    t,
-		stackName:            stack.Name,
 		startTime:            time.Now(),
-		client1:              client.NewFireFly(t, base1, namespace),
+		client1:              client.NewFireFly(t, baseURL, namespace),
 		unregisteredAccounts: stackState.Accounts[2:],
 		namespace:            namespace,
-		adminHost1:           admin1,
-		configFile1:          filepath.Join(stackDir, "runtime", "config", "firefly_core_0.yml"),
 	}
 
 	t.Logf("Blockchain provider: %s", stack.BlockchainProvider)
@@ -128,3 +117,10 @@ func beforeE2ETest(t *testing.T) *testState {
 	}
 	return ts
 }
+
+func randomName(t *testing.T) string {
+	b := make([]byte, 5)
+	_, err := rand.Read(b)
+	assert.NoError(t, err)
+	return fmt.Sprintf("e2e_%x", b)
+}
diff --git a/test/e2e/gateway/tokens.go b/test/e2e/gateway/tokens.go
index 8e67a2711..81cb49a22 100644
--- a/test/e2e/gateway/tokens.go
+++ b/test/e2e/gateway/tokens.go
@@ -56,7 +56,7 @@ func (suite *TokensTestSuite) TestE2EFungibleTokensAsync() {
 
 	received1 := e2e.WsReader(suite.testState.ws1)
 
-	poolName := fmt.Sprintf("pool_%s", e2e.RandomName(suite.T()))
+	poolName := fmt.Sprintf("pool_%s", randomName(suite.T()))
 	suite.T().Logf("Pool name: %s", poolName)
 
 	pool := &core.TokenPool{
diff --git a/test/e2e/gateway/tokens_only.go b/test/e2e/gateway/tokens_only.go
deleted file mode 100644
index 3b49060ad..000000000
--- a/test/e2e/gateway/tokens_only.go
+++ /dev/null
@@ -1,128 +0,0 @@
-// Copyright © 2022 Kaleido, Inc.
-//
-// SPDX-License-Identifier: Apache-2.0
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package gateway
-
-import (
-	"fmt"
-
-	"github.com/hyperledger/firefly-common/pkg/fftypes"
-	"github.com/hyperledger/firefly/pkg/core"
-	"github.com/hyperledger/firefly/test/e2e"
-	"github.com/hyperledger/firefly/test/e2e/client"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/suite"
-)
-
-type TokensOnlyTestSuite struct {
-	suite.Suite
-	testState *testState
-	connector string
-	db        string
-	key       string
-}
-
-func (suite *TokensOnlyTestSuite) SetupSuite() {
-	suite.testState = beforeE2ETest(suite.T())
-	stack := e2e.ReadStack(suite.T())
-	stackState := e2e.ReadStackState(suite.T())
-
-	suite.connector = stack.TokenProviders[0]
-	suite.db = stack.Database
-	account := stackState.Accounts[0].(map[string]interface{})
-	suite.key = account["address"].(string)
-}
-
-func (suite *TokensOnlyTestSuite) BeforeTest(suiteName, testName string) {
-	suite.testState = beforeE2ETest(suite.T())
-}
-
-func (suite *TokensOnlyTestSuite) AfterTest(suiteName, testName string) {
-	e2e.VerifyAllOperationsSucceeded(suite.T(), []*client.FireFlyClient{suite.testState.client1}, suite.testState.startTime)
-}
-
-func (suite *TokensOnlyTestSuite) TestTokensOnlyNamespaces() {
-	defer suite.testState.Done()
-	testNamespace := e2e.RandomName(suite.T())
-	suite.T().Logf("Test namespace: %s", testNamespace)
-
-	namespaceInfo := map[string]interface{}{
-		"name": testNamespace,
-		"asset": map[string]interface{}{
-			"manager": map[string]interface{}{
-				"keynormalization": "none",
-			},
-		},
-		"plugins": []string{"database0", suite.connector},
-	}
-
-	// Add the new namespace
-	data1 := e2e.ReadConfig(suite.T(), suite.testState.configFile1)
-	e2e.AddNamespace(data1, namespaceInfo)
-	e2e.WriteConfig(suite.T(), suite.testState.configFile1, data1)
-
-	admin1 := client.NewResty(suite.T())
-	admin1.SetBaseURL(suite.testState.adminHost1 + "/spi/v1")
-
-	e2e.ResetFireFly(suite.T(), admin1)
-	e2e.PollForUp(suite.T(), suite.testState.client1)
-
-	client1 := client.NewFireFly(suite.T(), suite.testState.client1.Hostname, testNamespace)
-	eventNames := "token_pool_confirmed|token_transfer_confirmed"
-	queryString := fmt.Sprintf("namespace=%s&ephemeral&autoack&filter.events=%s", testNamespace, eventNames)
-	received1 := e2e.WsReader(client1.WebSocket(suite.T(), queryString, nil))
-
-	// Attempt async token operations on new namespace
-	poolName := fmt.Sprintf("pool_%s", e2e.RandomName(suite.T()))
-	suite.T().Logf("Pool name: %s", poolName)
-
-	pool := &core.TokenPool{
-		Name: poolName,
-		Key:  suite.key,
-		Type: core.TokenTypeFungible,
-	}
-	poolResp := client1.CreateTokenPool(suite.T(), pool, false)
-	poolID := poolResp.ID
-
-	e2e.WaitForEvent(suite.T(), received1, core.EventTypePoolConfirmed, poolID)
-	pools := client1.GetTokenPools(suite.T(), suite.testState.startTime)
-	assert.Equal(suite.T(), 1, len(pools))
-	assert.Equal(suite.T(), testNamespace, pools[0].Namespace)
-	assert.Equal(suite.T(), suite.connector, pools[0].Connector)
-	assert.Equal(suite.T(), poolName, pools[0].Name)
-	assert.Equal(suite.T(), core.TokenTypeFungible, pools[0].Type)
-	assert.NotEmpty(suite.T(), pools[0].Locator)
-
-	transfer := &core.TokenTransferInput{
-		TokenTransfer: core.TokenTransfer{Amount: *fftypes.NewFFBigInt(1), Key: suite.key},
-		Pool:          poolName,
-	}
-	transferOut := client1.MintTokens(suite.T(), transfer, false)
-	e2e.WaitForEvent(suite.T(), received1, core.EventTypeTransferConfirmed, transferOut.LocalID)
-	e2e.ValidateAccountBalances(suite.T(), client1, poolID, "", map[string]int64{
-		suite.key: 1,
-	})
-
-	transfer = &core.TokenTransferInput{
-		TokenTransfer: core.TokenTransfer{Amount: *fftypes.NewFFBigInt(1), Key: suite.key},
-		Pool:          poolName,
-	}
-	transferOut = client1.BurnTokens(suite.T(), transfer, false)
-	e2e.WaitForEvent(suite.T(), received1, core.EventTypeTransferConfirmed, transferOut.LocalID)
-	e2e.ValidateAccountBalances(suite.T(), client1, poolID, "", map[string]int64{
-		suite.key: 0,
-	})
-}
diff --git a/test/e2e/multiparty/common.go b/test/e2e/multiparty/common.go
index c890cb28c..d1197a71a 100644
--- a/test/e2e/multiparty/common.go
+++ b/test/e2e/multiparty/common.go
@@ -18,21 +18,26 @@ package multiparty
 
 import (
 	"context"
+	"crypto/rand"
 	"crypto/sha256"
 	"encoding/base64"
 	"fmt"
+	"io/ioutil"
 	"net/http"
 	"os"
 	"path/filepath"
 	"testing"
 	"time"
 
+	"github.com/go-resty/resty/v2"
 	"github.com/gorilla/websocket"
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/test/e2e"
 	"github.com/hyperledger/firefly/test/e2e/client"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/yaml.v2"
 )
 
 const (
@@ -251,3 +256,43 @@ func validateReceivedMessages(ts *testState, client *client.FireFlyClient, topic
 	// Flip data (returned in most recent order) into delivery order
 	return returnData
 }
+
+func randomName(t *testing.T) string {
+	b := make([]byte, 5)
+	_, err := rand.Read(b)
+	assert.NoError(t, err)
+	return fmt.Sprintf("e2e_%x", b)
+}
+
+func readConfig(t *testing.T, configFile string) map[string]interface{} {
+	yfile, err := ioutil.ReadFile(configFile)
+	assert.NoError(t, err)
+	data := make(map[string]interface{})
+	err = yaml.Unmarshal(yfile, &data)
+	assert.NoError(t, err)
+	return data
+}
+
+func writeConfig(t *testing.T, configFile string, data map[string]interface{}) {
+	out, err := yaml.Marshal(data)
+	assert.NoError(t, err)
+	f, err := os.Create(configFile)
+	assert.NoError(t, err)
+	_, err = f.Write(out)
+	assert.NoError(t, err)
+	f.Close()
+}
+
+func addNamespace(data map[string]interface{}, ns map[string]interface{}) {
+	namespaces := data["namespaces"].(map[interface{}]interface{})
+	predefined := namespaces["predefined"].([]interface{})
+	namespaces["predefined"] = append(predefined, ns)
+}
+
+func resetFireFly(t *testing.T, client *resty.Client) {
+	resp, err := client.R().
+		SetBody(map[string]interface{}{}).
+		Post("/reset")
+	require.NoError(t, err)
+	assert.Equal(t, 204, resp.StatusCode())
+}
diff --git a/test/e2e/multiparty/contract_migration.go b/test/e2e/multiparty/contract_migration.go
index 3205f768d..d9d1ac114 100644
--- a/test/e2e/multiparty/contract_migration.go
+++ b/test/e2e/multiparty/contract_migration.go
@@ -55,7 +55,7 @@ func (suite *ContractMigrationTestSuite) TestContractMigration() {
 }
 
 func runMigrationTest(suite *ContractMigrationTestSuite, address1, address2 string, startOnV1 bool) {
-	testNamespace := e2e.RandomName(suite.T())
+	testNamespace := randomName(suite.T())
 	suite.T().Logf("Test namespace: %s", testNamespace)
 
 	org := map[string]interface{}{}
@@ -77,17 +77,17 @@ func runMigrationTest(suite *ContractMigrationTestSuite, address1, address2 stri
 	data := &core.DataRefOrValue{Value: fftypes.JSONAnyPtr(`"test"`)}
 
 	// Add the new namespace to both config files
-	data1 := e2e.ReadConfig(suite.T(), suite.testState.configFile1)
+	data1 := readConfig(suite.T(), suite.testState.configFile1)
 	org["name"] = suite.testState.org1.Name
 	org["key"] = suite.testState.org1key.Value
-	e2e.AddNamespace(data1, namespaceInfo)
-	e2e.WriteConfig(suite.T(), suite.testState.configFile1, data1)
+	addNamespace(data1, namespaceInfo)
+	writeConfig(suite.T(), suite.testState.configFile1, data1)
 
-	data2 := e2e.ReadConfig(suite.T(), suite.testState.configFile2)
+	data2 := readConfig(suite.T(), suite.testState.configFile2)
 	org["name"] = suite.testState.org2.Name
 	org["key"] = suite.testState.org2key.Value
-	e2e.AddNamespace(data2, namespaceInfo)
-	e2e.WriteConfig(suite.T(), suite.testState.configFile2, data2)
+	addNamespace(data2, namespaceInfo)
+	writeConfig(suite.T(), suite.testState.configFile2, data2)
 
 	admin1 := client.NewResty(suite.T())
 	admin2 := client.NewResty(suite.T())
@@ -95,8 +95,8 @@ func runMigrationTest(suite *ContractMigrationTestSuite, address1, address2 stri
 	admin2.SetBaseURL(suite.testState.adminHost2 + "/spi/v1")
 
 	// Reset both nodes to pick up the new namespace
-	e2e.ResetFireFly(suite.T(), admin1)
-	e2e.ResetFireFly(suite.T(), admin2)
+	resetFireFly(suite.T(), admin1)
+	resetFireFly(suite.T(), admin2)
 	e2e.PollForUp(suite.T(), suite.testState.client1)
 	e2e.PollForUp(suite.T(), suite.testState.client2)
 
diff --git a/test/e2e/multiparty/namespace_alias.go b/test/e2e/multiparty/namespace_alias.go
index c203221e2..7ba36a7fc 100644
--- a/test/e2e/multiparty/namespace_alias.go
+++ b/test/e2e/multiparty/namespace_alias.go
@@ -49,9 +49,9 @@ func (suite *NamespaceAliasSuite) TestNamespaceMapping() {
 	defer suite.testState.Done()
 
 	address := deployContract(suite.T(), suite.testState.stackName, "firefly/Firefly.json")
-	localNamespace1 := e2e.RandomName(suite.T())
-	localNamespace2 := e2e.RandomName(suite.T())
-	remoteNamespace := e2e.RandomName(suite.T())
+	localNamespace1 := randomName(suite.T())
+	localNamespace2 := randomName(suite.T())
+	remoteNamespace := randomName(suite.T())
 	suite.T().Logf("Test namespace: local1=%s, local2=%s, remote=%s", localNamespace1, localNamespace2, remoteNamespace)
 
 	org := map[string]interface{}{}
@@ -70,19 +70,19 @@ func (suite *NamespaceAliasSuite) TestNamespaceMapping() {
 	data := &core.DataRefOrValue{Value: fftypes.JSONAnyPtr(`"test"`)}
 
 	// Add the new namespace to both config files
-	data1 := e2e.ReadConfig(suite.T(), suite.testState.configFile1)
+	data1 := readConfig(suite.T(), suite.testState.configFile1)
 	namespaceInfo["name"] = localNamespace1
 	org["name"] = suite.testState.org1.Name
 	org["key"] = suite.testState.org1key.Value
-	e2e.AddNamespace(data1, namespaceInfo)
-	e2e.WriteConfig(suite.T(), suite.testState.configFile1, data1)
+	addNamespace(data1, namespaceInfo)
+	writeConfig(suite.T(), suite.testState.configFile1, data1)
 
-	data2 := e2e.ReadConfig(suite.T(), suite.testState.configFile2)
+	data2 := readConfig(suite.T(), suite.testState.configFile2)
 	namespaceInfo["name"] = localNamespace2
 	org["name"] = suite.testState.org2.Name
 	org["key"] = suite.testState.org2key.Value
-	e2e.AddNamespace(data2, namespaceInfo)
-	e2e.WriteConfig(suite.T(), suite.testState.configFile2, data2)
+	addNamespace(data2, namespaceInfo)
+	writeConfig(suite.T(), suite.testState.configFile2, data2)
 
 	admin1 := client.NewResty(suite.T())
 	admin2 := client.NewResty(suite.T())
@@ -90,8 +90,8 @@ func (suite *NamespaceAliasSuite) TestNamespaceMapping() {
 	admin2.SetBaseURL(suite.testState.adminHost2 + "/spi/v1")
 
 	// Reset both nodes to pick up the new namespace
-	e2e.ResetFireFly(suite.T(), admin1)
-	e2e.ResetFireFly(suite.T(), admin2)
+	resetFireFly(suite.T(), admin1)
+	resetFireFly(suite.T(), admin2)
 	e2e.PollForUp(suite.T(), suite.testState.client1)
 	e2e.PollForUp(suite.T(), suite.testState.client2)
 
diff --git a/test/e2e/multiparty/tokens.go b/test/e2e/multiparty/tokens.go
index e0d4e9ba4..af02d12f2 100644
--- a/test/e2e/multiparty/tokens.go
+++ b/test/e2e/multiparty/tokens.go
@@ -53,7 +53,7 @@ func (suite *TokensTestSuite) TestE2EFungibleTokensAsync() {
 	received1 := e2e.WsReader(suite.testState.ws1)
 	received2 := e2e.WsReader(suite.testState.ws2)
 
-	poolName := fmt.Sprintf("pool_%s", e2e.RandomName(suite.T()))
+	poolName := fmt.Sprintf("pool_%s", randomName(suite.T()))
 	suite.T().Logf("Pool name: %s", poolName)
 
 	pool := &core.TokenPool{
@@ -219,7 +219,7 @@ func (suite *TokensTestSuite) TestE2ENonFungibleTokensSync() {
 	received1 := e2e.WsReader(suite.testState.ws1)
 	received2 := e2e.WsReader(suite.testState.ws2)
 
-	poolName := fmt.Sprintf("pool_%s", e2e.RandomName(suite.T()))
+	poolName := fmt.Sprintf("pool_%s", randomName(suite.T()))
 	suite.T().Logf("Pool name: %s", poolName)
 
 	pool := &core.TokenPool{
diff --git a/test/e2e/runners/ethereum_gateway_test.go b/test/e2e/runners/ethereum_gateway_test.go
index 91afbd057..d0f53cfbb 100644
--- a/test/e2e/runners/ethereum_gateway_test.go
+++ b/test/e2e/runners/ethereum_gateway_test.go
@@ -26,5 +26,4 @@ import (
 func TestEthereumGatewayE2ESuite(t *testing.T) {
 	suite.Run(t, new(gateway.TokensTestSuite))
 	suite.Run(t, new(gateway.EthereumContractTestSuite))
-	suite.Run(t, new(gateway.TokensOnlyTestSuite))
 }
diff --git a/test/e2e/stack.go b/test/e2e/stack.go
index 74f322511..1d9392194 100644
--- a/test/e2e/stack.go
+++ b/test/e2e/stack.go
@@ -27,7 +27,6 @@ type Stack struct {
 	BlockchainProvider    string    `json:"blockchainProvider"`
 	TokenProviders        []string  `json:"tokenProviders"`
 	Members               []*Member `json:"members,omitempty"`
-	Database              string    `json:"database"`
 }
 
 type StackState struct {