Repository: hyperledger/firefly
Pull request #880 (2022-06-24 15:42:17)

Collapse batchpin.Submitter into multiparty.Manager


Patch:
diff --git a/Makefile b/Makefile
index 89091bdc3..cce56c32e 100644
--- a/Makefile
+++ b/Makefile
@@ -52,7 +52,6 @@ $(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/batchpin,         Submitter,          batchpinmocks))
 $(eval $(call makemock, internal/sysmessaging,     SystemEvents,       sysmessagingmocks))
 $(eval $(call makemock, internal/sysmessaging,     MessageSender,      sysmessagingmocks))
 $(eval $(call makemock, internal/sysmessaging,     LocalNodeInfo,      sysmessagingmocks))
@@ -101,4 +100,4 @@ manifest:
 docker:
 		./docker_build.sh $(DOCKER_ARGS)
 docs: .ALWAYS
-		cd docs && bundle install && bundle exec jekyll build && bundle exec htmlproofer --disable-external --allow-hash-href --assume-extension ./_site --url-swap '^/firefly/:/' --url-ignore /127.0.0.1/,/localhost/
\ No newline at end of file
+		cd docs && bundle install && bundle exec jekyll build && bundle exec htmlproofer --disable-external --allow-hash-href --assume-extension ./_site --url-swap '^/firefly/:/' --url-ignore /127.0.0.1/,/localhost/
diff --git a/internal/batchpin/batchpin.go b/internal/batchpin/batchpin.go
deleted file mode 100644
index 247e23dbf..000000000
--- a/internal/batchpin/batchpin.go
+++ /dev/null
@@ -1,91 +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 batchpin
-
-import (
-	"context"
-
-	"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/identity"
-	"github.com/hyperledger/firefly/internal/metrics"
-	"github.com/hyperledger/firefly/internal/multiparty"
-	"github.com/hyperledger/firefly/internal/operations"
-	"github.com/hyperledger/firefly/pkg/core"
-	"github.com/hyperledger/firefly/pkg/database"
-)
-
-type Submitter interface {
-	core.Named
-
-	SubmitPinnedBatch(ctx context.Context, batch *core.BatchPersisted, contexts []*fftypes.Bytes32, payloadRef string) error
-
-	// From operations.OperationHandler
-	PrepareOperation(ctx context.Context, op *core.Operation) (*core.PreparedOperation, error)
-	RunOperation(ctx context.Context, op *core.PreparedOperation) (outputs fftypes.JSONObject, complete bool, err error)
-}
-
-type batchPinSubmitter struct {
-	namespace  string
-	database   database.Plugin
-	identity   identity.Manager
-	multiparty multiparty.Manager
-	metrics    metrics.Manager
-	operations operations.Manager
-}
-
-func NewBatchPinSubmitter(ctx context.Context, ns string, di database.Plugin, im identity.Manager, multiparty multiparty.Manager, mm metrics.Manager, om operations.Manager) (Submitter, error) {
-	if di == nil || im == nil || mm == nil || om == nil {
-		return nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "BatchPinSubmitter")
-	}
-	bp := &batchPinSubmitter{
-		namespace:  ns,
-		database:   di,
-		identity:   im,
-		multiparty: multiparty,
-		metrics:    mm,
-		operations: om,
-	}
-	om.RegisterHandler(ctx, bp, []core.OpType{
-		core.OpTypeBlockchainPinBatch,
-	})
-	return bp, nil
-}
-
-func (bp *batchPinSubmitter) Name() string {
-	return "BatchPinSubmitter"
-}
-
-func (bp *batchPinSubmitter) SubmitPinnedBatch(ctx context.Context, batch *core.BatchPersisted, contexts []*fftypes.Bytes32, payloadRef string) error {
-	// The pending blockchain transaction
-	op := core.NewOperation(
-		bp.multiparty,
-		batch.Namespace,
-		batch.TX.ID,
-		core.OpTypeBlockchainPinBatch)
-	addBatchPinInputs(op, batch.ID, contexts, payloadRef)
-	if err := bp.operations.AddOrReuseOperation(ctx, op); err != nil {
-		return err
-	}
-
-	if bp.metrics.IsMetricsEnabled() {
-		bp.metrics.CountBatchPin()
-	}
-	_, err := bp.operations.RunOperation(ctx, opBatchPin(op, batch, contexts, payloadRef))
-	return err
-}
diff --git a/internal/broadcast/manager.go b/internal/broadcast/manager.go
index 13dc0c7de..e71aaf238 100644
--- a/internal/broadcast/manager.go
+++ b/internal/broadcast/manager.go
@@ -24,12 +24,12 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/batch"
-	"github.com/hyperledger/firefly/internal/batchpin"
 	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/data"
 	"github.com/hyperledger/firefly/internal/identity"
 	"github.com/hyperledger/firefly/internal/metrics"
+	"github.com/hyperledger/firefly/internal/multiparty"
 	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/internal/syncasync"
 	"github.com/hyperledger/firefly/internal/sysmessaging"
@@ -70,14 +70,14 @@ type broadcastManager struct {
 	exchange              dataexchange.Plugin
 	sharedstorage         sharedstorage.Plugin
 	syncasync             syncasync.Bridge
-	batchpin              batchpin.Submitter
+	multiparty            multiparty.Manager
 	maxBatchPayloadLength int64
 	metrics               metrics.Manager
 	operations            operations.Manager
 }
 
-func NewBroadcastManager(ctx context.Context, ns string, di database.Plugin, bi blockchain.Plugin, dx dataexchange.Plugin, si sharedstorage.Plugin, im identity.Manager, dm data.Manager, ba batch.Manager, sa syncasync.Bridge, bp batchpin.Submitter, mm metrics.Manager, om operations.Manager) (Manager, error) {
-	if di == nil || im == nil || dm == nil || bi == nil || dx == nil || si == nil || ba == nil || mm == nil || om == nil {
+func NewBroadcastManager(ctx context.Context, ns string, di database.Plugin, bi blockchain.Plugin, dx dataexchange.Plugin, si sharedstorage.Plugin, im identity.Manager, dm data.Manager, ba batch.Manager, sa syncasync.Bridge, mult multiparty.Manager, mm metrics.Manager, om operations.Manager) (Manager, error) {
+	if di == nil || im == nil || dm == nil || bi == nil || dx == nil || si == nil || ba == nil || mm == nil || om == nil || mult == nil {
 		return nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "BroadcastManager")
 	}
 	bm := &broadcastManager{
@@ -90,7 +90,7 @@ func NewBroadcastManager(ctx context.Context, ns string, di database.Plugin, bi
 		exchange:              dx,
 		sharedstorage:         si,
 		syncasync:             sa,
-		batchpin:              bp,
+		multiparty:            mult,
 		maxBatchPayloadLength: config.GetByteSize(coreconfig.BroadcastBatchPayloadLimit),
 		metrics:               mm,
 		operations:            om,
@@ -152,7 +152,7 @@ func (bm *broadcastManager) dispatchBatch(ctx context.Context, state *batch.Disp
 	}
 	payloadRef := outputs.GetString("payloadRef")
 	log.L(ctx).Infof("Pinning broadcast batch %s with author=%s key=%s payloadRef=%s", batch.ID, batch.Author, batch.Key, payloadRef)
-	return bm.batchpin.SubmitPinnedBatch(ctx, &state.Persisted, state.Pins, payloadRef)
+	return bm.multiparty.SubmitBatchPin(ctx, &state.Persisted, state.Pins, payloadRef)
 }
 
 func (bm *broadcastManager) uploadBlobs(ctx context.Context, tx *fftypes.UUID, data core.DataArray) error {
diff --git a/internal/database/postgres/postgres.go b/internal/database/postgres/postgres.go
index 9af8c78e1..0277d1301 100644
--- a/internal/database/postgres/postgres.go
+++ b/internal/database/postgres/postgres.go
@@ -42,8 +42,8 @@ func (psql *Postgres) Init(ctx context.Context, config config.Section) error {
 	return psql.SQLCommon.Init(ctx, psql, config, capabilities)
 }
 
-func (psql *Postgres) SetHandler(namespace string, handler database.Callbacks) {
-	psql.SQLCommon.SetHandler(namespace, handler)
+func (psql *Postgres) SetHandler(handler database.Callbacks) {
+	psql.SQLCommon.SetHandler(handler)
 }
 
 func (psql *Postgres) Name() string {
diff --git a/internal/database/sqlcommon/sqlcommon.go b/internal/database/sqlcommon/sqlcommon.go
index 654796887..311748277 100644
--- a/internal/database/sqlcommon/sqlcommon.go
+++ b/internal/database/sqlcommon/sqlcommon.go
@@ -45,47 +45,35 @@ type SQLCommon struct {
 }
 
 type callbacks struct {
-	handlers map[string]database.Callbacks
+	handlers []database.Callbacks
 }
 
 func (cb *callbacks) OrderedUUIDCollectionNSEvent(resType database.OrderedUUIDCollectionNS, eventType core.ChangeEventType, ns string, id *fftypes.UUID, sequence int64) {
-	if cb, ok := cb.handlers[ns]; ok {
-		cb.OrderedUUIDCollectionNSEvent(resType, eventType, ns, id, sequence)
-	}
-	if cb, ok := cb.handlers[database.GlobalHandler]; ok {
+	for _, cb := range cb.handlers {
 		cb.OrderedUUIDCollectionNSEvent(resType, eventType, ns, id, sequence)
 	}
 }
 
 func (cb *callbacks) OrderedCollectionNSEvent(resType database.OrderedCollectionNS, eventType core.ChangeEventType, ns string, sequence int64) {
-	if cb, ok := cb.handlers[ns]; ok {
-		cb.OrderedCollectionNSEvent(resType, eventType, ns, sequence)
-	}
-	if cb, ok := cb.handlers[database.GlobalHandler]; ok {
+	for _, cb := range cb.handlers {
 		cb.OrderedCollectionNSEvent(resType, eventType, ns, sequence)
 	}
 }
 
 func (cb *callbacks) UUIDCollectionNSEvent(resType database.UUIDCollectionNS, eventType core.ChangeEventType, ns string, id *fftypes.UUID) {
-	if cb, ok := cb.handlers[ns]; ok {
-		cb.UUIDCollectionNSEvent(resType, eventType, ns, id)
-	}
-	if cb, ok := cb.handlers[database.GlobalHandler]; ok {
+	for _, cb := range cb.handlers {
 		cb.UUIDCollectionNSEvent(resType, eventType, ns, id)
 	}
 }
 
 func (cb *callbacks) UUIDCollectionEvent(resType database.UUIDCollection, eventType core.ChangeEventType, id *fftypes.UUID) {
-	if cb, ok := cb.handlers[database.GlobalHandler]; ok {
+	for _, cb := range cb.handlers {
 		cb.UUIDCollectionEvent(resType, eventType, id)
 	}
 }
 
 func (cb *callbacks) HashCollectionNSEvent(resType database.HashCollectionNS, eventType core.ChangeEventType, ns string, hash *fftypes.Bytes32) {
-	if cb, ok := cb.handlers[ns]; ok {
-		cb.HashCollectionNSEvent(resType, eventType, ns, hash)
-	}
-	if cb, ok := cb.handlers[database.GlobalHandler]; ok {
+	for _, cb := range cb.handlers {
 		cb.HashCollectionNSEvent(resType, eventType, ns, hash)
 	}
 }
@@ -142,11 +130,8 @@ func (s *SQLCommon) Init(ctx context.Context, provider Provider, config config.S
 	return nil
 }
 
-func (s *SQLCommon) SetHandler(namespace string, handler database.Callbacks) {
-	if s.callbacks.handlers == nil {
-		s.callbacks.handlers = make(map[string]database.Callbacks)
-	}
-	s.callbacks.handlers[namespace] = handler
+func (s *SQLCommon) SetHandler(handler database.Callbacks) {
+	s.callbacks.handlers = append(s.callbacks.handlers, handler)
 }
 
 func (s *SQLCommon) Capabilities() *database.Capabilities { return s.capabilities }
diff --git a/internal/database/sqlite3/sqlite3.go b/internal/database/sqlite3/sqlite3.go
index ecbbaa46a..a8402cb90 100644
--- a/internal/database/sqlite3/sqlite3.go
+++ b/internal/database/sqlite3/sqlite3.go
@@ -58,8 +58,8 @@ func (sqlite *SQLite3) Init(ctx context.Context, config config.Section) error {
 	return sqlite.SQLCommon.Init(ctx, sqlite, config, capabilities)
 }
 
-func (sqlite *SQLite3) SetHandler(namespace string, handler database.Callbacks) {
-	sqlite.SQLCommon.SetHandler(namespace, handler)
+func (sqlite *SQLite3) SetHandler(handler database.Callbacks) {
+	sqlite.SQLCommon.SetHandler(handler)
 }
 
 func (sqlite *SQLite3) Name() string {
diff --git a/internal/dataexchange/ffdx/dxevent.go b/internal/dataexchange/ffdx/dxevent.go
index e23a1ab65..8020e3c73 100644
--- a/internal/dataexchange/ffdx/dxevent.go
+++ b/internal/dataexchange/ffdx/dxevent.go
@@ -17,8 +17,7 @@
 package ffdx
 
 import (
-	"encoding/json"
-	"fmt"
+	"strings"
 
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/i18n"
@@ -93,7 +92,6 @@ func (e *dxEvent) TransferResult() *dataexchange.TransferResult {
 }
 
 func (h *FFDX) dispatchEvent(msg *wsEvent) {
-	var namespace string
 	var err error
 	e := &dxEvent{ffdx: h, id: msg.EventID, requestID: msg.RequestID}
 	switch msg.Type {
@@ -131,21 +129,10 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 			},
 		}
 	case messageReceived:
-		// De-serialize the transport wrapper
-		var wrapper *core.TransportWrapper
-		err = json.Unmarshal([]byte(msg.Message), &wrapper)
-		switch {
-		case err != nil:
-			err = fmt.Errorf("invalid transmission from peer '%s': %s", msg.Sender, err)
-		case wrapper.Batch == nil:
-			err = fmt.Errorf("invalid transmission from peer '%s': nil batch", msg.Sender)
-		default:
-			namespace = wrapper.Batch.Namespace
-			e.dxType = dataexchange.DXEventTypeMessageReceived
-			e.messageReceived = &dataexchange.MessageReceived{
-				PeerID:    msg.Sender,
-				Transport: wrapper,
-			}
+		e.dxType = dataexchange.DXEventTypeMessageReceived
+		e.messageReceived = &dataexchange.MessageReceived{
+			PeerID: msg.Sender,
+			Data:   []byte(msg.Message),
 		}
 	case blobFailed:
 		e.dxType = dataexchange.DXEventTypeTransferResult
@@ -174,10 +161,14 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 		var hash *fftypes.Bytes32
 		hash, err = fftypes.ParseBytes32(h.ctx, msg.Hash)
 		if err == nil {
-			_, namespace, _ = splitBlobPath(msg.Path)
+			var ns string
+			pathParts := strings.Split(msg.Path, "/")
+			if len(pathParts) >= 2 {
+				ns = pathParts[len(pathParts)-2]
+			}
 			e.dxType = dataexchange.DXEventTypePrivateBlobReceived
 			e.privateBlobReceived = &dataexchange.PrivateBlobReceived{
-				Namespace:  namespace,
+				Namespace:  ns,
 				PeerID:     msg.Sender,
 				Hash:       *hash,
 				Size:       msg.Size,
@@ -197,15 +188,11 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 	default:
 		err = i18n.NewError(h.ctx, coremsgs.MsgUnexpectedDXMessageType, msg.Type)
 	}
-
 	// If we couldn't dispatch the event we received, we still ack it
 	if err != nil {
 		log.L(h.ctx).Warnf("Failed to dispatch DX event: %s", err)
 		e.Ack()
 	} else {
-		if namespace == "" && msg.RequestID != "" {
-			namespace, _, _ = core.ParseNamespacedOpID(h.ctx, msg.RequestID)
-		}
-		h.callbacks.DXEvent(h.ctx, namespace, e)
+		h.callbacks.DXEvent(e)
 	}
 }
diff --git a/internal/dataexchange/ffdx/ffdx.go b/internal/dataexchange/ffdx/ffdx.go
index 23c6abfd2..e4359d4e0 100644
--- a/internal/dataexchange/ffdx/ffdx.go
+++ b/internal/dataexchange/ffdx/ffdx.go
@@ -23,7 +23,6 @@ import (
 	"io"
 	"net/http"
 	"strconv"
-	"strings"
 	"sync"
 
 	"github.com/go-resty/resty/v2"
@@ -51,36 +50,15 @@ type FFDX struct {
 }
 
 type callbacks struct {
-	handlers map[string]dataexchange.Callbacks
+	handlers []dataexchange.Callbacks
 }
 
-func (cb *callbacks) DXEvent(ctx context.Context, namespace string, event dataexchange.DXEvent) {
-	if handler, ok := cb.handlers[namespace]; ok {
-		handler.DXEvent(event)
-	} else {
-		log.L(ctx).Errorf("unknown namespace on event '%s'", event.EventID())
-		event.Ack()
+func (cb *callbacks) DXEvent(event dataexchange.DXEvent) {
+	for _, cb := range cb.handlers {
+		cb.DXEvent(event)
 	}
 }
 
-func splitLast(s string, sep string) (string, string) {
-	split := strings.LastIndex(s, sep)
-	if split == -1 {
-		return "", s
-	}
-	return s[:split], s[split+1:]
-}
-
-func splitBlobPath(path string) (prefix, namespace, id string) {
-	path, id = splitLast(path, "/")
-	path, namespace = splitLast(path, "/")
-	return path, namespace, id
-}
-
-func joinBlobPath(namespace, id string) string {
-	return fmt.Sprintf("%s/%s", namespace, id)
-}
-
 const (
 	dxHTTPHeaderHash = "dx-hash"
 	dxHTTPHeaderSize = "dx-size"
@@ -143,7 +121,7 @@ func (h *FFDX) Name() string {
 func (h *FFDX) Init(ctx context.Context, config config.Section) (err error) {
 	h.ctx = log.WithLogField(ctx, "dx", "https")
 	h.ackChannel = make(chan *ack)
-	h.callbacks.handlers = make(map[string]dataexchange.Callbacks)
+
 	h.needsInit = config.GetBool(DataExchangeInitEnabled)
 
 	if config.GetString(ffresty.HTTPConfigURL) == "" {
@@ -170,8 +148,8 @@ func (h *FFDX) SetNodes(nodes []fftypes.JSONObject) {
 	h.nodes = nodes
 }
 
-func (h *FFDX) SetHandler(namespace string, handler dataexchange.Callbacks) {
-	h.callbacks.handlers[namespace] = handler
+func (h *FFDX) SetHandler(handler dataexchange.Callbacks) {
+	h.callbacks.handlers = append(h.callbacks.handlers, handler)
 }
 
 func (h *FFDX) Start() error {
@@ -249,7 +227,7 @@ func (h *FFDX) AddPeer(ctx context.Context, peer fftypes.JSONObject) (err error)
 }
 
 func (h *FFDX) UploadBlob(ctx context.Context, ns string, id fftypes.UUID, content io.Reader) (payloadRef string, hash *fftypes.Bytes32, size int64, err error) {
-	payloadRef = joinBlobPath(ns, id.String())
+	payloadRef = fmt.Sprintf("%s/%s", ns, &id)
 	var upload uploadBlob
 	res, err := h.client.R().SetContext(ctx).
 		SetFileReader("file", id.String(), content).
diff --git a/internal/events/batch_pin_complete.go b/internal/events/batch_pin_complete.go
index d08d5dbf9..7f16ab7bd 100644
--- a/internal/events/batch_pin_complete.go
+++ b/internal/events/batch_pin_complete.go
@@ -41,7 +41,7 @@ func (em *eventManager) BatchPinComplete(batchPin *blockchain.BatchPin, signingK
 		return nil // move on
 	}
 	if batchPin.Namespace != em.namespace {
-		log.L(em.ctx).Debugf("Ignoring batch pin from different namespace '%s'", batchPin.Namespace)
+		log.L(em.ctx).Debugf("Ignoring BatchPin from different namespace '%s'", batchPin.Namespace)
 		return nil // move on
 	}
 
diff --git a/internal/events/dx_callbacks.go b/internal/events/dx_callbacks.go
index 4c8ad8549..c1668fc9e 100644
--- a/internal/events/dx_callbacks.go
+++ b/internal/events/dx_callbacks.go
@@ -19,6 +19,7 @@ package events
 import (
 	"context"
 	"database/sql/driver"
+	"encoding/json"
 
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/log"
@@ -183,15 +184,27 @@ func (em *eventManager) messageReceived(dx dataexchange.Plugin, event dataexchan
 	l := log.L(em.ctx)
 
 	mr := event.MessageReceived()
-	l.Infof("Private batch received from %s peer '%s'", dx.Name(), mr.PeerID)
 
-	if mr.Transport.Batch.Namespace != em.namespace {
-		log.L(em.ctx).Debugf("Ignoring batch from different namespace '%s'", mr.Transport.Batch.Namespace)
+	// De-serialize the transport wrapper
+	var wrapper *core.TransportWrapper
+	err := json.Unmarshal(mr.Data, &wrapper)
+	if err != nil {
+		l.Errorf("Invalid transmission from %s peer '%s': %s", dx.Name(), mr.PeerID, err)
+		event.AckWithManifest("")
+		return
+	}
+	if wrapper.Batch == nil {
+		l.Errorf("Invalid transmission: nil batch")
 		event.AckWithManifest("")
 		return
 	}
+	if wrapper.Batch.Namespace != em.namespace {
+		log.L(em.ctx).Debugf("Ignoring batch from different namespace '%s'", wrapper.Batch.Namespace)
+		return
+	}
+	l.Infof("Private batch received from %s peer '%s' (len=%d)", dx.Name(), mr.PeerID, len(mr.Data))
 
-	manifestString, err := em.privateBatchReceived(mr.PeerID, mr.Transport.Batch, mr.Transport.Group)
+	manifestString, err := em.privateBatchReceived(mr.PeerID, wrapper.Batch, wrapper.Group)
 	if err != nil {
 		l.Warnf("Exited while persisting batch: %s", err)
 		// We do NOT ack here as we broke out of the retry
@@ -211,7 +224,6 @@ func (em *eventManager) privateBlobReceived(dx dataexchange.Plugin, event dataex
 	}
 	if br.Namespace != em.namespace {
 		log.L(em.ctx).Debugf("Ignoring blob from different namespace '%s'", br.Namespace)
-		event.Ack() // Still confirm the event
 		return
 	}
 
diff --git a/internal/identity/tbd/tbd.go b/internal/identity/tbd/tbd.go
index 716b6eb19..1fb5aac02 100644
--- a/internal/identity/tbd/tbd.go
+++ b/internal/identity/tbd/tbd.go
@@ -37,7 +37,7 @@ func (tbd *TBD) Init(ctx context.Context, config config.Section) (err error) {
 	return nil
 }
 
-func (tbd *TBD) SetHandler(namespace string, handler identity.Callbacks) {
+func (tbd *TBD) SetHandler(handler identity.Callbacks) {
 }
 
 func (tbd *TBD) Start() error {
diff --git a/internal/multiparty/manager.go b/internal/multiparty/manager.go
index 60521d485..7616b0a4d 100644
--- a/internal/multiparty/manager.go
+++ b/internal/multiparty/manager.go
@@ -24,11 +24,16 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/coremsgs"
+	"github.com/hyperledger/firefly/internal/metrics"
+	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/pkg/blockchain"
 	"github.com/hyperledger/firefly/pkg/core"
+	"github.com/hyperledger/firefly/pkg/database"
 )
 
 type Manager interface {
+	core.Named
+
 	// ConfigureContract initializes the subscription to the FireFly contract
 	// - Checks the provided contract info against the plugin's configuration, and updates it as needed
 	// - Initializes the contract info for performing BatchPin transactions, and initializes subscriptions for BatchPin events
@@ -44,12 +49,14 @@ type Manager interface {
 	GetNetworkVersion() int
 
 	// SubmitBatchPin sequences a batch of message globally to all viewers of a given ledger
-	SubmitBatchPin(ctx context.Context, nsOpID string, signingKey string, batch *blockchain.BatchPin) error
+	SubmitBatchPin(ctx context.Context, batch *core.BatchPersisted, contexts []*fftypes.Bytes32, payloadRef string) error
 
 	// SubmitNetworkAction writes a special "BatchPin" event which signals the plugin to take an action
 	SubmitNetworkAction(ctx context.Context, nsOpID string, signingKey string, action core.NetworkActionType) error
 
-	core.Named
+	// From operations.OperationHandler
+	PrepareOperation(ctx context.Context, op *core.Operation) (*core.PreparedOperation, error)
+	RunOperation(ctx context.Context, op *core.PreparedOperation) (outputs fftypes.JSONObject, complete bool, err error)
 }
 
 type Contract struct {
@@ -59,9 +66,12 @@ type Contract struct {
 
 type multipartyManager struct {
 	ctx            context.Context
+	namespace      string
+	database       database.Plugin
 	blockchain     blockchain.Plugin
+	operations     operations.Manager
+	metrics        metrics.Manager
 	contracts      []Contract
-	namespace      string
 	activeContract struct {
 		location       *fftypes.JSONAny
 		firstEvent     string
@@ -70,13 +80,27 @@ type multipartyManager struct {
 	}
 }
 
-func NewMultipartyManager(ctx context.Context, ns string, contracts []Contract, bi blockchain.Plugin) Manager {
-	return &multipartyManager{
+func NewMultipartyManager(ctx context.Context, ns string, contracts []Contract, di database.Plugin, bi blockchain.Plugin, om operations.Manager, mm metrics.Manager) (Manager, error) {
+	if di == nil || bi == nil || mm == nil || om == nil {
+		return nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "MultipartyManager")
+	}
+	mp := &multipartyManager{
 		ctx:        ctx,
 		namespace:  ns,
 		contracts:  contracts,
+		database:   di,
 		blockchain: bi,
+		operations: om,
+		metrics:    mm,
 	}
+	om.RegisterHandler(ctx, mp, []core.OpType{
+		core.OpTypeBlockchainPinBatch,
+	})
+	return mp, nil
+}
+
+func (mm *multipartyManager) Name() string {
+	return "MultipartyManager"
 }
 
 func (mm *multipartyManager) ConfigureContract(ctx context.Context, contracts *core.FireFlyContracts) (err error) {
@@ -145,10 +169,21 @@ func (mm *multipartyManager) SubmitNetworkAction(ctx context.Context, nsOpID str
 	return mm.blockchain.SubmitNetworkAction(ctx, nsOpID, signingKey, action, mm.activeContract.location)
 }
 
-func (mm *multipartyManager) SubmitBatchPin(ctx context.Context, nsOpID string, signingKey string, batch *blockchain.BatchPin) error {
-	return mm.blockchain.SubmitBatchPin(ctx, nsOpID, signingKey, batch, mm.activeContract.location)
-}
+func (mm *multipartyManager) SubmitBatchPin(ctx context.Context, batch *core.BatchPersisted, contexts []*fftypes.Bytes32, payloadRef string) error {
+	// The pending blockchain transaction
+	op := core.NewOperation(
+		mm.blockchain,
+		batch.Namespace,
+		batch.TX.ID,
+		core.OpTypeBlockchainPinBatch)
+	addBatchPinInputs(op, batch.ID, contexts, payloadRef)
+	if err := mm.operations.AddOrReuseOperation(ctx, op); err != nil {
+		return err
+	}
 
-func (mm *multipartyManager) Name() string {
-	return mm.blockchain.Name()
+	if mm.metrics.IsMetricsEnabled() {
+		mm.metrics.CountBatchPin()
+	}
+	_, err := mm.operations.RunOperation(ctx, opBatchPin(op, batch, contexts, payloadRef))
+	return err
 }
diff --git a/internal/batchpin/operations.go b/internal/multiparty/operations.go
similarity index 90%
rename from internal/batchpin/operations.go
rename to internal/multiparty/operations.go
index 6ad8165fa..ed4dfd5c1 100644
--- a/internal/batchpin/operations.go
+++ b/internal/multiparty/operations.go
@@ -14,7 +14,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package batchpin
+package multiparty
 
 import (
 	"context"
@@ -62,14 +62,14 @@ func retrieveBatchPinInputs(ctx context.Context, op *core.Operation) (batchID *f
 	return batchID, contexts, payloadRef, nil
 }
 
-func (bp *batchPinSubmitter) PrepareOperation(ctx context.Context, op *core.Operation) (*core.PreparedOperation, error) {
+func (mm *multipartyManager) PrepareOperation(ctx context.Context, op *core.Operation) (*core.PreparedOperation, error) {
 	switch op.Type {
 	case core.OpTypeBlockchainPinBatch:
 		batchID, contexts, payloadRef, err := retrieveBatchPinInputs(ctx, op)
 		if err != nil {
 			return nil, err
 		}
-		batch, err := bp.database.GetBatchByID(ctx, bp.namespace, batchID)
+		batch, err := mm.database.GetBatchByID(ctx, mm.namespace, batchID)
 		if err != nil {
 			return nil, err
 		} else if batch == nil {
@@ -82,25 +82,25 @@ func (bp *batchPinSubmitter) PrepareOperation(ctx context.Context, op *core.Oper
 	}
 }
 
-func (bp *batchPinSubmitter) RunOperation(ctx context.Context, op *core.PreparedOperation) (outputs fftypes.JSONObject, complete bool, err error) {
+func (mm *multipartyManager) RunOperation(ctx context.Context, op *core.PreparedOperation) (outputs fftypes.JSONObject, complete bool, err error) {
 	switch data := op.Data.(type) {
 	case batchPinData:
 		batch := data.Batch
-		return nil, false, bp.multiparty.SubmitBatchPin(ctx, op.NamespacedIDString(), batch.Key, &blockchain.BatchPin{
+		return nil, false, mm.blockchain.SubmitBatchPin(ctx, op.NamespacedIDString(), batch.Key, &blockchain.BatchPin{
 			Namespace:       batch.Namespace,
 			TransactionID:   batch.TX.ID,
 			BatchID:         batch.ID,
 			BatchHash:       batch.Hash,
 			BatchPayloadRef: data.PayloadRef,
 			Contexts:        data.Contexts,
-		})
+		}, mm.activeContract.location)
 
 	default:
 		return nil, false, i18n.NewError(ctx, coremsgs.MsgOperationDataIncorrect, op.Data)
 	}
 }
 
-func (bp *batchPinSubmitter) OnOperationUpdate(ctx context.Context, op *core.Operation, update *operations.OperationUpdate) error {
+func (mm *multipartyManager) OnOperationUpdate(ctx context.Context, op *core.Operation, update *operations.OperationUpdate) error {
 	return nil
 }
 
diff --git a/internal/namespace/manager.go b/internal/namespace/manager.go
index 7aedda6ba..9822eeaff 100644
--- a/internal/namespace/manager.go
+++ b/internal/namespace/manager.go
@@ -555,7 +555,7 @@ func (nm *namespaceManager) initPlugins(ctx context.Context) (err error) {
 		if err = entry.plugin.Init(ctx, entry.config); err != nil {
 			return err
 		}
-		entry.plugin.SetHandler(database.GlobalHandler, nm)
+		entry.plugin.SetHandler(nm)
 	}
 	for _, entry := range nm.plugins.blockchain {
 		if err = entry.plugin.Init(ctx, entry.config, nm.metrics); err != nil {
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index e9102a6e9..069731ae4 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -24,7 +24,6 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/assets"
 	"github.com/hyperledger/firefly/internal/batch"
-	"github.com/hyperledger/firefly/internal/batchpin"
 	"github.com/hyperledger/firefly/internal/broadcast"
 	"github.com/hyperledger/firefly/internal/contracts"
 	"github.com/hyperledger/firefly/internal/coremsgs"
@@ -184,7 +183,6 @@ type orchestrator struct {
 	definitions    definitions.DefinitionHandler
 	data           data.Manager
 	syncasync      syncasync.Bridge
-	batchpin       batchpin.Submitter
 	assets         assets.Manager
 	bc             boundCallbacks
 	contracts      contracts.Manager
@@ -361,9 +359,8 @@ func (or *orchestrator) MultiParty() multiparty.Manager {
 }
 
 func (or *orchestrator) initPlugins(ctx context.Context) (err error) {
-	or.plugins.Database.Plugin.SetHandler(or.namespace, or)
+	or.plugins.Database.Plugin.SetHandler(or)
 	or.plugins.Blockchain.Plugin.SetHandler(&or.bc)
-	or.plugins.SharedStorage.Plugin.SetHandler(or.namespace, &or.bc)
 
 	fb := database.IdentityQueryFactory.NewFilter(ctx)
 	nodes, _, err := or.database().GetIdentities(ctx, or.namespace, fb.And(
@@ -377,7 +374,9 @@ func (or *orchestrator) initPlugins(ctx context.Context) (err error) {
 		nodeInfo[i] = node.Profile
 	}
 	or.plugins.DataExchange.Plugin.SetNodes(nodeInfo)
-	or.plugins.DataExchange.Plugin.SetHandler(or.namespace, &or.bc)
+	or.plugins.DataExchange.Plugin.SetHandler(&or.bc)
+
+	or.plugins.SharedStorage.Plugin.SetHandler(&or.bc)
 
 	for _, token := range or.plugins.Tokens {
 		if err := token.Plugin.SetHandler(or.namespace, &or.bc); err != nil {
@@ -390,17 +389,6 @@ func (or *orchestrator) initPlugins(ctx context.Context) (err error) {
 
 func (or *orchestrator) initComponents(ctx context.Context) (err error) {
 
-	if or.multiparty == nil {
-		or.multiparty = multiparty.NewMultipartyManager(or.ctx, or.namespace, or.config.Multiparty.Contracts, or.blockchain())
-	}
-
-	if or.identity == nil {
-		or.identity, err = identity.NewIdentityManager(ctx, or.namespace, or.config.DefaultKey, or.config.Multiparty.OrgName, or.config.Multiparty.OrgKey, or.database(), or.blockchain(), or.multiparty)
-		if err != nil {
-			return err
-		}
-	}
-
 	if or.data == nil {
 		or.data, err = data.NewDataManager(ctx, or.namespace, or.database(), or.sharedstorage(), or.dataexchange())
 		if err != nil {
@@ -418,29 +406,37 @@ func (or *orchestrator) initComponents(ctx context.Context) (err error) {
 		}
 	}
 
-	or.syncasync = syncasync.NewSyncAsyncBridge(ctx, or.namespace, or.database(), or.data)
+	if or.multiparty == nil {
+		or.multiparty, err = multiparty.NewMultipartyManager(or.ctx, or.namespace, or.config.Multiparty.Contracts, or.database(), or.blockchain(), or.operations, or.metrics)
+		if err != nil {
+			return err
+		}
+	}
 
-	if or.batch == nil {
-		or.batch, err = batch.NewBatchManager(ctx, or.namespace, or, or.database(), or.data, or.txHelper)
+	if or.identity == nil {
+		or.identity, err = identity.NewIdentityManager(ctx, or.namespace, or.config.DefaultKey, or.config.Multiparty.OrgName, or.config.Multiparty.OrgKey, or.database(), or.blockchain(), or.multiparty)
 		if err != nil {
 			return err
 		}
 	}
 
-	if or.batchpin == nil {
-		if or.batchpin, err = batchpin.NewBatchPinSubmitter(ctx, or.namespace, or.database(), or.identity, or.multiparty, or.metrics, or.operations); err != nil {
+	or.syncasync = syncasync.NewSyncAsyncBridge(ctx, or.namespace, or.database(), or.data)
+
+	if or.batch == nil {
+		or.batch, err = batch.NewBatchManager(ctx, or.namespace, or, or.database(), or.data, or.txHelper)
+		if err != nil {
 			return err
 		}
 	}
 
 	if or.messaging == nil {
-		if or.messaging, err = privatemessaging.NewPrivateMessaging(ctx, or.namespace, or.database(), or.dataexchange(), or.blockchain(), or.identity, or.batch, or.data, or.syncasync, or.batchpin, or.metrics, or.operations); err != nil {
+		if or.messaging, err = privatemessaging.NewPrivateMessaging(ctx, or.namespace, or.database(), or.dataexchange(), or.blockchain(), or.identity, or.batch, or.data, or.syncasync, or.multiparty, or.metrics, or.operations); err != nil {
 			return err
 		}
 	}
 
 	if or.broadcast == nil {
-		if or.broadcast, err = broadcast.NewBroadcastManager(ctx, or.namespace, or.database(), or.blockchain(), or.dataexchange(), or.sharedstorage(), or.identity, or.data, or.batch, or.syncasync, or.batchpin, or.metrics, or.operations); err != nil {
+		if or.broadcast, err = broadcast.NewBroadcastManager(ctx, or.namespace, or.database(), or.blockchain(), or.dataexchange(), or.sharedstorage(), or.identity, or.data, or.batch, or.syncasync, or.multiparty, or.metrics, or.operations); err != nil {
 			return err
 		}
 	}
diff --git a/internal/privatemessaging/privatemessaging.go b/internal/privatemessaging/privatemessaging.go
index 33be7d7e9..8aa8b675f 100644
--- a/internal/privatemessaging/privatemessaging.go
+++ b/internal/privatemessaging/privatemessaging.go
@@ -26,12 +26,12 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly-common/pkg/retry"
 	"github.com/hyperledger/firefly/internal/batch"
-	"github.com/hyperledger/firefly/internal/batchpin"
 	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/data"
 	"github.com/hyperledger/firefly/internal/identity"
 	"github.com/hyperledger/firefly/internal/metrics"
+	"github.com/hyperledger/firefly/internal/multiparty"
 	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/internal/syncasync"
 	"github.com/hyperledger/firefly/internal/sysmessaging"
@@ -70,7 +70,7 @@ type privateMessaging struct {
 	blockchain            blockchain.Plugin
 	data                  data.Manager
 	syncasync             syncasync.Bridge
-	batchpin              batchpin.Submitter
+	multiparty            multiparty.Manager
 	retry                 retry.Retry
 	localNodeName         string
 	localNodeID           *fftypes.UUID // lookup and cached on first use, as might not be registered at startup
@@ -86,8 +86,8 @@ type blobTransferTracker struct {
 	op       *core.PreparedOperation
 }
 
-func NewPrivateMessaging(ctx context.Context, ns string, di database.Plugin, dx dataexchange.Plugin, bi blockchain.Plugin, im identity.Manager, ba batch.Manager, dm data.Manager, sa syncasync.Bridge, bp batchpin.Submitter, mm metrics.Manager, om operations.Manager) (Manager, error) {
-	if di == nil || im == nil || dx == nil || bi == nil || ba == nil || dm == nil || mm == nil || om == nil {
+func NewPrivateMessaging(ctx context.Context, ns string, di database.Plugin, dx dataexchange.Plugin, bi blockchain.Plugin, im identity.Manager, ba batch.Manager, dm data.Manager, sa syncasync.Bridge, mult multiparty.Manager, mm metrics.Manager, om operations.Manager) (Manager, error) {
+	if di == nil || im == nil || dx == nil || bi == nil || ba == nil || dm == nil || mm == nil || om == nil || mult == nil {
 		return nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "PrivateMessaging")
 	}
 
@@ -100,7 +100,7 @@ func NewPrivateMessaging(ctx context.Context, ns string, di database.Plugin, dx
 		blockchain:    bi,
 		data:          dm,
 		syncasync:     sa,
-		batchpin:      bp,
+		multiparty:    mult,
 		localNodeName: config.GetString(coreconfig.NodeName),
 		groupManager: groupManager{
 			namespace:     ns,
@@ -172,7 +172,7 @@ func (pm *privateMessaging) dispatchPinnedBatch(ctx context.Context, state *batc
 	}
 
 	log.L(ctx).Infof("Pinning private batch %s with author=%s key=%s group=%s", state.Persisted.ID, state.Persisted.Author, state.Persisted.Key, state.Persisted.Group)
-	return pm.batchpin.SubmitPinnedBatch(ctx, &state.Persisted, state.Pins, "" /* no payloadRef for private */)
+	return pm.multiparty.SubmitBatchPin(ctx, &state.Persisted, state.Pins, "" /* no payloadRef for private */)
 }
 
 func (pm *privateMessaging) dispatchUnpinnedBatch(ctx context.Context, state *batch.DispatchState) error {
diff --git a/internal/sharedstorage/ipfs/ipfs.go b/internal/sharedstorage/ipfs/ipfs.go
index c3ac2c226..a7a994a9e 100644
--- a/internal/sharedstorage/ipfs/ipfs.go
+++ b/internal/sharedstorage/ipfs/ipfs.go
@@ -67,7 +67,7 @@ func (i *IPFS) Init(ctx context.Context, config config.Section) error {
 	return nil
 }
 
-func (i *IPFS) SetHandler(namespace string, handler sharedstorage.Callbacks) {
+func (i *IPFS) SetHandler(handler sharedstorage.Callbacks) {
 }
 
 func (i *IPFS) Capabilities() *sharedstorage.Capabilities {
diff --git a/internal/tokens/fftokens/fftokens.go b/internal/tokens/fftokens/fftokens.go
index ec423492e..a068061cf 100644
--- a/internal/tokens/fftokens/fftokens.go
+++ b/internal/tokens/fftokens/fftokens.go
@@ -54,7 +54,8 @@ func (cb *callbacks) TokenOpUpdate(plugin tokens.Plugin, nsOpID string, txState
 
 func (cb *callbacks) TokenPoolCreated(namespace string, plugin tokens.Plugin, pool *tokens.TokenPool) error {
 	if namespace == "" {
-		// Older token subscriptions don't populate namespace, so deliver the event to every listener
+		// Older token subscriptions don't populate namespace, so deliver the event to every listener and let them filter
+		// TODO: deprecate this path
 		for _, cb := range cb.handlers {
 			if err := cb.TokenPoolCreated(plugin, pool); err != nil {
 				return err
@@ -70,7 +71,8 @@ func (cb *callbacks) TokenPoolCreated(namespace string, plugin tokens.Plugin, po
 
 func (cb *callbacks) TokensTransferred(namespace string, plugin tokens.Plugin, transfer *tokens.TokenTransfer) error {
 	if namespace == "" {
-		// Older token subscriptions don't populate namespace, so deliver the event to every listener
+		// Older token subscriptions don't populate namespace, so deliver the event to every listener and let them filter
+		// TODO: deprecate this path
 		for _, cb := range cb.handlers {
 			if err := cb.TokensTransferred(plugin, transfer); err != nil {
 				return err
@@ -86,7 +88,8 @@ func (cb *callbacks) TokensTransferred(namespace string, plugin tokens.Plugin, t
 
 func (cb *callbacks) TokensApproved(namespace string, plugin tokens.Plugin, approval *tokens.TokenApproval) error {
 	if namespace == "" {
-		// Older token subscriptions don't populate namespace, so deliver the event to every listener
+		// Older token subscriptions don't populate namespace, so deliver the event to every listener and let them filter
+		// TODO: deprecate this path
 		for _, cb := range cb.handlers {
 			if err := cb.TokensApproved(plugin, approval); err != nil {
 				return err
diff --git a/mocks/batchpinmocks/submitter.go b/mocks/batchpinmocks/submitter.go
deleted file mode 100644
index 8c0fbbfdc..000000000
--- a/mocks/batchpinmocks/submitter.go
+++ /dev/null
@@ -1,98 +0,0 @@
-// Code generated by mockery v1.0.0. DO NOT EDIT.
-
-package batchpinmocks
-
-import (
-	context "context"
-
-	fftypes "github.com/hyperledger/firefly-common/pkg/fftypes"
-	core "github.com/hyperledger/firefly/pkg/core"
-
-	mock "github.com/stretchr/testify/mock"
-)
-
-// Submitter is an autogenerated mock type for the Submitter type
-type Submitter struct {
-	mock.Mock
-}
-
-// Name provides a mock function with given fields:
-func (_m *Submitter) Name() string {
-	ret := _m.Called()
-
-	var r0 string
-	if rf, ok := ret.Get(0).(func() string); ok {
-		r0 = rf()
-	} else {
-		r0 = ret.Get(0).(string)
-	}
-
-	return r0
-}
-
-// PrepareOperation provides a mock function with given fields: ctx, op
-func (_m *Submitter) PrepareOperation(ctx context.Context, op *core.Operation) (*core.PreparedOperation, error) {
-	ret := _m.Called(ctx, op)
-
-	var r0 *core.PreparedOperation
-	if rf, ok := ret.Get(0).(func(context.Context, *core.Operation) *core.PreparedOperation); ok {
-		r0 = rf(ctx, op)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*core.PreparedOperation)
-		}
-	}
-
-	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *core.Operation) error); ok {
-		r1 = rf(ctx, op)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
-// RunOperation provides a mock function with given fields: ctx, op
-func (_m *Submitter) RunOperation(ctx context.Context, op *core.PreparedOperation) (fftypes.JSONObject, bool, error) {
-	ret := _m.Called(ctx, op)
-
-	var r0 fftypes.JSONObject
-	if rf, ok := ret.Get(0).(func(context.Context, *core.PreparedOperation) fftypes.JSONObject); ok {
-		r0 = rf(ctx, op)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(fftypes.JSONObject)
-		}
-	}
-
-	var r1 bool
-	if rf, ok := ret.Get(1).(func(context.Context, *core.PreparedOperation) bool); ok {
-		r1 = rf(ctx, op)
-	} else {
-		r1 = ret.Get(1).(bool)
-	}
-
-	var r2 error
-	if rf, ok := ret.Get(2).(func(context.Context, *core.PreparedOperation) error); ok {
-		r2 = rf(ctx, op)
-	} else {
-		r2 = ret.Error(2)
-	}
-
-	return r0, r1, r2
-}
-
-// SubmitPinnedBatch provides a mock function with given fields: ctx, batch, contexts, payloadRef
-func (_m *Submitter) SubmitPinnedBatch(ctx context.Context, batch *core.BatchPersisted, contexts []*fftypes.Bytes32, payloadRef string) error {
-	ret := _m.Called(ctx, batch, contexts, payloadRef)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *core.BatchPersisted, []*fftypes.Bytes32, string) error); ok {
-		r0 = rf(ctx, batch, contexts, payloadRef)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
diff --git a/mocks/databasemocks/plugin.go b/mocks/databasemocks/plugin.go
index 5d2f0eaf4..1d81a0d04 100644
--- a/mocks/databasemocks/plugin.go
+++ b/mocks/databasemocks/plugin.go
@@ -2410,9 +2410,9 @@ func (_m *Plugin) RunAsGroup(ctx context.Context, fn func(context.Context) error
 	return r0
 }
 
-// SetHandler provides a mock function with given fields: namespace, handler
-func (_m *Plugin) SetHandler(namespace string, handler database.Callbacks) {
-	_m.Called(namespace, handler)
+// SetHandler provides a mock function with given fields: handler
+func (_m *Plugin) SetHandler(handler database.Callbacks) {
+	_m.Called(handler)
 }
 
 // UpdateBatch provides a mock function with given fields: ctx, id, update
diff --git a/mocks/dataexchangemocks/plugin.go b/mocks/dataexchangemocks/plugin.go
index db294a069..9b1c406c3 100644
--- a/mocks/dataexchangemocks/plugin.go
+++ b/mocks/dataexchangemocks/plugin.go
@@ -174,9 +174,9 @@ func (_m *Plugin) SendMessage(ctx context.Context, nsOpID string, peerID string,
 	return r0
 }
 
-// SetHandler provides a mock function with given fields: namespace, handler
-func (_m *Plugin) SetHandler(namespace string, handler dataexchange.Callbacks) {
-	_m.Called(namespace, handler)
+// SetHandler provides a mock function with given fields: handler
+func (_m *Plugin) SetHandler(handler dataexchange.Callbacks) {
+	_m.Called(handler)
 }
 
 // SetNodes provides a mock function with given fields: nodes
diff --git a/mocks/identitymocks/plugin.go b/mocks/identitymocks/plugin.go
index 7917b124e..1cf2899ed 100644
--- a/mocks/identitymocks/plugin.go
+++ b/mocks/identitymocks/plugin.go
@@ -66,9 +66,9 @@ func (_m *Plugin) Name() string {
 	return r0
 }
 
-// SetHandler provides a mock function with given fields: namespace, handler
-func (_m *Plugin) SetHandler(namespace string, handler identity.Callbacks) {
-	_m.Called(namespace, handler)
+// SetHandler provides a mock function with given fields: handler
+func (_m *Plugin) SetHandler(handler identity.Callbacks) {
+	_m.Called(handler)
 }
 
 // Start provides a mock function with given fields:
diff --git a/mocks/multipartymocks/manager.go b/mocks/multipartymocks/manager.go
index 154835ed5..3b584dd66 100644
--- a/mocks/multipartymocks/manager.go
+++ b/mocks/multipartymocks/manager.go
@@ -61,13 +61,66 @@ func (_m *Manager) Name() string {
 	return r0
 }
 
-// SubmitBatchPin provides a mock function with given fields: ctx, nsOpID, signingKey, batch
-func (_m *Manager) SubmitBatchPin(ctx context.Context, nsOpID string, signingKey string, batch *blockchain.BatchPin) error {
-	ret := _m.Called(ctx, nsOpID, signingKey, batch)
+// PrepareOperation provides a mock function with given fields: ctx, op
+func (_m *Manager) PrepareOperation(ctx context.Context, op *core.Operation) (*core.PreparedOperation, error) {
+	ret := _m.Called(ctx, op)
+
+	var r0 *core.PreparedOperation
+	if rf, ok := ret.Get(0).(func(context.Context, *core.Operation) *core.PreparedOperation); ok {
+		r0 = rf(ctx, op)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*core.PreparedOperation)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, *core.Operation) error); ok {
+		r1 = rf(ctx, op)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+// RunOperation provides a mock function with given fields: ctx, op
+func (_m *Manager) RunOperation(ctx context.Context, op *core.PreparedOperation) (fftypes.JSONObject, bool, error) {
+	ret := _m.Called(ctx, op)
+
+	var r0 fftypes.JSONObject
+	if rf, ok := ret.Get(0).(func(context.Context, *core.PreparedOperation) fftypes.JSONObject); ok {
+		r0 = rf(ctx, op)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(fftypes.JSONObject)
+		}
+	}
+
+	var r1 bool
+	if rf, ok := ret.Get(1).(func(context.Context, *core.PreparedOperation) bool); ok {
+		r1 = rf(ctx, op)
+	} else {
+		r1 = ret.Get(1).(bool)
+	}
+
+	var r2 error
+	if rf, ok := ret.Get(2).(func(context.Context, *core.PreparedOperation) error); ok {
+		r2 = rf(ctx, op)
+	} else {
+		r2 = ret.Error(2)
+	}
+
+	return r0, r1, r2
+}
+
+// SubmitBatchPin provides a mock function with given fields: ctx, batch, contexts, payloadRef
+func (_m *Manager) SubmitBatchPin(ctx context.Context, batch *core.BatchPersisted, contexts []*fftypes.Bytes32, payloadRef string) error {
+	ret := _m.Called(ctx, batch, contexts, payloadRef)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, *blockchain.BatchPin) error); ok {
-		r0 = rf(ctx, nsOpID, signingKey, batch)
+	if rf, ok := ret.Get(0).(func(context.Context, *core.BatchPersisted, []*fftypes.Bytes32, string) error); ok {
+		r0 = rf(ctx, batch, contexts, payloadRef)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/mocks/sharedstoragemocks/plugin.go b/mocks/sharedstoragemocks/plugin.go
index c57521921..e53897766 100644
--- a/mocks/sharedstoragemocks/plugin.go
+++ b/mocks/sharedstoragemocks/plugin.go
@@ -91,9 +91,9 @@ func (_m *Plugin) Name() string {
 	return r0
 }
 
-// SetHandler provides a mock function with given fields: namespace, handler
-func (_m *Plugin) SetHandler(namespace string, handler sharedstorage.Callbacks) {
-	_m.Called(namespace, handler)
+// SetHandler provides a mock function with given fields: handler
+func (_m *Plugin) SetHandler(handler sharedstorage.Callbacks) {
+	_m.Called(handler)
 }
 
 // UploadData provides a mock function with given fields: ctx, data
diff --git a/pkg/core/operation.go b/pkg/core/operation.go
index af939b76a..4d145f491 100644
--- a/pkg/core/operation.go
+++ b/pkg/core/operation.go
@@ -129,7 +129,7 @@ func (po *PreparedOperation) NamespacedIDString() string {
 func ParseNamespacedOpID(ctx context.Context, nsIDStr string) (string, *fftypes.UUID, error) {
 	nsIDSplit := strings.Split(nsIDStr, ":")
 	if len(nsIDSplit) != 2 {
-		return "", nil, i18n.NewError(ctx, coremsgs.MsgInvalidNamespaceUUID, nsIDStr)
+		return "", nil, i18n.NewError(context.Background(), coremsgs.MsgInvalidNamespaceUUID, nsIDStr)
 	}
 	ns := nsIDSplit[0]
 	uuidStr := nsIDSplit[1]
diff --git a/pkg/database/plugin.go b/pkg/database/plugin.go
index 232721fa3..e58096c7a 100644
--- a/pkg/database/plugin.go
+++ b/pkg/database/plugin.go
@@ -43,11 +43,6 @@ const (
 	UpsertOptimizationExisting
 )
 
-const (
-	// Pseudo-namespace to register a global callback handler, which will receive all namespaced and non-namespaced events
-	GlobalHandler = "ff:global"
-)
-
 // Plugin is the interface implemented by each plugin
 type Plugin interface {
 	PersistenceInterface // Split out to aid pluggability the next level down (SQL provider etc.)
@@ -59,8 +54,7 @@ type Plugin interface {
 	Init(ctx context.Context, config config.Section) error
 
 	// SetHandler registers a handler to receive callbacks
-	// If namespace is set, plugin will attempt to deliver only events for that namespace
-	SetHandler(namespace string, handler Callbacks)
+	SetHandler(handler Callbacks)
 
 	// Capabilities returns capabilities - not called until after Init
 	Capabilities() *Capabilities
diff --git a/pkg/dataexchange/plugin.go b/pkg/dataexchange/plugin.go
index 285ff13af..523a2d7b6 100644
--- a/pkg/dataexchange/plugin.go
+++ b/pkg/dataexchange/plugin.go
@@ -67,8 +67,7 @@ type Plugin interface {
 	SetNodes(nodes []fftypes.JSONObject)
 
 	// SetHandler registers a handler to receive callbacks
-	// If namespace is set, plugin will attempt to deliver only events for that namespace
-	SetHandler(namespace string, handler Callbacks)
+	SetHandler(handler Callbacks)
 
 	// Data exchange interface must not deliver any events until start is called
 	Start() error
@@ -126,8 +125,8 @@ const (
 )
 
 type MessageReceived struct {
-	PeerID    string
-	Transport *core.TransportWrapper
+	PeerID string
+	Data   []byte
 }
 
 type PrivateBlobReceived struct {
diff --git a/pkg/events/plugin.go b/pkg/events/plugin.go
index 8697f4d47..abd2a99a7 100644
--- a/pkg/events/plugin.go
+++ b/pkg/events/plugin.go
@@ -36,7 +36,6 @@ type Plugin interface {
 	Init(ctx context.Context, config config.Section) error
 
 	// SetHandler registers a handler to receive callbacks
-	// If namespace is set, plugin will attempt to deliver only events for that namespace
 	SetHandler(namespace string, handler Callbacks) error
 
 	// Capabilities returns capabilities - not called until after Init
diff --git a/pkg/identity/plugin.go b/pkg/identity/plugin.go
index 7fa46d944..cfa43a0a5 100644
--- a/pkg/identity/plugin.go
+++ b/pkg/identity/plugin.go
@@ -34,8 +34,7 @@ type Plugin interface {
 	Init(ctx context.Context, config config.Section) error
 
 	// SetHandler registers a handler to receive callbacks
-	// If namespace is set, plugin will attempt to deliver only events for that namespace
-	SetHandler(namespace string, handler Callbacks)
+	SetHandler(handler Callbacks)
 
 	// Blockchain interface must not deliver any events until start is called
 	Start() error
diff --git a/pkg/sharedstorage/plugin.go b/pkg/sharedstorage/plugin.go
index 49ab0ef44..ff8fcbe49 100644
--- a/pkg/sharedstorage/plugin.go
+++ b/pkg/sharedstorage/plugin.go
@@ -35,8 +35,7 @@ type Plugin interface {
 	Init(ctx context.Context, config config.Section) error
 
 	// SetHandler registers a handler to receive callbacks
-	// If namespace is set, plugin will attempt to deliver only events for that namespace
-	SetHandler(namespace string, handler Callbacks)
+	SetHandler(handler Callbacks)
 
 	// Capabilities returns capabilities - not called until after Init
 	Capabilities() *Capabilities
diff --git a/pkg/tokens/plugin.go b/pkg/tokens/plugin.go
index d9cf249b5..309266a72 100644
--- a/pkg/tokens/plugin.go
+++ b/pkg/tokens/plugin.go
@@ -36,7 +36,6 @@ type Plugin interface {
 	Init(ctx context.Context, name string, config config.Section) error
 
 	// SetHandler registers a handler to receive callbacks
-	// If namespace is set, plugin will attempt to deliver only events for that namespace
 	SetHandler(namespace string, handler Callbacks) error
 
 	// Blockchain interface must not deliver any events until start is called


Test patch:
diff --git a/internal/batchpin/batchpin_test.go b/internal/batchpin/batchpin_test.go
deleted file mode 100644
index 8e095694d..000000000
--- a/internal/batchpin/batchpin_test.go
+++ /dev/null
@@ -1,180 +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 batchpin
-
-import (
-	"context"
-	"fmt"
-	"testing"
-
-	"github.com/hyperledger/firefly-common/pkg/config"
-	"github.com/hyperledger/firefly-common/pkg/fftypes"
-	"github.com/hyperledger/firefly/internal/coreconfig"
-	"github.com/hyperledger/firefly/mocks/databasemocks"
-	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
-	"github.com/hyperledger/firefly/mocks/metricsmocks"
-	"github.com/hyperledger/firefly/mocks/multipartymocks"
-	"github.com/hyperledger/firefly/mocks/operationmocks"
-	"github.com/hyperledger/firefly/pkg/core"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
-)
-
-var utConfig = config.RootSection("metrics")
-
-func newTestBatchPinSubmitter(t *testing.T, enableMetrics bool) *batchPinSubmitter {
-	coreconfig.Reset()
-
-	mdi := &databasemocks.Plugin{}
-	mim := &identitymanagermocks.Manager{}
-	mmp := &multipartymocks.Manager{}
-	mmi := &metricsmocks.Manager{}
-	mom := &operationmocks.Manager{}
-	mmi.On("IsMetricsEnabled").Return(enableMetrics)
-	mom.On("RegisterHandler", mock.Anything, mock.Anything, mock.Anything)
-	if enableMetrics {
-		mmi.On("CountBatchPin").Return()
-	}
-	mmp.On("Name").Return("ut").Maybe()
-	bps, err := NewBatchPinSubmitter(context.Background(), "ns1", mdi, mim, mmp, mmi, mom)
-	assert.NoError(t, err)
-	return bps.(*batchPinSubmitter)
-}
-
-func TestInitFail(t *testing.T) {
-	_, err := NewBatchPinSubmitter(context.Background(), "", nil, nil, nil, nil, nil)
-	assert.Regexp(t, "FF10128", err)
-}
-
-func TestName(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
-	assert.Equal(t, "BatchPinSubmitter", bp.Name())
-}
-
-func TestSubmitPinnedBatchOk(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
-	ctx := context.Background()
-
-	mdi := bp.database.(*databasemocks.Plugin)
-	mmi := bp.metrics.(*metricsmocks.Manager)
-	mom := bp.operations.(*operationmocks.Manager)
-
-	batch := &core.BatchPersisted{
-		BatchHeader: core.BatchHeader{
-			ID: fftypes.NewUUID(),
-			SignerRef: core.SignerRef{
-				Author: "id1",
-				Key:    "0x12345",
-			},
-		},
-		TX: core.TransactionRef{
-			ID: fftypes.NewUUID(),
-		},
-	}
-	contexts := []*fftypes.Bytes32{}
-
-	mom.On("AddOrReuseOperation", ctx, mock.MatchedBy(func(op *core.Operation) bool {
-		assert.Equal(t, core.OpTypeBlockchainPinBatch, op.Type)
-		assert.Equal(t, "ut", op.Plugin)
-		assert.Equal(t, *batch.TX.ID, *op.Transaction)
-		assert.Equal(t, "payload1", op.Input.GetString("payloadRef"))
-		return true
-	})).Return(nil)
-	mmi.On("IsMetricsEnabled").Return(false)
-	mom.On("RunOperation", mock.Anything, mock.MatchedBy(func(op *core.PreparedOperation) bool {
-		data := op.Data.(batchPinData)
-		return op.Type == core.OpTypeBlockchainPinBatch && data.Batch == batch
-	})).Return(nil, nil)
-
-	err := bp.SubmitPinnedBatch(ctx, batch, contexts, "payload1")
-	assert.NoError(t, err)
-
-	mdi.AssertExpectations(t)
-	mmi.AssertExpectations(t)
-	mom.AssertExpectations(t)
-}
-
-func TestSubmitPinnedBatchWithMetricsOk(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, true)
-	ctx := context.Background()
-
-	mdi := bp.database.(*databasemocks.Plugin)
-	mmi := bp.metrics.(*metricsmocks.Manager)
-	mom := bp.operations.(*operationmocks.Manager)
-
-	batch := &core.BatchPersisted{
-		BatchHeader: core.BatchHeader{
-			ID: fftypes.NewUUID(),
-			SignerRef: core.SignerRef{
-				Author: "id1",
-				Key:    "0x12345",
-			},
-		},
-		TX: core.TransactionRef{
-			ID: fftypes.NewUUID(),
-		},
-	}
-	contexts := []*fftypes.Bytes32{}
-
-	mom.On("AddOrReuseOperation", ctx, mock.MatchedBy(func(op *core.Operation) bool {
-		assert.Equal(t, core.OpTypeBlockchainPinBatch, op.Type)
-		assert.Equal(t, "ut", op.Plugin)
-		assert.Equal(t, *batch.TX.ID, *op.Transaction)
-		assert.Equal(t, "payload1", op.Input.GetString("payloadRef"))
-		return true
-	})).Return(nil)
-	mmi.On("IsMetricsEnabled").Return(true)
-	mom.On("RunOperation", mock.Anything, mock.MatchedBy(func(op *core.PreparedOperation) bool {
-		data := op.Data.(batchPinData)
-		return op.Type == core.OpTypeBlockchainPinBatch && data.Batch == batch
-	})).Return(nil, nil)
-
-	err := bp.SubmitPinnedBatch(ctx, batch, contexts, "payload1")
-	assert.NoError(t, err)
-
-	mdi.AssertExpectations(t)
-	mmi.AssertExpectations(t)
-	mom.AssertExpectations(t)
-}
-
-func TestSubmitPinnedBatchOpFail(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
-	ctx := context.Background()
-
-	mom := bp.operations.(*operationmocks.Manager)
-	mmi := bp.metrics.(*metricsmocks.Manager)
-
-	batch := &core.BatchPersisted{
-		BatchHeader: core.BatchHeader{
-			ID: fftypes.NewUUID(),
-			SignerRef: core.SignerRef{
-				Author: "id1",
-				Key:    "0x12345",
-			},
-		},
-		TX: core.TransactionRef{
-			ID: fftypes.NewUUID(),
-		},
-	}
-	contexts := []*fftypes.Bytes32{}
-
-	mom.On("AddOrReuseOperation", ctx, mock.Anything).Return(fmt.Errorf("pop"))
-	mmi.On("IsMetricsEnabled").Return(false)
-	err := bp.SubmitPinnedBatch(ctx, batch, contexts, "payload1")
-	assert.Regexp(t, "pop", err)
-
-}
diff --git a/internal/broadcast/manager_test.go b/internal/broadcast/manager_test.go
index a8eff6734..6ef2cbe24 100644
--- a/internal/broadcast/manager_test.go
+++ b/internal/broadcast/manager_test.go
@@ -27,13 +27,13 @@ import (
 	"github.com/hyperledger/firefly/internal/data"
 	"github.com/hyperledger/firefly/internal/operations"
 	"github.com/hyperledger/firefly/mocks/batchmocks"
-	"github.com/hyperledger/firefly/mocks/batchpinmocks"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/dataexchangemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
 	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
 	"github.com/hyperledger/firefly/mocks/metricsmocks"
+	"github.com/hyperledger/firefly/mocks/multipartymocks"
 	"github.com/hyperledger/firefly/mocks/operationmocks"
 	"github.com/hyperledger/firefly/mocks/sharedstoragemocks"
 	"github.com/hyperledger/firefly/mocks/syncasyncmocks"
@@ -52,7 +52,7 @@ func newTestBroadcastCommon(t *testing.T, metricsEnabled bool) (*broadcastManage
 	mba := &batchmocks.Manager{}
 	mdx := &dataexchangemocks.Plugin{}
 	msa := &syncasyncmocks.Bridge{}
-	mbp := &batchpinmocks.Submitter{}
+	mmp := &multipartymocks.Manager{}
 	mmi := &metricsmocks.Manager{}
 	mom := &operationmocks.Manager{}
 	mmi.On("IsMetricsEnabled").Return(metricsEnabled)
@@ -76,7 +76,7 @@ func newTestBroadcastCommon(t *testing.T, metricsEnabled bool) (*broadcastManage
 	}
 
 	ctx, cancel := context.WithCancel(context.Background())
-	b, err := NewBroadcastManager(ctx, "ns1", mdi, mbi, mdx, mpi, mim, mdm, mba, msa, mbp, mmi, mom)
+	b, err := NewBroadcastManager(ctx, "ns1", mdi, mbi, mdx, mpi, mim, mdm, mba, msa, mmp, mmi, mom)
 	assert.NoError(t, err)
 	return b.(*broadcastManager), cancel
 }
@@ -247,10 +247,10 @@ func TestDispatchBatchSubmitBatchPinSucceed(t *testing.T) {
 	}
 
 	mdi := bm.database.(*databasemocks.Plugin)
-	mbp := bm.batchpin.(*batchpinmocks.Submitter)
+	mmp := bm.multiparty.(*multipartymocks.Manager)
 	mom := bm.operations.(*operationmocks.Manager)
 	mom.On("AddOrReuseOperation", mock.Anything, mock.Anything).Return(nil)
-	mbp.On("SubmitPinnedBatch", mock.Anything, mock.Anything, mock.Anything, "payload1").Return(nil)
+	mmp.On("SubmitBatchPin", mock.Anything, mock.Anything, mock.Anything, "payload1").Return(nil)
 	mom.On("RunOperation", mock.Anything, mock.MatchedBy(func(op *core.PreparedOperation) bool {
 		data := op.Data.(uploadBatchData)
 		return op.Type == core.OpTypeSharedStorageUploadBatch && data.Batch.ID.Equals(state.Persisted.ID)
@@ -260,7 +260,7 @@ func TestDispatchBatchSubmitBatchPinSucceed(t *testing.T) {
 	assert.NoError(t, err)
 
 	mdi.AssertExpectations(t)
-	mbp.AssertExpectations(t)
+	mmp.AssertExpectations(t)
 	mom.AssertExpectations(t)
 }
 
@@ -279,10 +279,10 @@ func TestDispatchBatchSubmitBroadcastFail(t *testing.T) {
 	}
 
 	mdi := bm.database.(*databasemocks.Plugin)
-	mbp := bm.batchpin.(*batchpinmocks.Submitter)
+	mmp := bm.multiparty.(*multipartymocks.Manager)
 	mom := bm.operations.(*operationmocks.Manager)
 	mom.On("AddOrReuseOperation", mock.Anything, mock.Anything).Return(nil)
-	mbp.On("SubmitPinnedBatch", mock.Anything, mock.Anything, mock.Anything, "payload1").Return(fmt.Errorf("pop"))
+	mmp.On("SubmitBatchPin", mock.Anything, mock.Anything, mock.Anything, "payload1").Return(fmt.Errorf("pop"))
 	mom.On("RunOperation", mock.Anything, mock.MatchedBy(func(op *core.PreparedOperation) bool {
 		data := op.Data.(uploadBatchData)
 		return op.Type == core.OpTypeSharedStorageUploadBatch && data.Batch.ID.Equals(state.Persisted.ID)
@@ -292,7 +292,7 @@ func TestDispatchBatchSubmitBroadcastFail(t *testing.T) {
 	assert.EqualError(t, err, "pop")
 
 	mdi.AssertExpectations(t)
-	mbp.AssertExpectations(t)
+	mmp.AssertExpectations(t)
 	mom.AssertExpectations(t)
 }
 
diff --git a/internal/database/postgres/postgres_test.go b/internal/database/postgres/postgres_test.go
index 60a4f7de7..dea012436 100644
--- a/internal/database/postgres/postgres_test.go
+++ b/internal/database/postgres/postgres_test.go
@@ -29,7 +29,7 @@ import (
 
 func TestPostgresProvider(t *testing.T) {
 	psql := &Postgres{}
-	psql.SetHandler("ns", &databasemocks.Callbacks{})
+	psql.SetHandler(&databasemocks.Callbacks{})
 	config := config.RootSection("unittest")
 	psql.InitConfig(config)
 	config.Set(sqlcommon.SQLConfDatasourceURL, "!bad connection")
diff --git a/internal/database/sqlcommon/contractapis_sql_test.go b/internal/database/sqlcommon/contractapis_sql_test.go
index c7552f268..e977fc596 100644
--- a/internal/database/sqlcommon/contractapis_sql_test.go
+++ b/internal/database/sqlcommon/contractapis_sql_test.go
@@ -126,7 +126,7 @@ func TestContractAPIDBFailUpdate(t *testing.T) {
 func TestUpsertContractAPIIDMismatch(t *testing.T) {
 	s, db := newMockProvider().init()
 	callbacks := &databasemocks.Callbacks{}
-	s.SetHandler("ns1", callbacks)
+	s.SetHandler(callbacks)
 	apiID := fftypes.NewUUID()
 	api := &core.ContractAPI{
 		ID:        apiID,
diff --git a/internal/database/sqlcommon/provider_mock_test.go b/internal/database/sqlcommon/provider_mock_test.go
index 2331c6415..91a6a2751 100644
--- a/internal/database/sqlcommon/provider_mock_test.go
+++ b/internal/database/sqlcommon/provider_mock_test.go
@@ -64,7 +64,7 @@ func newMockProvider() *mockProvider {
 // init is a convenience to init for tests that aren't testing init itself
 func (mp *mockProvider) init() (*mockProvider, sqlmock.Sqlmock) {
 	_ = mp.Init(context.Background(), mp, mp.config, mp.capabilities)
-	mp.SetHandler(database.GlobalHandler, mp.callbacks)
+	mp.SetHandler(mp.callbacks)
 	return mp, mp.mdb
 }
 
diff --git a/internal/database/sqlcommon/provider_sqlitego_test.go b/internal/database/sqlcommon/provider_sqlitego_test.go
index d57e7f02a..1b6a64360 100644
--- a/internal/database/sqlcommon/provider_sqlitego_test.go
+++ b/internal/database/sqlcommon/provider_sqlitego_test.go
@@ -65,7 +65,7 @@ func newSQLiteTestProvider(t *testing.T) (*sqliteGoTestProvider, func()) {
 
 	err = tp.Init(context.Background(), tp, tp.config, tp.capabilities)
 	assert.NoError(tp.t, err)
-	tp.SetHandler(database.GlobalHandler, tp.callbacks)
+	tp.SetHandler(tp.callbacks)
 
 	return tp, func() {
 		tp.Close()
diff --git a/internal/database/sqlcommon/sqlcommon_test.go b/internal/database/sqlcommon/sqlcommon_test.go
index 641e9a61b..27f3b0c54 100644
--- a/internal/database/sqlcommon/sqlcommon_test.go
+++ b/internal/database/sqlcommon/sqlcommon_test.go
@@ -26,9 +26,7 @@ import (
 	sq "github.com/Masterminds/squirrel"
 	"github.com/golang-migrate/migrate/v4"
 	"github.com/hyperledger/firefly-common/pkg/config"
-	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
-	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/stretchr/testify/assert"
 )
@@ -363,26 +361,3 @@ func TestInsertTxRowsIncompleteReturn(t *testing.T) {
 	err = s.insertTxRows(ctx, "table1", tx, sb, nil, []int64{1, 2}, false)
 	assert.Regexp(t, "FF10116", err)
 }
-
-func TestNamespaceCallbacks(t *testing.T) {
-	tcb := &databasemocks.Callbacks{}
-	cb := callbacks{
-		handlers: map[string]database.Callbacks{
-			"ns1": tcb,
-		},
-	}
-	id := fftypes.NewUUID()
-	hash := fftypes.NewRandB32()
-
-	tcb.On("OrderedUUIDCollectionNSEvent", database.CollectionMessages, core.ChangeEventTypeCreated, "ns1", id, int64(1)).Return()
-	tcb.On("OrderedCollectionNSEvent", database.CollectionPins, core.ChangeEventTypeCreated, "ns1", int64(1)).Return()
-	tcb.On("UUIDCollectionNSEvent", database.CollectionOperations, core.ChangeEventTypeCreated, "ns1", id).Return()
-	tcb.On("UUIDCollectionEvent", database.CollectionNamespaces, core.ChangeEventTypeCreated, id).Return()
-	tcb.On("HashCollectionNSEvent", database.CollectionGroups, core.ChangeEventTypeUpdated, "ns1", hash).Return()
-
-	cb.OrderedUUIDCollectionNSEvent(database.CollectionMessages, core.ChangeEventTypeCreated, "ns1", id, 1)
-	cb.OrderedCollectionNSEvent(database.CollectionPins, core.ChangeEventTypeCreated, "ns1", 1)
-	cb.UUIDCollectionNSEvent(database.CollectionOperations, core.ChangeEventTypeCreated, "ns1", id)
-	cb.UUIDCollectionEvent(database.CollectionNamespaces, core.ChangeEventTypeCreated, id)
-	cb.HashCollectionNSEvent(database.CollectionGroups, core.ChangeEventTypeUpdated, "ns1", hash)
-}
diff --git a/internal/database/sqlcommon/tokenpool_sql_test.go b/internal/database/sqlcommon/tokenpool_sql_test.go
index 9cbb14fb9..72be8a5a8 100644
--- a/internal/database/sqlcommon/tokenpool_sql_test.go
+++ b/internal/database/sqlcommon/tokenpool_sql_test.go
@@ -175,7 +175,7 @@ func TestUpsertTokenPoolFailCommit(t *testing.T) {
 func TestUpsertTokenPoolUpdateIDMismatch(t *testing.T) {
 	s, db := newMockProvider().init()
 	callbacks := &databasemocks.Callbacks{}
-	s.SetHandler("ns1", callbacks)
+	s.SetHandler(callbacks)
 	poolID := fftypes.NewUUID()
 	pool := &core.TokenPool{
 		ID:        poolID,
diff --git a/internal/database/sqlite3/sqlite3_test.go b/internal/database/sqlite3/sqlite3_test.go
index aa43df616..cac1659da 100644
--- a/internal/database/sqlite3/sqlite3_test.go
+++ b/internal/database/sqlite3/sqlite3_test.go
@@ -32,7 +32,7 @@ import (
 
 func TestSQLite3GoProvider(t *testing.T) {
 	sqlite := &SQLite3{}
-	sqlite.SetHandler("ns", &databasemocks.Callbacks{})
+	sqlite.SetHandler(&databasemocks.Callbacks{})
 	config := config.RootSection("unittest")
 	sqlite.InitConfig(config)
 	config.Set(sqlcommon.SQLConfDatasourceURL, "!wrong://")
diff --git a/internal/dataexchange/ffdx/ffdx_test.go b/internal/dataexchange/ffdx/ffdx_test.go
index d511bd51b..de41c9790 100644
--- a/internal/dataexchange/ffdx/ffdx_test.go
+++ b/internal/dataexchange/ffdx/ffdx_test.go
@@ -73,38 +73,6 @@ func newTestFFDX(t *testing.T, manifestEnabled bool) (h *FFDX, toServer, fromSer
 	}
 }
 
-func TestSplitBlobPath(t *testing.T) {
-	prefix, namespace, id := splitBlobPath("")
-	assert.Equal(t, "", prefix)
-	assert.Equal(t, "", namespace)
-	assert.Equal(t, "", id)
-
-	prefix, namespace, id = splitBlobPath("123")
-	assert.Equal(t, "", prefix)
-	assert.Equal(t, "", namespace)
-	assert.Equal(t, "123", id)
-
-	prefix, namespace, id = splitBlobPath("ns1/123")
-	assert.Equal(t, "", prefix)
-	assert.Equal(t, "ns1", namespace)
-	assert.Equal(t, "123", id)
-
-	prefix, namespace, id = splitBlobPath("/ns1/123")
-	assert.Equal(t, "", prefix)
-	assert.Equal(t, "ns1", namespace)
-	assert.Equal(t, "123", id)
-
-	prefix, namespace, id = splitBlobPath("/root/test/ns1/123")
-	assert.Equal(t, "/root/test", prefix)
-	assert.Equal(t, "ns1", namespace)
-	assert.Equal(t, "123", id)
-}
-
-func TestJoinBlobPath(t *testing.T) {
-	path := joinBlobPath("ns1", "123")
-	assert.Equal(t, "ns1/123", path)
-}
-
 func TestInitBadURL(t *testing.T) {
 	coreconfig.Reset()
 	h := &FFDX{}
@@ -461,14 +429,11 @@ func TestTransferBlobError(t *testing.T) {
 	assert.Regexp(t, "FF10229", err)
 }
 
-func TestBadEvents(t *testing.T) {
+func TestEvents(t *testing.T) {
 
 	h, toServer, fromServer, _, done := newTestFFDX(t, false)
 	defer done()
 
-	mcb := &dataexchangemocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
-
 	err := h.Start()
 	assert.NoError(t, err)
 
@@ -477,118 +442,84 @@ func TestBadEvents(t *testing.T) {
 	msg := <-toServer
 	assert.Equal(t, `{"action":"ack","id":"0"}`, string(msg))
 
-	namespacedID := fmt.Sprintf("ns2:%s", fftypes.NewUUID())
-	fromServer <- `{"id":"1","type":"message-failed","requestID":"` + namespacedID + `"}`
-	msg = <-toServer
-	assert.Equal(t, `{"action":"ack","id":"1"}`, string(msg))
-
-	fromServer <- `{"id":"2","type":"message-received","sender":"peer1","message":"bad!"}`
-	msg = <-toServer
-	assert.Equal(t, `{"action":"ack","id":"2"}`, string(msg))
-
-	fromServer <- `{"id":"3","type":"message-received","sender":"peer1","message":"{}"}`
-	msg = <-toServer
-	assert.Equal(t, `{"action":"ack","id":"3"}`, string(msg))
-
-	mcb.AssertExpectations(t)
-}
-
-func TestMessageEvents(t *testing.T) {
-
-	h, toServer, fromServer, _, done := newTestFFDX(t, false)
-	defer done()
-
 	mcb := &dataexchangemocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
-
-	err := h.Start()
-	assert.NoError(t, err)
+	h.SetHandler(mcb)
 
-	namespacedID1 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "1" &&
-			ev.NamespacedID() == namespacedID1 &&
+			ev.NamespacedID() == "tx12345" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID1 &&
+			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusFailed &&
 			ev.TransferResult().Error == "pop"
 	})).Run(acker()).Return(nil)
-	fromServer <- `{"id":"1","type":"message-failed","requestID":"` + namespacedID1 + `","error":"pop"}`
-	msg := <-toServer
+	fromServer <- `{"id":"1","type":"message-failed","requestID":"tx12345","error":"pop"}`
+	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"1"}`, string(msg))
 
-	namespacedID2 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "2" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID2 &&
+			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusSucceeded
 	})).Run(acker()).Return(nil)
-	fromServer <- `{"id":"2","type":"message-delivered","requestID":"` + namespacedID2 + `"}`
+	fromServer <- `{"id":"2","type":"message-delivered","requestID":"tx12345"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"2"}`, string(msg))
 
-	namespacedID3 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "3" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID3 &&
+			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusSucceeded &&
 			ev.TransferResult().Manifest == `{"manifest":true}` &&
 			ev.TransferResult().Info.String() == `{"signatures":"and stuff"}`
 	})).Run(acker()).Return(nil)
-	fromServer <- `{"id":"3","type":"message-acknowledged","requestID":"` + namespacedID3 + `","info":{"signatures":"and stuff"},"manifest":"{\"manifest\":true}"}`
+	fromServer <- `{"id":"3","type":"message-acknowledged","requestID":"tx12345","info":{"signatures":"and stuff"},"manifest":"{\"manifest\":true}"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"3"}`, string(msg))
 
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "4" &&
 			ev.Type() == dataexchange.DXEventTypeMessageReceived &&
-			ev.MessageReceived().PeerID == "peer1"
+			ev.MessageReceived().PeerID == "peer1" &&
+			string(ev.MessageReceived().Data) == "message1"
 	})).Run(manifestAcker(`{"manifest":true}`)).Return(nil)
-	fromServer <- `{"id":"4","type":"message-received","sender":"peer1","message":"{\"batch\":{\"namespace\":\"ns1\"}}"}`
+	fromServer <- `{"id":"4","type":"message-received","sender":"peer1","message":"message1"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"4","manifest":"{\"manifest\":true}"}`, string(msg))
 
-	mcb.AssertExpectations(t)
-}
-
-func TestBlobEvents(t *testing.T) {
-
-	h, toServer, fromServer, _, done := newTestFFDX(t, false)
-	defer done()
-
-	mcb := &dataexchangemocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
-
-	err := h.Start()
-	assert.NoError(t, err)
-
-	namespacedID5 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "5" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID5 &&
+			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusFailed &&
 			ev.TransferResult().Error == "pop"
 	})).Run(acker()).Return(nil)
-	fromServer <- `{"id":"5","type":"blob-failed","requestID":"` + namespacedID5 + `","error":"pop"}`
-	msg := <-toServer
+	fromServer <- `{"id":"5","type":"blob-failed","requestID":"tx12345","error":"pop"}`
+	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"5"}`, string(msg))
 
-	namespacedID6 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "6" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID6 &&
+			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusSucceeded &&
 			ev.TransferResult().Info.String() == `{"some":"details"}`
 	})).Run(acker()).Return(nil)
-	fromServer <- `{"id":"6","type":"blob-delivered","requestID":"` + namespacedID6 + `","info":{"some":"details"}}`
+	fromServer <- `{"id":"6","type":"blob-delivered","requestID":"tx12345","info":{"some":"details"}}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"6"}`, string(msg))
 
+	fromServer <- `{"id":"7","type":"blob-received","sender":"peer1","path":"ns1/! not a UUID"}`
+	msg = <-toServer
+	assert.Equal(t, `{"action":"ack","id":"7"}`, string(msg))
+
 	u := fftypes.NewUUID()
+	fromServer <- fmt.Sprintf(`{"id":"8","type":"blob-received","sender":"peer1","path":"ns1/%s","hash":"!wrong","size":-1}`, u.String())
+	msg = <-toServer
+	assert.Equal(t, `{"action":"ack","id":"8"}`, string(msg))
+
 	hash := fftypes.NewRandB32()
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "9" &&
@@ -599,15 +530,14 @@ func TestBlobEvents(t *testing.T) {
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"9"}`, string(msg))
 
-	namespacedID10 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "10" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
-			ev.TransferResult().TrackingID == namespacedID10 &&
+			ev.TransferResult().TrackingID == "tx12345" &&
 			ev.TransferResult().Status == core.OpStatusSucceeded &&
 			ev.TransferResult().Info.String() == `{"signatures":"and stuff"}`
 	})).Run(acker()).Return(nil)
-	fromServer <- `{"id":"10","type":"blob-acknowledged","requestID":"` + namespacedID10 + `","info":{"signatures":"and stuff"},"manifest":"{\"manifest\":true}"}`
+	fromServer <- `{"id":"10","type":"blob-acknowledged","requestID":"tx12345","info":{"signatures":"and stuff"},"manifest":"{\"manifest\":true}"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"10"}`, string(msg))
 
@@ -628,25 +558,23 @@ func TestEventsWithManifest(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"0"}`, string(msg))
 
 	mcb := &dataexchangemocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
+	h.SetHandler(mcb)
 
-	namespacedID1 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "1" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().Status == core.OpStatusPending
 	})).Run(acker()).Return(nil)
-	fromServer <- `{"id":"1","type":"message-delivered","requestID":"` + namespacedID1 + `"}`
+	fromServer <- `{"id":"1","type":"message-delivered","requestID":"tx12345"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"1"}`, string(msg))
 
-	namespacedID2 := fmt.Sprintf("ns1:%s", fftypes.NewUUID())
 	mcb.On("DXEvent", mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "2" &&
 			ev.Type() == dataexchange.DXEventTypeTransferResult &&
 			ev.TransferResult().Status == core.OpStatusPending
 	})).Run(acker()).Return(nil)
-	fromServer <- `{"id":"2","type":"blob-delivered","requestID":"` + namespacedID2 + `"}`
+	fromServer <- `{"id":"2","type":"blob-delivered","requestID":"tx12345"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"2"}`, string(msg))
 
@@ -658,7 +586,7 @@ func TestEventLoopReceiveClosed(t *testing.T) {
 	wsm := &wsmocks.WSClient{}
 	h := &FFDX{
 		ctx:       context.Background(),
-		callbacks: callbacks{handlers: map[string]dataexchange.Callbacks{"ns1": dxc}},
+		callbacks: callbacks{handlers: []dataexchange.Callbacks{dxc}},
 		wsconn:    wsm,
 	}
 	r := make(chan []byte)
@@ -674,7 +602,7 @@ func TestEventLoopSendClosed(t *testing.T) {
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	h := &FFDX{
 		ctx:        ctx,
-		callbacks:  callbacks{handlers: map[string]dataexchange.Callbacks{"ns1": dxc}},
+		callbacks:  callbacks{handlers: []dataexchange.Callbacks{dxc}},
 		wsconn:     wsm,
 		ackChannel: make(chan *ack, 1),
 	}
@@ -695,7 +623,7 @@ func TestEventLoopClosedContext(t *testing.T) {
 	cancel()
 	h := &FFDX{
 		ctx:       ctx,
-		callbacks: callbacks{handlers: map[string]dataexchange.Callbacks{"ns1": dxc}},
+		callbacks: callbacks{handlers: []dataexchange.Callbacks{dxc}},
 		wsconn:    wsm,
 	}
 	r := make(chan []byte, 1)
diff --git a/internal/events/dx_callbacks_test.go b/internal/events/dx_callbacks_test.go
index e797236c1..e808b1af5 100644
--- a/internal/events/dx_callbacks_test.go
+++ b/internal/events/dx_callbacks_test.go
@@ -18,6 +18,7 @@ package events
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"strings"
 	"testing"
@@ -35,15 +36,15 @@ import (
 	"github.com/stretchr/testify/mock"
 )
 
-func sampleBatchTransfer(t *testing.T, txType core.TransactionType) (*core.Batch, *core.TransportWrapper) {
+func sampleBatchTransfer(t *testing.T, txType core.TransactionType) (*core.Batch, []byte) {
 	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, core.BatchTypePrivate, txType, core.DataArray{data})
-	b := &core.TransportWrapper{
+	b, _ := json.Marshal(&core.TransportWrapper{
 		Batch: batch,
 		Group: &core.Group{
 			Hash: fftypes.NewRandB32(),
 		},
-	}
+	})
 	return batch, b
 }
 
@@ -81,18 +82,18 @@ func newTestNode(name string, owner *core.Identity) *core.Identity {
 	return identity
 }
 
-func newMessageReceivedNoAck(peerID string, transport *core.TransportWrapper) *dataexchangemocks.DXEvent {
+func newMessageReceivedNoAck(peerID string, data []byte) *dataexchangemocks.DXEvent {
 	mde := &dataexchangemocks.DXEvent{}
 	mde.On("MessageReceived").Return(&dataexchange.MessageReceived{
-		PeerID:    peerID,
-		Transport: transport,
+		PeerID: peerID,
+		Data:   data,
 	})
 	mde.On("Type").Return(dataexchange.DXEventTypeMessageReceived).Maybe()
 	return mde
 }
 
-func newMessageReceived(peerID string, transport *core.TransportWrapper, expectedManifest string) *dataexchangemocks.DXEvent {
-	mde := newMessageReceivedNoAck(peerID, transport)
+func newMessageReceived(peerID string, data []byte, expectedManifest string) *dataexchangemocks.DXEvent {
+	mde := newMessageReceivedNoAck(peerID, data)
 	mde.On("AckWithManifest", expectedManifest).Return()
 	return mde
 }
@@ -182,12 +183,12 @@ func TestMessageReceiveOkBadBatchIgnored(t *testing.T) {
 	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, core.BatchTypePrivate, core.TransactionTypeBatchPin, core.DataArray{data})
 	batch.Payload.TX.Type = core.TransactionTypeTokenPool
-	b := &core.TransportWrapper{
+	b, _ := json.Marshal(&core.TransportWrapper{
 		Batch: batch,
 		Group: &core.Group{
 			Hash: fftypes.NewRandB32(),
 		},
-	}
+	})
 
 	org1 := newTestOrg("org1")
 	node1 := newTestNode("node1", org1)
@@ -237,6 +238,65 @@ func TestMessageReceivePersistBatchError(t *testing.T) {
 	mim.AssertExpectations(t)
 }
 
+func TestMessageReceivedBadData(t *testing.T) {
+	em, cancel := newTestEventManager(t)
+	defer cancel()
+
+	mdx := &dataexchangemocks.Plugin{}
+	mdx.On("Name").Return("utdx")
+
+	mde := newMessageReceived("peer1", []byte(`!{}`), "")
+	em.messageReceived(mdx, mde)
+
+	mde.AssertExpectations(t)
+
+}
+
+func TestMessageReceivedUnknownType(t *testing.T) {
+	em, cancel := newTestEventManager(t)
+	defer cancel()
+
+	mdx := &dataexchangemocks.Plugin{}
+	mdx.On("Name").Return("utdx")
+
+	mde := newMessageReceived("peer1", []byte(`{
+		"type": "unknown"
+	}`), "")
+	em.messageReceived(mdx, mde)
+
+	mde.AssertExpectations(t)
+}
+
+func TestMessageReceivedNilBatch(t *testing.T) {
+	em, cancel := newTestEventManager(t)
+	defer cancel()
+
+	mdx := &dataexchangemocks.Plugin{}
+	mdx.On("Name").Return("utdx")
+
+	mde := newMessageReceived("peer1", []byte(`{
+		"type": "batch"
+	}`), "")
+	em.messageReceived(mdx, mde)
+
+	mde.AssertExpectations(t)
+}
+
+func TestMessageReceivedNilMessage(t *testing.T) {
+	em, cancel := newTestEventManager(t)
+	defer cancel()
+
+	mdx := &dataexchangemocks.Plugin{}
+	mdx.On("Name").Return("utdx")
+
+	mde := newMessageReceived("peer1", []byte(`{
+		"type": "message"
+	}`), "")
+	em.messageReceived(mdx, mde)
+
+	mde.AssertExpectations(t)
+}
+
 func TestMessageReceivedWrongNS(t *testing.T) {
 	em, cancel := newTestEventManager(t)
 	defer cancel()
@@ -247,13 +307,29 @@ func TestMessageReceivedWrongNS(t *testing.T) {
 	mdx := &dataexchangemocks.Plugin{}
 	mdx.On("Name").Return("utdx")
 
-	mde := newMessageReceived("peer1", b, "")
+	mde := newMessageReceivedNoAck("peer1", b)
 	em.messageReceived(mdx, mde)
 
 	mde.AssertExpectations(t)
 
 }
 
+func TestMessageReceivedNilGroup(t *testing.T) {
+	em, cancel := newTestEventManager(t)
+	defer cancel()
+
+	mdx := &dataexchangemocks.Plugin{}
+	mdx.On("Name").Return("utdx")
+
+	mde := newMessageReceived("peer1", []byte(`{
+		"type": "message",
+		"message": {}
+	}`), "")
+	em.messageReceived(mdx, mde)
+
+	mde.AssertExpectations(t)
+}
+
 func TestMessageReceiveNodeLookupError(t *testing.T) {
 	em, cancel := newTestEventManager(t)
 	cancel() // to stop retry
@@ -263,9 +339,9 @@ func TestMessageReceiveNodeLookupError(t *testing.T) {
 			Namespace: "ns1",
 		},
 	}
-	b := &core.TransportWrapper{
+	b, _ := json.Marshal(&core.TransportWrapper{
 		Batch: batch,
-	}
+	})
 
 	mim := em.identity.(*identitymanagermocks.Manager)
 	mim.On("FindIdentityForVerifier", em.ctx, []core.IdentityType{core.IdentityTypeNode}, &core.VerifierRef{
@@ -447,7 +523,8 @@ func TestPrivateBlobReceivedWrongNS(t *testing.T) {
 	mdx := &dataexchangemocks.Plugin{}
 	mdx.On("Name").Return("utdx")
 
-	mde := newPrivateBlobReceived("peer1", hash, 12345, "ns1/path1")
+	// no ack as we are simulating termination mid retry
+	mde := newPrivateBlobReceivedNoAck("peer1", hash, 12345, "ns1/path1")
 	em.privateBlobReceived(mdx, mde)
 
 	mde.AssertExpectations(t)
diff --git a/internal/multiparty/manager_test.go b/internal/multiparty/manager_test.go
index 5e32fc105..0557361e9 100644
--- a/internal/multiparty/manager_test.go
+++ b/internal/multiparty/manager_test.go
@@ -23,6 +23,9 @@ import (
 
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
+	"github.com/hyperledger/firefly/mocks/databasemocks"
+	"github.com/hyperledger/firefly/mocks/metricsmocks"
+	"github.com/hyperledger/firefly/mocks/operationmocks"
 	"github.com/hyperledger/firefly/pkg/blockchain"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
@@ -31,29 +34,57 @@ import (
 
 type testMultipartyManager struct {
 	multipartyManager
+	mdi *databasemocks.Plugin
 	mbi *blockchainmocks.Plugin
+	mom *operationmocks.Manager
+	mmi *metricsmocks.Manager
+}
+
+func (mp *testMultipartyManager) cleanup(t *testing.T) {
+	mp.mdi.AssertExpectations(t)
+	mp.mbi.AssertExpectations(t)
+	mp.mom.AssertExpectations(t)
+	mp.mmi.AssertExpectations(t)
 }
 
 func newTestMultipartyManager() *testMultipartyManager {
 	nm := &testMultipartyManager{
+		mdi: &databasemocks.Plugin{},
 		mbi: &blockchainmocks.Plugin{},
+		mom: &operationmocks.Manager{},
+		mmi: &metricsmocks.Manager{},
 		multipartyManager: multipartyManager{
-			ctx: context.Background(),
+			ctx:       context.Background(),
+			namespace: "ns1",
 		},
 	}
 
+	nm.multipartyManager.database = nm.mdi
 	nm.multipartyManager.blockchain = nm.mbi
-	nm.mbi.On("Name").Return("ethereum")
-	nm.mbi.On("SubmitBatchPin", context.Background(), mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
-	nm.mbi.On("SubmitNetworkAction", context.Background(), mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	nm.multipartyManager.operations = nm.mom
+	nm.multipartyManager.metrics = nm.mmi
 	return nm
 }
 
 func TestNewMultipartyManager(t *testing.T) {
+	mdi := &databasemocks.Plugin{}
 	mbi := &blockchainmocks.Plugin{}
+	mom := &operationmocks.Manager{}
+	mmi := &metricsmocks.Manager{}
 	contracts := make([]Contract, 0)
-	nm := NewMultipartyManager(context.Background(), "namespace", contracts, mbi)
+	mom.On("RegisterHandler", mock.Anything, mock.Anything, []core.OpType{
+		core.OpTypeBlockchainPinBatch,
+	}).Return()
+	nm, err := NewMultipartyManager(context.Background(), "namespace", contracts, mdi, mbi, mom, mmi)
 	assert.NotNil(t, nm)
+	assert.NoError(t, err)
+	assert.Equal(t, "MultipartyManager", nm.Name())
+}
+
+func TestInitFail(t *testing.T) {
+	contracts := make([]Contract, 0)
+	_, err := NewMultipartyManager(context.Background(), "namespace", contracts, nil, nil, nil, nil)
+	assert.Regexp(t, "FF10128", err)
 }
 
 func TestConfigureContract(t *testing.T) {
@@ -220,12 +251,6 @@ func TestConfigureContractNetworkVersionFail(t *testing.T) {
 	assert.Regexp(t, "pop", err)
 }
 
-func TestGetBlockchainName(t *testing.T) {
-	mp := newTestMultipartyManager()
-	name := mp.Name()
-	assert.Equal(t, "ethereum", name)
-}
-
 func TestSubmitNetworkAction(t *testing.T) {
 	contracts := make([]Contract, 1)
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
@@ -240,6 +265,7 @@ func TestSubmitNetworkAction(t *testing.T) {
 	mp := newTestMultipartyManager()
 	mp.mbi.On("GetNetworkVersion", mock.Anything, mock.Anything).Return(1, nil)
 	mp.mbi.On("AddFireflySubscription", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("test", nil)
+	mp.mbi.On("SubmitNetworkAction", mock.Anything, "test", "0x123", core.NetworkActionTerminate, mock.Anything).Return(nil)
 	mp.multipartyManager.contracts = contracts
 
 	cf := &core.FireFlyContracts{
@@ -252,32 +278,104 @@ func TestSubmitNetworkAction(t *testing.T) {
 	assert.Nil(t, err)
 }
 
-func TestSubmitBatchPin(t *testing.T) {
-	contracts := make([]Contract, 1)
-	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
-		"address": "0x123",
-	}.String())
-	contract := Contract{
-		FirstEvent: "0",
-		Location:   location,
+func TestSubmitBatchPinOk(t *testing.T) {
+	mp := newTestMultipartyManager()
+	defer mp.cleanup(t)
+	ctx := context.Background()
+
+	batch := &core.BatchPersisted{
+		BatchHeader: core.BatchHeader{
+			ID: fftypes.NewUUID(),
+			SignerRef: core.SignerRef{
+				Author: "id1",
+				Key:    "0x12345",
+			},
+		},
+		TX: core.TransactionRef{
+			ID: fftypes.NewUUID(),
+		},
 	}
+	contexts := []*fftypes.Bytes32{}
+
+	mp.mbi.On("Name").Return("ut")
+	mp.mom.On("AddOrReuseOperation", ctx, mock.MatchedBy(func(op *core.Operation) bool {
+		assert.Equal(t, core.OpTypeBlockchainPinBatch, op.Type)
+		assert.Equal(t, "ut", op.Plugin)
+		assert.Equal(t, *batch.TX.ID, *op.Transaction)
+		assert.Equal(t, "payload1", op.Input.GetString("payloadRef"))
+		return true
+	})).Return(nil)
+	mp.mmi.On("IsMetricsEnabled").Return(false)
+	mp.mom.On("RunOperation", mock.Anything, mock.MatchedBy(func(op *core.PreparedOperation) bool {
+		data := op.Data.(batchPinData)
+		return op.Type == core.OpTypeBlockchainPinBatch && data.Batch == batch
+	})).Return(nil, nil)
+
+	err := mp.SubmitBatchPin(ctx, batch, contexts, "payload1")
+	assert.NoError(t, err)
+}
 
-	contracts[0] = contract
+func TestSubmitPinnedBatchWithMetricsOk(t *testing.T) {
 	mp := newTestMultipartyManager()
-	mp.mbi.On("GetNetworkVersion", mock.Anything, mock.Anything).Return(1, nil)
-	mp.mbi.On("AddFireflySubscription", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("test", nil)
-	mp.multipartyManager.contracts = contracts
-
-	cf := &core.FireFlyContracts{
-		Active: core.FireFlyContractInfo{Index: 0},
+	defer mp.cleanup(t)
+	ctx := context.Background()
+
+	batch := &core.BatchPersisted{
+		BatchHeader: core.BatchHeader{
+			ID: fftypes.NewUUID(),
+			SignerRef: core.SignerRef{
+				Author: "id1",
+				Key:    "0x12345",
+			},
+		},
+		TX: core.TransactionRef{
+			ID: fftypes.NewUUID(),
+		},
 	}
+	contexts := []*fftypes.Bytes32{}
+
+	mp.mbi.On("Name").Return("ut")
+	mp.mom.On("AddOrReuseOperation", ctx, mock.MatchedBy(func(op *core.Operation) bool {
+		assert.Equal(t, core.OpTypeBlockchainPinBatch, op.Type)
+		assert.Equal(t, "ut", op.Plugin)
+		assert.Equal(t, *batch.TX.ID, *op.Transaction)
+		assert.Equal(t, "payload1", op.Input.GetString("payloadRef"))
+		return true
+	})).Return(nil)
+	mp.mmi.On("IsMetricsEnabled").Return(true)
+	mp.mmi.On("CountBatchPin").Return()
+	mp.mom.On("RunOperation", mock.Anything, mock.MatchedBy(func(op *core.PreparedOperation) bool {
+		data := op.Data.(batchPinData)
+		return op.Type == core.OpTypeBlockchainPinBatch && data.Batch == batch
+	})).Return(nil, nil)
+
+	err := mp.SubmitBatchPin(ctx, batch, contexts, "payload1")
+	assert.NoError(t, err)
+}
 
-	batchPin := &blockchain.BatchPin{}
+func TestSubmitPinnedBatchOpFail(t *testing.T) {
+	mp := newTestMultipartyManager()
+	defer mp.cleanup(t)
+	ctx := context.Background()
+
+	batch := &core.BatchPersisted{
+		BatchHeader: core.BatchHeader{
+			ID: fftypes.NewUUID(),
+			SignerRef: core.SignerRef{
+				Author: "id1",
+				Key:    "0x12345",
+			},
+		},
+		TX: core.TransactionRef{
+			ID: fftypes.NewUUID(),
+		},
+	}
+	contexts := []*fftypes.Bytes32{}
 
-	err := mp.ConfigureContract(context.Background(), cf)
-	assert.NoError(t, err)
-	err = mp.SubmitBatchPin(context.Background(), "ns1", "0x123", batchPin)
-	assert.Nil(t, err)
+	mp.mbi.On("Name").Return("ut")
+	mp.mom.On("AddOrReuseOperation", ctx, mock.Anything).Return(fmt.Errorf("pop"))
+	err := mp.SubmitBatchPin(ctx, batch, contexts, "payload1")
+	assert.Regexp(t, "pop", err)
 }
 
 func TestGetNetworkVersion(t *testing.T) {
diff --git a/internal/batchpin/operations_test.go b/internal/multiparty/operations_test.go
similarity index 64%
rename from internal/batchpin/operations_test.go
rename to internal/multiparty/operations_test.go
index 5ba65ff89..40d4db315 100644
--- a/internal/batchpin/operations_test.go
+++ b/internal/multiparty/operations_test.go
@@ -13,7 +13,8 @@
 // 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 batchpin
+
+package multiparty
 
 import (
 	"context"
@@ -21,15 +22,14 @@ import (
 	"testing"
 
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
-	"github.com/hyperledger/firefly/mocks/databasemocks"
-	"github.com/hyperledger/firefly/mocks/multipartymocks"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
 )
 
 func TestPrepareAndRunBatchPin(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
+	mp := newTestMultipartyManager()
+	defer mp.cleanup(t)
 
 	op := &core.Operation{
 		Type:      core.OpTypeBlockchainPinBatch,
@@ -50,46 +50,45 @@ func TestPrepareAndRunBatchPin(t *testing.T) {
 	}
 	addBatchPinInputs(op, batch.ID, contexts, "payload1")
 
-	mmp := bp.multiparty.(*multipartymocks.Manager)
-	mdi := bp.database.(*databasemocks.Plugin)
-	mdi.On("GetBatchByID", context.Background(), "ns1", batch.ID).Return(batch, nil)
-	mmp.On("SubmitBatchPin", context.Background(), "ns1:"+op.ID.String(), "0x123", mock.Anything).Return(nil)
+	mp.mdi.On("GetBatchByID", context.Background(), "ns1", batch.ID).Return(batch, nil)
+	mp.mbi.On("SubmitBatchPin", context.Background(), "ns1:"+op.ID.String(), "0x123", mock.Anything, mock.Anything).Return(nil)
 
-	po, err := bp.PrepareOperation(context.Background(), op)
+	po, err := mp.PrepareOperation(context.Background(), op)
 	assert.NoError(t, err)
 	assert.Equal(t, batch, po.Data.(batchPinData).Batch)
 
-	_, complete, err := bp.RunOperation(context.Background(), opBatchPin(op, batch, contexts, "payload1"))
+	_, complete, err := mp.RunOperation(context.Background(), opBatchPin(op, batch, contexts, "payload1"))
 
 	assert.False(t, complete)
 	assert.NoError(t, err)
-
-	mdi.AssertExpectations(t)
 }
 
 func TestPrepareOperationNotSupported(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
+	mp := newTestMultipartyManager()
+	defer mp.cleanup(t)
 
-	po, err := bp.PrepareOperation(context.Background(), &core.Operation{})
+	po, err := mp.PrepareOperation(context.Background(), &core.Operation{})
 
 	assert.Nil(t, po)
 	assert.Regexp(t, "FF10371", err)
 }
 
 func TestPrepareOperationBatchPinBadBatch(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
+	mp := newTestMultipartyManager()
+	defer mp.cleanup(t)
 
 	op := &core.Operation{
 		Type:  core.OpTypeBlockchainPinBatch,
 		Input: fftypes.JSONObject{"batch": "bad"},
 	}
 
-	_, err := bp.PrepareOperation(context.Background(), op)
+	_, err := mp.PrepareOperation(context.Background(), op)
 	assert.Regexp(t, "FF00138", err)
 }
 
 func TestPrepareOperationBatchPinBadContext(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
+	mp := newTestMultipartyManager()
+	defer mp.cleanup(t)
 
 	op := &core.Operation{
 		Type: core.OpTypeBlockchainPinBatch,
@@ -99,21 +98,23 @@ func TestPrepareOperationBatchPinBadContext(t *testing.T) {
 		},
 	}
 
-	_, err := bp.PrepareOperation(context.Background(), op)
+	_, err := mp.PrepareOperation(context.Background(), op)
 	assert.Regexp(t, "FF00107", err)
 }
 
 func TestRunOperationNotSupported(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
+	mp := newTestMultipartyManager()
+	defer mp.cleanup(t)
 
-	_, complete, err := bp.RunOperation(context.Background(), &core.PreparedOperation{})
+	_, complete, err := mp.RunOperation(context.Background(), &core.PreparedOperation{})
 
 	assert.False(t, complete)
 	assert.Regexp(t, "FF10378", err)
 }
 
 func TestPrepareOperationBatchPinError(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
+	mp := newTestMultipartyManager()
+	defer mp.cleanup(t)
 
 	batchID := fftypes.NewUUID()
 	op := &core.Operation{
@@ -124,15 +125,15 @@ func TestPrepareOperationBatchPinError(t *testing.T) {
 		},
 	}
 
-	mdi := bp.database.(*databasemocks.Plugin)
-	mdi.On("GetBatchByID", context.Background(), "ns1", batchID).Return(nil, fmt.Errorf("pop"))
+	mp.mdi.On("GetBatchByID", context.Background(), "ns1", batchID).Return(nil, fmt.Errorf("pop"))
 
-	_, err := bp.PrepareOperation(context.Background(), op)
+	_, err := mp.PrepareOperation(context.Background(), op)
 	assert.EqualError(t, err, "pop")
 }
 
 func TestPrepareOperationBatchPinNotFound(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
+	mp := newTestMultipartyManager()
+	defer mp.cleanup(t)
 
 	batchID := fftypes.NewUUID()
 	op := &core.Operation{
@@ -143,14 +144,14 @@ func TestPrepareOperationBatchPinNotFound(t *testing.T) {
 		},
 	}
 
-	mdi := bp.database.(*databasemocks.Plugin)
-	mdi.On("GetBatchByID", context.Background(), "ns1", batchID).Return(nil, nil)
+	mp.mdi.On("GetBatchByID", context.Background(), "ns1", batchID).Return(nil, nil)
 
-	_, err := bp.PrepareOperation(context.Background(), op)
+	_, err := mp.PrepareOperation(context.Background(), op)
 	assert.Regexp(t, "FF10109", err)
 }
 
 func TestOperationUpdate(t *testing.T) {
-	bp := newTestBatchPinSubmitter(t, false)
-	assert.NoError(t, bp.OnOperationUpdate(context.Background(), nil, nil))
+	mp := newTestMultipartyManager()
+	defer mp.cleanup(t)
+	assert.NoError(t, mp.OnOperationUpdate(context.Background(), nil, nil))
 }
diff --git a/internal/namespace/manager_test.go b/internal/namespace/manager_test.go
index 9ce16f359..a021aa485 100644
--- a/internal/namespace/manager_test.go
+++ b/internal/namespace/manager_test.go
@@ -44,7 +44,6 @@ import (
 	"github.com/hyperledger/firefly/mocks/spieventsmocks"
 	"github.com/hyperledger/firefly/mocks/tokenmocks"
 	"github.com/hyperledger/firefly/pkg/core"
-	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/tokens"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/assert"
@@ -137,7 +136,7 @@ func TestInit(t *testing.T) {
 	nm.utOrchestrator = mo
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", database.GlobalHandler, mock.Anything).Return()
+	nm.mdi.On("SetHandler", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
@@ -174,7 +173,7 @@ func TestInitBlockchainFail(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", database.GlobalHandler, mock.Anything).Return()
+	nm.mdi.On("SetHandler", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(fmt.Errorf("pop"))
 
 	ctx, cancelCtx := context.WithCancel(context.Background())
@@ -189,7 +188,7 @@ func TestInitDataExchangeFail(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", database.GlobalHandler, mock.Anything).Return()
+	nm.mdi.On("SetHandler", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 
@@ -205,7 +204,7 @@ func TestInitSharedStorageFail(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", database.GlobalHandler, mock.Anything).Return()
+	nm.mdi.On("SetHandler", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
@@ -222,7 +221,7 @@ func TestInitTokensFail(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", database.GlobalHandler, mock.Anything).Return()
+	nm.mdi.On("SetHandler", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
@@ -240,7 +239,7 @@ func TestInitEventsFail(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", database.GlobalHandler, mock.Anything).Return()
+	nm.mdi.On("SetHandler", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
@@ -257,14 +256,12 @@ func TestInitOrchestratorFail(t *testing.T) {
 	defer nm.cleanup(t)
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", database.GlobalHandler, mock.Anything).Return()
-	nm.mdi.On("SetHandler", "default", mock.Anything).Return()
+	nm.mdi.On("SetHandler", mock.Anything).Return()
 	nm.mdi.On("GetIdentities", mock.Anything, "default", mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mbi.On("SetHandler", mock.Anything).Return()
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mps.On("SetHandler", "default", mock.Anything).Return()
 	nm.mti.On("Init", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	nm.mev.On("Init", mock.Anything, mock.Anything).Return(nil)
 
@@ -288,7 +285,7 @@ func TestInitVersion1(t *testing.T) {
 	mmp.On("GetNetworkVersion").Return(1)
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", database.GlobalHandler, mock.Anything).Return()
+	nm.mdi.On("SetHandler", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
@@ -321,7 +318,7 @@ func TestInitVersion1Fail(t *testing.T) {
 	mmp.On("GetNetworkVersion").Return(1)
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", database.GlobalHandler, mock.Anything).Return()
+	nm.mdi.On("SetHandler", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
@@ -715,7 +712,7 @@ func TestInitBadNamespace(t *testing.T) {
 	nm.utOrchestrator = &orchestratormocks.Orchestrator{}
 
 	nm.mdi.On("Init", mock.Anything, mock.Anything).Return(nil)
-	nm.mdi.On("SetHandler", database.GlobalHandler, mock.Anything).Return()
+	nm.mdi.On("SetHandler", mock.Anything).Return()
 	nm.mbi.On("Init", mock.Anything, mock.Anything, nm.mmi).Return(nil)
 	nm.mdx.On("Init", mock.Anything, mock.Anything).Return(nil)
 	nm.mps.On("Init", mock.Anything, mock.Anything).Return(nil)
diff --git a/internal/orchestrator/orchestrator_test.go b/internal/orchestrator/orchestrator_test.go
index 7eb810aa6..38e4d9297 100644
--- a/internal/orchestrator/orchestrator_test.go
+++ b/internal/orchestrator/orchestrator_test.go
@@ -25,7 +25,6 @@ import (
 	"github.com/hyperledger/firefly/internal/identity"
 	"github.com/hyperledger/firefly/mocks/assetmocks"
 	"github.com/hyperledger/firefly/mocks/batchmocks"
-	"github.com/hyperledger/firefly/mocks/batchpinmocks"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
 	"github.com/hyperledger/firefly/mocks/broadcastmocks"
 	"github.com/hyperledger/firefly/mocks/contractmocks"
@@ -73,7 +72,6 @@ type testOrchestrator struct {
 	mcm *contractmocks.Manager
 	mmi *metricsmocks.Manager
 	mom *operationmocks.Manager
-	mbp *batchpinmocks.Submitter
 	mth *txcommonmocks.Helper
 	msd *shareddownloadmocks.Manager
 	mae *spieventsmocks.Manager
@@ -99,7 +97,6 @@ func (tor *testOrchestrator) cleanup(t *testing.T) {
 	tor.mcm.AssertExpectations(t)
 	tor.mmi.AssertExpectations(t)
 	tor.mom.AssertExpectations(t)
-	tor.mbp.AssertExpectations(t)
 	tor.mth.AssertExpectations(t)
 	tor.msd.AssertExpectations(t)
 	tor.mae.AssertExpectations(t)
@@ -133,7 +130,6 @@ func newTestOrchestrator() *testOrchestrator {
 		mcm: &contractmocks.Manager{},
 		mmi: &metricsmocks.Manager{},
 		mom: &operationmocks.Manager{},
-		mbp: &batchpinmocks.Submitter{},
 		mth: &txcommonmocks.Helper{},
 		msd: &shareddownloadmocks.Manager{},
 		mae: &spieventsmocks.Manager{},
@@ -152,7 +148,6 @@ func newTestOrchestrator() *testOrchestrator {
 	tor.orchestrator.contracts = tor.mcm
 	tor.orchestrator.metrics = tor.mmi
 	tor.orchestrator.operations = tor.mom
-	tor.orchestrator.batchpin = tor.mbp
 	tor.orchestrator.sharedDownload = tor.msd
 	tor.orchestrator.txHelper = tor.mth
 	tor.orchestrator.definitions = tor.mdh
@@ -191,12 +186,12 @@ func TestNewOrchestrator(t *testing.T) {
 func TestInitOK(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
-	or.mdi.On("SetHandler", "ns", mock.Anything).Return()
+	or.mdi.On("SetHandler", mock.Anything).Return()
 	or.mbi.On("SetHandler", mock.Anything).Return()
 	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return([]*core.Identity{{}}, nil, nil)
-	or.mdx.On("SetHandler", "ns", mock.Anything).Return()
+	or.mdx.On("SetHandler", mock.Anything).Return()
 	or.mdx.On("SetNodes", mock.Anything).Return()
-	or.mps.On("SetHandler", "ns", mock.Anything).Return()
+	or.mps.On("SetHandler", mock.Anything).Return()
 	or.mti.On("SetHandler", "ns", mock.Anything).Return(nil)
 	err := or.Init(or.ctx, or.cancelCtx)
 	assert.NoError(t, err)
@@ -216,12 +211,12 @@ func TestInitOK(t *testing.T) {
 func TestInitTokenListenerFail(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
-	or.mdi.On("SetHandler", "ns", mock.Anything).Return()
+	or.mdi.On("SetHandler", mock.Anything).Return()
 	or.mbi.On("SetHandler", mock.Anything).Return()
 	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return([]*core.Identity{{}}, nil, nil)
-	or.mdx.On("SetHandler", "ns", mock.Anything).Return()
+	or.mdx.On("SetHandler", mock.Anything).Return()
 	or.mdx.On("SetNodes", mock.Anything).Return()
-	or.mps.On("SetHandler", "ns", mock.Anything).Return()
+	or.mps.On("SetHandler", mock.Anything).Return()
 	or.mti.On("SetHandler", "ns", mock.Anything).Return(fmt.Errorf("pop"))
 	err := or.Init(or.ctx, or.cancelCtx)
 	assert.EqualError(t, err, "pop")
@@ -230,9 +225,8 @@ func TestInitTokenListenerFail(t *testing.T) {
 func TestInitDataexchangeNodesFail(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
-	or.mdi.On("SetHandler", "ns", mock.Anything).Return()
+	or.mdi.On("SetHandler", mock.Anything).Return()
 	or.mbi.On("SetHandler", mock.Anything).Return()
-	or.mps.On("SetHandler", "ns", mock.Anything).Return()
 	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
 	ctx := context.Background()
 	err := or.initPlugins(ctx)
@@ -266,11 +260,11 @@ func TestInitNetworkMapComponentFail(t *testing.T) {
 	assert.Regexp(t, "FF10128", err)
 }
 
-func TestInitOperationComponentFail(t *testing.T) {
+func TestInitMultipartyComponentFail(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
 	or.plugins.Database.Plugin = nil
-	or.operations = nil
+	or.multiparty = nil
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -316,7 +310,6 @@ func TestInitIdentityComponentFail(t *testing.T) {
 	defer or.cleanup(t)
 	or.plugins.Database.Plugin = nil
 	or.identity = nil
-	or.multiparty = nil
 	err := or.initComponents(context.Background())
 	assert.Regexp(t, "FF10128", err)
 }
@@ -348,15 +341,6 @@ func TestInitDefinitionsComponentFail(t *testing.T) {
 	assert.Regexp(t, "FF10128", err)
 }
 
-func TestInitBatchPinComponentFail(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	or.plugins.Database.Plugin = nil
-	or.batchpin = nil
-	err := or.initComponents(context.Background())
-	assert.Regexp(t, "FF10128", err)
-}
-
 func TestInitOperationsComponentFail(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
diff --git a/internal/privatemessaging/privatemessaging_test.go b/internal/privatemessaging/privatemessaging_test.go
index 118bb379b..86ff52fe9 100644
--- a/internal/privatemessaging/privatemessaging_test.go
+++ b/internal/privatemessaging/privatemessaging_test.go
@@ -26,13 +26,13 @@ import (
 	"github.com/hyperledger/firefly/internal/batch"
 	"github.com/hyperledger/firefly/internal/coreconfig"
 	"github.com/hyperledger/firefly/mocks/batchmocks"
-	"github.com/hyperledger/firefly/mocks/batchpinmocks"
 	"github.com/hyperledger/firefly/mocks/blockchainmocks"
 	"github.com/hyperledger/firefly/mocks/databasemocks"
 	"github.com/hyperledger/firefly/mocks/dataexchangemocks"
 	"github.com/hyperledger/firefly/mocks/datamocks"
 	"github.com/hyperledger/firefly/mocks/identitymanagermocks"
 	"github.com/hyperledger/firefly/mocks/metricsmocks"
+	"github.com/hyperledger/firefly/mocks/multipartymocks"
 	"github.com/hyperledger/firefly/mocks/operationmocks"
 	"github.com/hyperledger/firefly/mocks/syncasyncmocks"
 	"github.com/hyperledger/firefly/pkg/core"
@@ -53,7 +53,7 @@ func newTestPrivateMessagingCommon(t *testing.T, metricsEnabled bool) (*privateM
 	mba := &batchmocks.Manager{}
 	mdm := &datamocks.Manager{}
 	msa := &syncasyncmocks.Bridge{}
-	mbp := &batchpinmocks.Submitter{}
+	mmp := &multipartymocks.Manager{}
 	mmi := &metricsmocks.Manager{}
 	mom := &operationmocks.Manager{}
 	mockRunAsGroupPassthrough(mdi)
@@ -77,7 +77,7 @@ func newTestPrivateMessagingCommon(t *testing.T, metricsEnabled bool) (*privateM
 	mom.On("RegisterHandler", mock.Anything, mock.Anything, mock.Anything)
 
 	ctx, cancel := context.WithCancel(context.Background())
-	pm, err := NewPrivateMessaging(ctx, "ns1", mdi, mdx, mbi, mim, mba, mdm, msa, mbp, mmi, mom)
+	pm, err := NewPrivateMessaging(ctx, "ns1", mdi, mdx, mbi, mim, mba, mdm, msa, mmp, mmi, mom)
 	assert.NoError(t, err)
 
 	// Default mocks to save boilerplate in the tests
@@ -130,7 +130,7 @@ func TestDispatchBatchWithBlobs(t *testing.T) {
 	blob1 := fftypes.NewRandB32()
 
 	mdi := pm.database.(*databasemocks.Plugin)
-	mbp := pm.batchpin.(*batchpinmocks.Submitter)
+	mmp := pm.multiparty.(*multipartymocks.Manager)
 	mdx := pm.exchange.(*dataexchangemocks.Plugin)
 	mim := pm.identity.(*identitymanagermocks.Manager)
 	mom := pm.operations.(*operationmocks.Manager)
@@ -179,7 +179,7 @@ func TestDispatchBatchWithBlobs(t *testing.T) {
 		return *data.Node.ID == *node2.ID
 	})).Return(nil, nil)
 
-	mbp.On("SubmitPinnedBatch", pm.ctx, mock.Anything, mock.Anything, "").Return(nil)
+	mmp.On("SubmitBatchPin", pm.ctx, mock.Anything, mock.Anything, "").Return(nil)
 
 	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
 		Persisted: core.BatchPersisted{
@@ -205,7 +205,7 @@ func TestDispatchBatchWithBlobs(t *testing.T) {
 	assert.NoError(t, err)
 
 	mdi.AssertExpectations(t)
-	mbp.AssertExpectations(t)
+	mmp.AssertExpectations(t)
 	mdx.AssertExpectations(t)
 	mim.AssertExpectations(t)
 	mom.AssertExpectations(t)
@@ -234,9 +234,6 @@ func TestSendAndSubmitBatchBadID(t *testing.T) {
 	mdi := pm.database.(*databasemocks.Plugin)
 	mdi.On("GetGroupByHash", pm.ctx, "ns1", mock.Anything).Return(nil, fmt.Errorf("pop"))
 
-	mbp := pm.batchpin.(*batchpinmocks.Submitter)
-	mbp.On("SubmitPinnedBatch", pm.ctx, mock.Anything, mock.Anything, "").Return(fmt.Errorf("pop"))
-
 	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
 		Persisted: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
@@ -468,8 +465,8 @@ func TestWriteTransactionSubmitBatchPinFail(t *testing.T) {
 		PayloadRef: "/blob/1",
 	}, nil)
 
-	mbp := pm.batchpin.(*batchpinmocks.Submitter)
-	mbp.On("SubmitPinnedBatch", pm.ctx, mock.Anything, mock.Anything, "").Return(fmt.Errorf("pop"))
+	mmp := pm.multiparty.(*multipartymocks.Manager)
+	mmp.On("SubmitBatchPin", pm.ctx, mock.Anything, mock.Anything, "").Return(fmt.Errorf("pop"))
 
 	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
 		Persisted: core.BatchPersisted{
@@ -489,7 +486,7 @@ func TestWriteTransactionSubmitBatchPinFail(t *testing.T) {
 
 	mdi.AssertExpectations(t)
 	mim.AssertExpectations(t)
-	mbp.AssertExpectations(t)
+	mmp.AssertExpectations(t)
 	mom.AssertExpectations(t)
 }