Repository: hyperledger/firefly
Pull request #916 (2022-08-01 17:44:54)

Add node-specific suffix to DX endpoint "id"
~~In a chain with #915~~
Depends on https://github.com/hyperledger/firefly-dataexchange-https/pull/62

UPDATE: found a number of bugs while adding E2E tests - so #918 contains everything from this PR, some additional fixes, and an E2E test. The comments below are still relevant though.

FireFly can now support many local namespaces that map to a single remote namespace (for multi-tenancy purposes). Each of these local namespaces may have their own "node" within FireFly, but they may all be sharing a single DX plugin. This change adds a "nodeName" qualifier to many of the dataexchange plugin calls, along with a specific implementation in the FFDX plugin code, to allow many nodes to exist behind a single DX.

Implementation outline:
* For newly created nodes, the FFDX connector will append a suffix to the member ID advertised by the plugin before storing or broadcasting it. This means each node will have a unique verifier ID even if they are sharing a DX member ID and certificate. The FFDX connector and FFDX plugin have agreed on `/` as a separator.
* The FFDX connector will ignore this suffix when routing messages (paying attention only to the member ID before the separator) - but will pass the full recipient string for further routing by FireFly on the far end.
* FireFly will use the combination of the received message's remote namespace and recipient ID to route it uniquely to a single event manager.

Should remain fully backwards compatible for nodes that were already broadcast/stored in the database with a non-suffixed peer ID.


Patch:
diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
index d7710304a..c7033c1f8 100644
--- a/internal/blockchain/ethereum/ethereum.go
+++ b/internal/blockchain/ethereum/ethereum.go
@@ -201,13 +201,6 @@ func (e *Ethereum) AddFireflySubscription(ctx context.Context, namespace core.Na
 		return "", err
 	}
 
-	switch firstEvent {
-	case string(core.SubOptsFirstEventOldest):
-		firstEvent = "0"
-	case string(core.SubOptsFirstEventNewest):
-		firstEvent = "latest"
-	}
-
 	version, err := e.GetNetworkVersion(ctx, location)
 	if err != nil {
 		return "", err
diff --git a/internal/data/data_manager.go b/internal/data/data_manager.go
index 18d60f437..d8c2f6f38 100644
--- a/internal/data/data_manager.go
+++ b/internal/data/data_manager.go
@@ -290,7 +290,7 @@ func (dm *dataManager) getMessageData(ctx context.Context, msg *core.Message) (d
 			return nil, false, err
 		}
 		if d == nil {
-			log.L(ctx).Warnf("Message %v data %d mising", msg.Header.ID, i)
+			log.L(ctx).Warnf("Message %v data %d missing", msg.Header.ID, i)
 			foundAll = false
 			continue
 		}
diff --git a/internal/dataexchange/ffdx/dxevent.go b/internal/dataexchange/ffdx/dxevent.go
index f715fb4b8..835819abd 100644
--- a/internal/dataexchange/ffdx/dxevent.go
+++ b/internal/dataexchange/ffdx/dxevent.go
@@ -86,6 +86,7 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 	var namespace string
 	var err error
 	e := &dxEvent{ffdx: h, id: msg.EventID}
+
 	switch msg.Type {
 	case messageFailed:
 		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
@@ -118,23 +119,6 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 			OnComplete:     e.Ack,
 		})
 		return
-	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,
-			}
-		}
 	case blobFailed:
 		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
 			NamespacedOpID: msg.RequestID,
@@ -156,6 +140,35 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 			OnComplete:     e.Ack,
 		})
 		return
+	case blobAcknowledged:
+		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
+			NamespacedOpID: msg.RequestID,
+			Status:         core.OpStatusSucceeded,
+			Output:         msg.Info,
+			VerifyManifest: h.capabilities.Manifest,
+			DXHash:         msg.Hash,
+			OnComplete:     e.Ack,
+		})
+		return
+
+	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,
+			}
+		}
+
 	case blobReceived:
 		var hash *fftypes.Bytes32
 		hash, err = fftypes.ParseBytes32(h.ctx, msg.Hash)
@@ -170,25 +183,16 @@ func (h *FFDX) dispatchEvent(msg *wsEvent) {
 				PayloadRef: msg.Path,
 			}
 		}
-	case blobAcknowledged:
-		h.callbacks.OperationUpdate(h.ctx, &core.OperationUpdate{
-			NamespacedOpID: msg.RequestID,
-			Status:         core.OpStatusSucceeded,
-			Output:         msg.Info,
-			VerifyManifest: h.capabilities.Manifest,
-			DXHash:         msg.Hash,
-			OnComplete:     e.Ack,
-		})
-		return
+
 	default:
 		err = i18n.NewError(h.ctx, coremsgs.MsgUnexpectedDXMessageType, msg.Type)
 	}
 
 	// If we couldn't dispatch the event we received, we still ack it
-	if err == nil {
-		h.callbacks.DXEvent(h.ctx, namespace, e)
-	} else {
+	if err != nil {
 		log.L(h.ctx).Warnf("Failed to dispatch DX event: %s", err)
 		e.Ack()
+	} else {
+		h.callbacks.DXEvent(h.ctx, namespace, msg.Recipient, e)
 	}
 }
diff --git a/internal/dataexchange/ffdx/ffdx.go b/internal/dataexchange/ffdx/ffdx.go
index d37be52e2..6317a8f8a 100644
--- a/internal/dataexchange/ffdx/ffdx.go
+++ b/internal/dataexchange/ffdx/ffdx.go
@@ -21,8 +21,6 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
-	"net/http"
-	"strconv"
 	"strings"
 	"sync"
 
@@ -38,6 +36,8 @@ import (
 	"github.com/hyperledger/firefly/pkg/dataexchange"
 )
 
+const DXIDSeparator = "/"
+
 type FFDX struct {
 	ctx          context.Context
 	capabilities *dataexchange.Capabilities
@@ -47,12 +47,17 @@ type FFDX struct {
 	needsInit    bool
 	initialized  bool
 	initMutex    sync.Mutex
-	nodes        []fftypes.JSONObject
+	nodes        map[string]*dxNode
 	ackChannel   chan *ack
 }
 
+type dxNode struct {
+	Name string
+	Peer fftypes.JSONObject
+}
+
 type callbacks struct {
-	plugin     dataexchange.Plugin
+	plugin     *FFDX
 	handlers   map[string]dataexchange.Callbacks
 	opHandlers map[string]core.OperationCallbacks
 }
@@ -67,11 +72,22 @@ func (cb *callbacks) OperationUpdate(ctx context.Context, update *core.Operation
 	}
 }
 
-func (cb *callbacks) DXEvent(ctx context.Context, namespace string, event dataexchange.DXEvent) {
-	if handler, ok := cb.handlers[namespace]; ok {
-		handler.DXEvent(cb.plugin, event)
+func (cb *callbacks) DXEvent(ctx context.Context, namespace, recipient string, event dataexchange.DXEvent) {
+	cb.plugin.initMutex.Lock()
+	nodeKey := namespace + ":" + recipient
+	node, ok := cb.plugin.nodes[nodeKey]
+	cb.plugin.initMutex.Unlock()
+
+	if ok {
+		key := namespace + ":" + node.Name
+		if handler, ok := cb.handlers[key]; ok {
+			handler.DXEvent(cb.plugin, event)
+		} else {
+			log.L(ctx).Errorf("No handler found for DX event '%s' namespace=%s node=%s", event.EventID(), namespace, node.Name)
+			event.Ack()
+		}
 	} else {
-		log.L(ctx).Errorf("No handler found for DX event '%s'", event.EventID())
+		log.L(ctx).Errorf("Unknown local node for DX event '%s' recipient=%s", event.EventID(), recipient)
 		event.Ack()
 	}
 }
@@ -94,11 +110,6 @@ func joinBlobPath(namespace, id string) string {
 	return fmt.Sprintf("%s/%s", namespace, id)
 }
 
-const (
-	dxHTTPHeaderHash = "dx-hash"
-	dxHTTPHeaderSize = "dx-size"
-)
-
 type msgType string
 
 const (
@@ -126,12 +137,14 @@ type sendMessage struct {
 	Message   string `json:"message"`
 	Recipient string `json:"recipient"`
 	RequestID string `json:"requestId"`
+	Sender    string `json:"sender"`
 }
 
 type transferBlob struct {
 	Path      string `json:"path"`
 	Recipient string `json:"recipient"`
 	RequestID string `json:"requestId"`
+	Sender    string `json:"sender"`
 }
 
 type wsAck struct {
@@ -162,6 +175,7 @@ func (h *FFDX) Init(ctx context.Context, config config.Section) (err error) {
 		opHandlers: make(map[string]core.OperationCallbacks),
 	}
 	h.needsInit = config.GetBool(DataExchangeInitEnabled)
+	h.nodes = make(map[string]*dxNode)
 
 	if config.GetString(ffresty.HTTPConfigURL) == "" {
 		return i18n.NewError(ctx, coremsgs.MsgMissingPluginConfig, "url", "dataexchange.ffdx")
@@ -183,12 +197,9 @@ func (h *FFDX) Init(ctx context.Context, config config.Section) (err error) {
 	return nil
 }
 
-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(remoteNamespace, nodeName string, handler dataexchange.Callbacks) {
+	key := remoteNamespace + ":" + nodeName
+	h.callbacks.handlers[key] = handler
 }
 
 func (h *FFDX) SetOperationHandler(namespace string, handler core.OperationCallbacks) {
@@ -210,8 +221,12 @@ func (h *FFDX) beforeConnect(ctx context.Context) error {
 	if h.needsInit {
 		h.initialized = false
 		var status dxStatus
+		var body []fftypes.JSONObject
+		for _, node := range h.nodes {
+			body = append(body, node.Peer)
+		}
 		res, err := h.client.R().SetContext(ctx).
-			SetBody(h.nodes).
+			SetBody(body).
 			SetResult(&status).
 			Post("/api/v1/init")
 		if err != nil || !res.IsSuccess() {
@@ -235,11 +250,15 @@ func (h *FFDX) checkInitialized(ctx context.Context) error {
 	return nil
 }
 
-func (h *FFDX) GetEndpointInfo(ctx context.Context) (peer fftypes.JSONObject, err error) {
-	if err := h.checkInitialized(ctx); err != nil {
-		return peer, err
+func (h *FFDX) GetPeerID(peer fftypes.JSONObject) string {
+	id := peer.GetString("nodeID")
+	if id == "" {
+		id = peer.GetString("id")
 	}
+	return id
+}
 
+func (h *FFDX) GetEndpointInfo(ctx context.Context, nodeName string) (peer fftypes.JSONObject, err error) {
 	res, err := h.client.R().SetContext(ctx).
 		SetResult(&peer).
 		Get("/api/v1/id")
@@ -251,21 +270,29 @@ func (h *FFDX) GetEndpointInfo(ctx context.Context) (peer fftypes.JSONObject, er
 		log.L(ctx).Errorf("Invalid DX info: %s", peer.String())
 		return nil, i18n.NewError(ctx, coremsgs.MsgDXInfoMissingID)
 	}
-	h.nodes = append(h.nodes, peer)
+	peer["nodeID"] = fmt.Sprintf("%s%s%s", id, DXIDSeparator, nodeName)
 	return peer, nil
 }
 
-func (h *FFDX) AddPeer(ctx context.Context, peer fftypes.JSONObject) (err error) {
-	if err := h.checkInitialized(ctx); err != nil {
-		return err
+func (h *FFDX) AddNode(ctx context.Context, remoteNamespace, nodeName string, peer fftypes.JSONObject) (err error) {
+	h.initMutex.Lock()
+	defer h.initMutex.Unlock()
+
+	key := remoteNamespace + ":" + h.GetPeerID(peer)
+	h.nodes[key] = &dxNode{
+		Peer: peer,
+		Name: nodeName,
 	}
 
-	res, err := h.client.R().SetContext(ctx).
-		SetBody(peer).
-		Put(fmt.Sprintf("/api/v1/peers/%s", peer.GetString("id")))
-	if err != nil || !res.IsSuccess() {
-		return ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgDXRESTErr)
+	if h.initialized {
+		res, err := h.client.R().SetContext(ctx).
+			SetBody(peer).
+			Put(fmt.Sprintf("/api/v1/peers/%s", peer.GetString("id")))
+		if err != nil || !res.IsSuccess() {
+			return ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgDXRESTErr)
+		}
 	}
+
 	return nil
 }
 
@@ -299,7 +326,7 @@ func (h *FFDX) DownloadBlob(ctx context.Context, payloadRef string) (content io.
 	return res.RawBody(), nil
 }
 
-func (h *FFDX) SendMessage(ctx context.Context, nsOpID, peerID string, data []byte) (err error) {
+func (h *FFDX) SendMessage(ctx context.Context, nsOpID string, peer, sender fftypes.JSONObject, data []byte) (err error) {
 	if err := h.checkInitialized(ctx); err != nil {
 		return err
 	}
@@ -308,8 +335,9 @@ func (h *FFDX) SendMessage(ctx context.Context, nsOpID, peerID string, data []by
 	res, err := h.client.R().SetContext(ctx).
 		SetBody(&sendMessage{
 			Message:   string(data),
-			Recipient: peerID,
+			Recipient: h.GetPeerID(peer),
 			RequestID: nsOpID,
+			Sender:    h.GetPeerID(sender),
 		}).
 		SetResult(&responseData).
 		Post("/api/v1/messages")
@@ -319,7 +347,7 @@ func (h *FFDX) SendMessage(ctx context.Context, nsOpID, peerID string, data []by
 	return nil
 }
 
-func (h *FFDX) TransferBlob(ctx context.Context, nsOpID, peerID, payloadRef string) (err error) {
+func (h *FFDX) TransferBlob(ctx context.Context, nsOpID string, peer, sender fftypes.JSONObject, payloadRef string) (err error) {
 	if err := h.checkInitialized(ctx); err != nil {
 		return err
 	}
@@ -328,8 +356,9 @@ func (h *FFDX) TransferBlob(ctx context.Context, nsOpID, peerID, payloadRef stri
 	res, err := h.client.R().SetContext(ctx).
 		SetBody(&transferBlob{
 			Path:      fmt.Sprintf("/%s", payloadRef),
-			Recipient: peerID,
+			Recipient: h.GetPeerID(peer),
 			RequestID: nsOpID,
+			Sender:    h.GetPeerID(sender),
 		}).
 		SetResult(&responseData).
 		Post("/api/v1/transfers")
@@ -339,30 +368,6 @@ func (h *FFDX) TransferBlob(ctx context.Context, nsOpID, peerID, payloadRef stri
 	return nil
 }
 
-func (h *FFDX) CheckBlobReceived(ctx context.Context, peerID, ns string, id fftypes.UUID) (hash *fftypes.Bytes32, size int64, err error) {
-	var responseData responseWithRequestID
-	res, err := h.client.R().SetContext(ctx).
-		SetResult(&responseData).
-		Head(fmt.Sprintf("/api/v1/blobs/%s/%s/%s", peerID, ns, id.String()))
-	if err == nil && res.StatusCode() == http.StatusNotFound {
-		return nil, -1, nil
-	}
-	if err != nil || !res.IsSuccess() {
-		return nil, -1, ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgDXRESTErr)
-	}
-	hashString := res.Header().Get(dxHTTPHeaderHash)
-	if hash, err = fftypes.ParseBytes32(ctx, hashString); err != nil {
-		return nil, -1, i18n.WrapError(ctx, err, coremsgs.MsgDXBadResponse, "hash", hashString)
-	}
-	sizeString := res.Header().Get(dxHTTPHeaderSize)
-	if sizeString != "" {
-		if size, err = strconv.ParseInt(sizeString, 10, 64); err != nil {
-			return nil, -1, i18n.WrapError(ctx, err, coremsgs.MsgDXBadResponse, "size", sizeString)
-		}
-	}
-	return hash, size, nil
-}
-
 func (h *FFDX) ackLoop() {
 	for {
 		select {
diff --git a/internal/definitions/handler.go b/internal/definitions/handler.go
index 637a0766f..5515495a8 100644
--- a/internal/definitions/handler.go
+++ b/internal/definitions/handler.go
@@ -77,33 +77,35 @@ func (dma DefinitionMessageAction) String() string {
 }
 
 type definitionHandler struct {
-	namespace  string
-	multiparty bool
-	database   database.Plugin
-	blockchain blockchain.Plugin   // optional
-	exchange   dataexchange.Plugin // optional
-	data       data.Manager
-	identity   identity.Manager
-	assets     assets.Manager
-	contracts  contracts.Manager // optional
-	tokenNames map[string]string // mapping of token connector remote name => name
+	namespace       string
+	remoteNamespace string
+	multiparty      bool
+	database        database.Plugin
+	blockchain      blockchain.Plugin   // optional
+	exchange        dataexchange.Plugin // optional
+	data            data.Manager
+	identity        identity.Manager
+	assets          assets.Manager
+	contracts       contracts.Manager // optional
+	tokenNames      map[string]string // mapping of token connector remote name => name
 }
 
-func newDefinitionHandler(ctx context.Context, ns string, multiparty bool, di database.Plugin, bi blockchain.Plugin, dx dataexchange.Plugin, dm data.Manager, im identity.Manager, am assets.Manager, cm contracts.Manager, tokenNames map[string]string) (*definitionHandler, error) {
+func newDefinitionHandler(ctx context.Context, ns, remoteNS string, multiparty bool, di database.Plugin, bi blockchain.Plugin, dx dataexchange.Plugin, dm data.Manager, im identity.Manager, am assets.Manager, cm contracts.Manager, tokenNames map[string]string) (*definitionHandler, error) {
 	if di == nil || dm == nil || im == nil || am == nil {
 		return nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "DefinitionHandler")
 	}
 	return &definitionHandler{
-		namespace:  ns,
-		multiparty: multiparty,
-		database:   di,
-		blockchain: bi,
-		exchange:   dx,
-		data:       dm,
-		identity:   im,
-		assets:     am,
-		contracts:  cm,
-		tokenNames: tokenNames,
+		namespace:       ns,
+		remoteNamespace: remoteNS,
+		multiparty:      multiparty,
+		database:        di,
+		blockchain:      bi,
+		exchange:        dx,
+		data:            dm,
+		identity:        im,
+		assets:          am,
+		contracts:       cm,
+		tokenNames:      tokenNames,
 	}, nil
 }
 
diff --git a/internal/definitions/handler_identity_claim.go b/internal/definitions/handler_identity_claim.go
index 2644e73ea..784d6bea8 100644
--- a/internal/definitions/handler_identity_claim.go
+++ b/internal/definitions/handler_identity_claim.go
@@ -89,7 +89,7 @@ func (dh *definitionHandler) getClaimVerifier(msg *identityMsgInfo, identity *co
 	switch identity.Type {
 	case core.IdentityTypeNode:
 		verifier.VerifierRef.Type = core.VerifierTypeFFDXPeerID
-		verifier.VerifierRef.Value = identity.Profile.GetString("id")
+		verifier.VerifierRef.Value = dh.exchange.GetPeerID(identity.Profile)
 	default:
 		verifier.VerifierRef.Type = dh.blockchain.VerifierType()
 		verifier.VerifierRef.Value = msg.Key
@@ -234,7 +234,7 @@ func (dh *definitionHandler) handleIdentityClaim(ctx context.Context, state *cor
 		state.AddPreFinalize(
 			func(ctx context.Context) error {
 				// Tell the data exchange about this node. Treat these errors like database errors - and return for retry processing
-				return dh.exchange.AddPeer(ctx, identity.Profile)
+				return dh.exchange.AddNode(ctx, dh.remoteNamespace, identity.Name, identity.Profile)
 			})
 	}
 
diff --git a/internal/definitions/sender.go b/internal/definitions/sender.go
index ea67c1585..aad58aed9 100644
--- a/internal/definitions/sender.go
+++ b/internal/definitions/sender.go
@@ -71,7 +71,7 @@ func fakeBatch(ctx context.Context, handler func(context.Context, *core.BatchSta
 	return err
 }
 
-func NewDefinitionSender(ctx context.Context, ns string, multiparty bool, di database.Plugin, bi blockchain.Plugin, dx dataexchange.Plugin, bm broadcast.Manager, im identity.Manager, dm data.Manager, am assets.Manager, cm contracts.Manager, tokenRemoteNames map[string]string) (Sender, Handler, error) {
+func NewDefinitionSender(ctx context.Context, ns, remoteNS string, multiparty bool, di database.Plugin, bi blockchain.Plugin, dx dataexchange.Plugin, bm broadcast.Manager, im identity.Manager, dm data.Manager, am assets.Manager, cm contracts.Manager, tokenRemoteNames map[string]string) (Sender, Handler, error) {
 	if di == nil || im == nil || dm == nil {
 		return nil, nil, i18n.NewError(ctx, coremsgs.MsgInitializationNilDepError, "DefinitionSender")
 	}
@@ -86,7 +86,7 @@ func NewDefinitionSender(ctx context.Context, ns string, multiparty bool, di dat
 		contracts:        cm,
 		tokenRemoteNames: tokenRemoteNames,
 	}
-	dh, err := newDefinitionHandler(ctx, ns, multiparty, di, bi, dx, dm, im, am, cm, reverseMap(tokenRemoteNames))
+	dh, err := newDefinitionHandler(ctx, ns, remoteNS, multiparty, di, bi, dx, dm, im, am, cm, reverseMap(tokenRemoteNames))
 	ds.handler = dh
 	return ds, dh, err
 }
diff --git a/internal/identity/identitymanager.go b/internal/identity/identitymanager.go
index 38dd73318..98ce8709d 100644
--- a/internal/identity/identitymanager.go
+++ b/internal/identity/identitymanager.go
@@ -61,7 +61,7 @@ type identityManager struct {
 	namespace              string
 	defaultKey             string
 	multipartyRootVerifier *core.VerifierRef
-	localNodeID            *fftypes.UUID
+	localNode              *core.Identity
 	identityCacheTTL       time.Duration
 	identityCache          *ccache.Cache
 	signingKeyCacheTTL     time.Duration
@@ -102,14 +102,13 @@ func ParseKeyNormalizationConfig(strConfigVal string) int {
 }
 
 func (im *identityManager) GetLocalNode(ctx context.Context) (node *core.Identity, err error) {
-	if im.localNodeID != nil {
-		return im.CachedIdentityLookupByID(ctx, im.localNodeID)
+	if im.localNode != nil {
+		return im.localNode, nil
 	}
 	nodeName := im.multiparty.LocalNode().Name
 	node, err = im.database.GetIdentityByName(ctx, core.IdentityTypeNode, im.namespace, nodeName)
 	if err == nil && node != nil {
-		im.localNodeID = node.ID
-		im.addToIdentityCache(node)
+		im.localNode = node
 	}
 	return node, err
 }
@@ -493,11 +492,6 @@ func (im *identityManager) CachedIdentityLookupMustExist(ctx context.Context, di
 	return identity, false, nil
 }
 
-func (im *identityManager) addToIdentityCache(identity *core.Identity) {
-	cacheKey := fmt.Sprintf("id=%s", identity.ID)
-	im.identityCache.Set(cacheKey, identity, im.identityCacheTTL)
-}
-
 func (im *identityManager) CachedIdentityLookupByID(ctx context.Context, id *fftypes.UUID) (identity *core.Identity, err error) {
 	// Use an LRU cache for the author identity, as it's likely for the same identity to be re-used over and over
 	cacheKey := fmt.Sprintf("id=%s", id)
@@ -510,7 +504,7 @@ func (im *identityManager) CachedIdentityLookupByID(ctx context.Context, id *fft
 			return identity, err
 		}
 		// Cache the result
-		im.addToIdentityCache(identity)
+		im.identityCache.Set(cacheKey, identity, im.identityCacheTTL)
 	}
 	return identity, nil
 }
diff --git a/internal/networkmap/register_node.go b/internal/networkmap/register_node.go
index 7f516c9a0..7a3d3f23b 100644
--- a/internal/networkmap/register_node.go
+++ b/internal/networkmap/register_node.go
@@ -44,11 +44,10 @@ func (nm *networkMap) RegisterNode(ctx context.Context, waitConfirm bool) (ident
 		},
 	}
 
-	dxInfo, err := nm.exchange.GetEndpointInfo(ctx)
+	nodeRequest.Profile, err = nm.exchange.GetEndpointInfo(ctx, localNodeName)
 	if err != nil {
 		return nil, err
 	}
-	nodeRequest.Profile = dxInfo
 
 	return nm.RegisterIdentity(ctx, nodeRequest, waitConfirm)
 }
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index 5f44e3b0d..204a709b8 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -370,12 +370,13 @@ func (or *orchestrator) initHandlers(ctx context.Context) (err error) {
 		if err != nil {
 			return err
 		}
-		nodeInfo := make([]fftypes.JSONObject, len(nodes))
-		for i, node := range nodes {
-			nodeInfo[i] = node.Profile
+		for _, node := range nodes {
+			err = or.plugins.DataExchange.Plugin.AddNode(ctx, or.namespace.RemoteName, node.Name, node.Profile)
+			if err != nil {
+				return err
+			}
 		}
-		or.plugins.DataExchange.Plugin.SetNodes(nodeInfo)
-		or.plugins.DataExchange.Plugin.SetHandler(or.namespace.RemoteName, or.events)
+		or.plugins.DataExchange.Plugin.SetHandler(or.namespace.RemoteName, or.config.Multiparty.Node.Name, or.events)
 		or.plugins.DataExchange.Plugin.SetOperationHandler(or.namespace.LocalName, &or.bc)
 	}
 
@@ -465,7 +466,7 @@ func (or *orchestrator) initManagers(ctx context.Context) (err error) {
 	}
 
 	if or.defsender == nil {
-		or.defsender, or.defhandler, err = definitions.NewDefinitionSender(ctx, or.namespace.LocalName, or.config.Multiparty.Enabled, or.database(), or.blockchain(), or.dataexchange(), or.broadcast, or.identity, or.data, or.assets, or.contracts, or.config.TokenRemoteNames)
+		or.defsender, or.defhandler, err = definitions.NewDefinitionSender(ctx, or.namespace.LocalName, or.namespace.RemoteName, or.config.Multiparty.Enabled, or.database(), or.blockchain(), or.dataexchange(), or.broadcast, or.identity, or.data, or.assets, or.contracts, or.config.TokenRemoteNames)
 		if err != nil {
 			return err
 		}
diff --git a/internal/privatemessaging/operations.go b/internal/privatemessaging/operations.go
index 6fc4de3c9..5c1243c2a 100644
--- a/internal/privatemessaging/operations.go
+++ b/internal/privatemessaging/operations.go
@@ -129,14 +129,23 @@ func (pm *privateMessaging) PrepareOperation(ctx context.Context, op *core.Opera
 func (pm *privateMessaging) RunOperation(ctx context.Context, op *core.PreparedOperation) (outputs fftypes.JSONObject, complete bool, err error) {
 	switch data := op.Data.(type) {
 	case transferBlobData:
-		return nil, false, pm.exchange.TransferBlob(ctx, op.NamespacedIDString(), data.Node.Profile.GetString("id"), data.Blob.PayloadRef)
+		localNode, err := pm.identity.GetLocalNode(ctx)
+		if err != nil {
+			return nil, false, err
+		}
+		return nil, false, pm.exchange.TransferBlob(ctx, op.NamespacedIDString(), data.Node.Profile, localNode.Profile, data.Blob.PayloadRef)
 
 	case batchSendData:
+		localNode, err := pm.identity.GetLocalNode(ctx)
+		if err != nil {
+			return nil, false, err
+		}
+
 		payload, err := json.Marshal(data.Transport)
 		if err != nil {
 			return nil, false, i18n.WrapError(ctx, err, coremsgs.MsgSerializationFailed)
 		}
-		return nil, false, pm.exchange.SendMessage(ctx, op.NamespacedIDString(), data.Node.Profile.GetString("id"), payload)
+		return nil, false, pm.exchange.SendMessage(ctx, op.NamespacedIDString(), data.Node.Profile, localNode.Profile, payload)
 
 	default:
 		return nil, false, i18n.NewError(ctx, coremsgs.MsgOperationDataIncorrect, op.Data)
diff --git a/manifest.json b/manifest.json
index aaa506e5f..48ce43099 100644
--- a/manifest.json
+++ b/manifest.json
@@ -11,8 +11,8 @@
   },
   "dataexchange-https": {
     "image": "ghcr.io/hyperledger/firefly-dataexchange-https",
-    "tag": "v1.0.0",
-    "sha": "1a19cdbdc3311930c59a4a8089db8ba843a9267cffc79c7a22331d373150e94f"
+    "tag": "v1.0.0-20220801-11",
+    "sha": "f113ccbd1ad166c3a382ec744af93b0701e1741fc97be2f4c60c3edb29767570"
   },
   "tokens-erc1155": {
     "image": "ghcr.io/hyperledger/firefly-tokens-erc1155",
diff --git a/mocks/dataexchangemocks/plugin.go b/mocks/dataexchangemocks/plugin.go
index 450185f35..725a72968 100644
--- a/mocks/dataexchangemocks/plugin.go
+++ b/mocks/dataexchangemocks/plugin.go
@@ -23,13 +23,13 @@ type Plugin struct {
 	mock.Mock
 }
 
-// AddPeer provides a mock function with given fields: ctx, peer
-func (_m *Plugin) AddPeer(ctx context.Context, peer fftypes.JSONObject) error {
-	ret := _m.Called(ctx, peer)
+// AddNode provides a mock function with given fields: ctx, remoteNamespace, nodeName, peer
+func (_m *Plugin) AddNode(ctx context.Context, remoteNamespace string, nodeName string, peer fftypes.JSONObject) error {
+	ret := _m.Called(ctx, remoteNamespace, nodeName, peer)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, fftypes.JSONObject) error); ok {
-		r0 = rf(ctx, peer)
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, fftypes.JSONObject) error); ok {
+		r0 = rf(ctx, remoteNamespace, nodeName, peer)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -53,36 +53,6 @@ func (_m *Plugin) Capabilities() *dataexchange.Capabilities {
 	return r0
 }
 
-// CheckBlobReceived provides a mock function with given fields: ctx, peerID, ns, id
-func (_m *Plugin) CheckBlobReceived(ctx context.Context, peerID string, ns string, id fftypes.UUID) (*fftypes.Bytes32, int64, error) {
-	ret := _m.Called(ctx, peerID, ns, id)
-
-	var r0 *fftypes.Bytes32
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, fftypes.UUID) *fftypes.Bytes32); ok {
-		r0 = rf(ctx, peerID, ns, id)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*fftypes.Bytes32)
-		}
-	}
-
-	var r1 int64
-	if rf, ok := ret.Get(1).(func(context.Context, string, string, fftypes.UUID) int64); ok {
-		r1 = rf(ctx, peerID, ns, id)
-	} else {
-		r1 = ret.Get(1).(int64)
-	}
-
-	var r2 error
-	if rf, ok := ret.Get(2).(func(context.Context, string, string, fftypes.UUID) error); ok {
-		r2 = rf(ctx, peerID, ns, id)
-	} else {
-		r2 = ret.Error(2)
-	}
-
-	return r0, r1, r2
-}
-
 // DownloadBlob provides a mock function with given fields: ctx, payloadRef
 func (_m *Plugin) DownloadBlob(ctx context.Context, payloadRef string) (io.ReadCloser, error) {
 	ret := _m.Called(ctx, payloadRef)
@@ -106,13 +76,13 @@ func (_m *Plugin) DownloadBlob(ctx context.Context, payloadRef string) (io.ReadC
 	return r0, r1
 }
 
-// GetEndpointInfo provides a mock function with given fields: ctx
-func (_m *Plugin) GetEndpointInfo(ctx context.Context) (fftypes.JSONObject, error) {
-	ret := _m.Called(ctx)
+// GetEndpointInfo provides a mock function with given fields: ctx, nodeName
+func (_m *Plugin) GetEndpointInfo(ctx context.Context, nodeName string) (fftypes.JSONObject, error) {
+	ret := _m.Called(ctx, nodeName)
 
 	var r0 fftypes.JSONObject
-	if rf, ok := ret.Get(0).(func(context.Context) fftypes.JSONObject); ok {
-		r0 = rf(ctx)
+	if rf, ok := ret.Get(0).(func(context.Context, string) fftypes.JSONObject); ok {
+		r0 = rf(ctx, nodeName)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).(fftypes.JSONObject)
@@ -120,8 +90,8 @@ func (_m *Plugin) GetEndpointInfo(ctx context.Context) (fftypes.JSONObject, erro
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context) error); ok {
-		r1 = rf(ctx)
+	if rf, ok := ret.Get(1).(func(context.Context, string) error); ok {
+		r1 = rf(ctx, nodeName)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -129,6 +99,20 @@ func (_m *Plugin) GetEndpointInfo(ctx context.Context) (fftypes.JSONObject, erro
 	return r0, r1
 }
 
+// GetPeerID provides a mock function with given fields: peer
+func (_m *Plugin) GetPeerID(peer fftypes.JSONObject) string {
+	ret := _m.Called(peer)
+
+	var r0 string
+	if rf, ok := ret.Get(0).(func(fftypes.JSONObject) string); ok {
+		r0 = rf(peer)
+	} else {
+		r0 = ret.Get(0).(string)
+	}
+
+	return r0
+}
+
 // Init provides a mock function with given fields: ctx, _a1
 func (_m *Plugin) Init(ctx context.Context, _a1 config.Section) error {
 	ret := _m.Called(ctx, _a1)
@@ -162,13 +146,13 @@ func (_m *Plugin) Name() string {
 	return r0
 }
 
-// SendMessage provides a mock function with given fields: ctx, nsOpID, peerID, data
-func (_m *Plugin) SendMessage(ctx context.Context, nsOpID string, peerID string, data []byte) error {
-	ret := _m.Called(ctx, nsOpID, peerID, data)
+// SendMessage provides a mock function with given fields: ctx, nsOpID, peer, sender, data
+func (_m *Plugin) SendMessage(ctx context.Context, nsOpID string, peer fftypes.JSONObject, sender fftypes.JSONObject, data []byte) error {
+	ret := _m.Called(ctx, nsOpID, peer, sender, data)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, []byte) error); ok {
-		r0 = rf(ctx, nsOpID, peerID, data)
+	if rf, ok := ret.Get(0).(func(context.Context, string, fftypes.JSONObject, fftypes.JSONObject, []byte) error); ok {
+		r0 = rf(ctx, nsOpID, peer, sender, data)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -176,14 +160,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)
-}
-
-// SetNodes provides a mock function with given fields: nodes
-func (_m *Plugin) SetNodes(nodes []fftypes.JSONObject) {
-	_m.Called(nodes)
+// SetHandler provides a mock function with given fields: remoteNamespace, nodeName, handler
+func (_m *Plugin) SetHandler(remoteNamespace string, nodeName string, handler dataexchange.Callbacks) {
+	_m.Called(remoteNamespace, nodeName, handler)
 }
 
 // SetOperationHandler provides a mock function with given fields: namespace, handler
@@ -205,13 +184,13 @@ func (_m *Plugin) Start() error {
 	return r0
 }
 
-// TransferBlob provides a mock function with given fields: ctx, nsOpID, peerID, payloadRef
-func (_m *Plugin) TransferBlob(ctx context.Context, nsOpID string, peerID string, payloadRef string) error {
-	ret := _m.Called(ctx, nsOpID, peerID, payloadRef)
+// TransferBlob provides a mock function with given fields: ctx, nsOpID, peer, sender, payloadRef
+func (_m *Plugin) TransferBlob(ctx context.Context, nsOpID string, peer fftypes.JSONObject, sender fftypes.JSONObject, payloadRef string) error {
+	ret := _m.Called(ctx, nsOpID, peer, sender, payloadRef)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, string) error); ok {
-		r0 = rf(ctx, nsOpID, peerID, payloadRef)
+	if rf, ok := ret.Get(0).(func(context.Context, string, fftypes.JSONObject, fftypes.JSONObject, string) error); ok {
+		r0 = rf(ctx, nsOpID, peer, sender, payloadRef)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/pkg/dataexchange/plugin.go b/pkg/dataexchange/plugin.go
index f301d06c4..36e59d4f6 100644
--- a/pkg/dataexchange/plugin.go
+++ b/pkg/dataexchange/plugin.go
@@ -63,12 +63,9 @@ type Plugin interface {
 	// Init initializes the plugin, with configuration
 	Init(ctx context.Context, config config.Section) error
 
-	// SetNodes initializes the known nodes from the database
-	SetNodes(nodes []fftypes.JSONObject)
-
 	// SetHandler registers a handler to receive callbacks
-	// Plugin will attempt (but is not guaranteed) to deliver events only for the given namespace
-	SetHandler(namespace string, handler Callbacks)
+	// Plugin will attempt (but is not guaranteed) to deliver events only for the given namespace and node
+	SetHandler(remoteNamespace, nodeName string, handler Callbacks)
 
 	// SetOperationHandler registers a handler to receive async operation status
 	// If namespace is set, plugin will attempt to deliver only events for that namespace
@@ -81,10 +78,11 @@ type Plugin interface {
 	Capabilities() *Capabilities
 
 	// GetEndpointInfo returns the information about the local endpoint
-	GetEndpointInfo(ctx context.Context) (peer fftypes.JSONObject, err error)
+	GetEndpointInfo(ctx context.Context, nodeName string) (peer fftypes.JSONObject, err error)
 
-	// AddPeer translates the configuration published by another peer, into a reference string that is used between DX and FireFly to refer to the peer
-	AddPeer(ctx context.Context, peer fftypes.JSONObject) (err error)
+	// AddNode registers details on a node in the multiparty network
+	// This may be information loaded from the database at init, or received in flight while running
+	AddNode(ctx context.Context, remoteNamespace, nodeName string, peer fftypes.JSONObject) (err error)
 
 	// UploadBlob streams a blob to storage, and returns the hash to confirm the hash calculated in Core matches the hash calculated in the plugin
 	UploadBlob(ctx context.Context, ns string, id fftypes.UUID, content io.Reader) (payloadRef string, hash *fftypes.Bytes32, size int64, err error)
@@ -92,15 +90,15 @@ type Plugin interface {
 	// DownloadBlob streams a received blob out of storage
 	DownloadBlob(ctx context.Context, payloadRef string) (content io.ReadCloser, err error)
 
-	// CheckBlobReceived confirms that a blob with the specified hash has been received from the specified peer
-	CheckBlobReceived(ctx context.Context, peerID, ns string, id fftypes.UUID) (hash *fftypes.Bytes32, size int64, err error)
-
 	// SendMessage sends an in-line package of data to another network node.
 	// Should return as quickly as possible for parallelism, then report completion asynchronously via the operation ID
-	SendMessage(ctx context.Context, nsOpID, peerID string, data []byte) (err error)
+	SendMessage(ctx context.Context, nsOpID string, peer, sender fftypes.JSONObject, data []byte) (err error)
 
 	// TransferBlob initiates a transfer of a previously stored blob to another node
-	TransferBlob(ctx context.Context, nsOpID, peerID string, payloadRef string) (err error)
+	TransferBlob(ctx context.Context, nsOpID string, peer, sender fftypes.JSONObject, payloadRef string) (err error)
+
+	// GetPeerID extracts the peer ID from the peer JSON
+	GetPeerID(peer fftypes.JSONObject) string
 }
 
 // Callbacks is the interface provided to the data exchange plugin, to allow it to pass events back to firefly.


Test patch:
diff --git a/internal/dataexchange/ffdx/ffdx_test.go b/internal/dataexchange/ffdx/ffdx_test.go
index 6f47c0640..46e09784c 100644
--- a/internal/dataexchange/ffdx/ffdx_test.go
+++ b/internal/dataexchange/ffdx/ffdx_test.go
@@ -152,10 +152,10 @@ func TestGetEndpointInfo(t *testing.T) {
 			"cert":     "cert data...",
 		}))
 
-	peer, err := h.GetEndpointInfo(context.Background())
+	peer, err := h.GetEndpointInfo(context.Background(), "node1")
 	assert.NoError(t, err)
-	assert.Equal(t, "peer1", peer.GetString("id"))
 	assert.Equal(t, fftypes.JSONObject{
+		"nodeID":   "peer1/node1",
 		"id":       "peer1",
 		"endpoint": "https://peer1.example.com",
 		"cert":     "cert data...",
@@ -172,7 +172,7 @@ func TestGetEndpointMissingID(t *testing.T) {
 			"cert":     "cert data...",
 		}))
 
-	_, err := h.GetEndpointInfo(context.Background())
+	_, err := h.GetEndpointInfo(context.Background(), "node1")
 	assert.Regexp(t, "FF10367", err)
 }
 
@@ -183,7 +183,7 @@ func TestGetEndpointInfoError(t *testing.T) {
 	httpmock.RegisterResponder("GET", fmt.Sprintf("%s/api/v1/id", httpURL),
 		httpmock.NewJsonResponderOrPanic(500, fftypes.JSONObject{}))
 
-	_, err := h.GetEndpointInfo(context.Background())
+	_, err := h.GetEndpointInfo(context.Background(), "node1")
 	assert.Regexp(t, "FF10229", err)
 }
 
@@ -204,7 +204,7 @@ func TestAddPeer(t *testing.T) {
 	httpmock.RegisterResponder("PUT", fmt.Sprintf("%s/api/v1/peers/peer1", httpURL),
 		httpmock.NewJsonResponderOrPanic(200, fftypes.JSONObject{}))
 
-	err := h.AddPeer(context.Background(), fftypes.JSONObject{
+	err := h.AddNode(context.Background(), "ns1", "node1", fftypes.JSONObject{
 		"id":       "peer1",
 		"endpoint": "https://peer1.example.com",
 		"cert":     "cert...",
@@ -219,7 +219,7 @@ func TestAddPeerError(t *testing.T) {
 	httpmock.RegisterResponder("PUT", fmt.Sprintf("%s/api/v1/peers/peer1", httpURL),
 		httpmock.NewJsonResponderOrPanic(500, fftypes.JSONObject{}))
 
-	err := h.AddPeer(context.Background(), fftypes.JSONObject{
+	err := h.AddNode(context.Background(), "ns1", "node1", fftypes.JSONObject{
 		"id": "peer1",
 	})
 	assert.Regexp(t, "FF10229", err)
@@ -289,112 +289,6 @@ func TestUploadBlobError(t *testing.T) {
 	assert.Regexp(t, "FF10229", err)
 }
 
-func TestCheckBlobReceivedOk(t *testing.T) {
-
-	h, _, _, httpURL, done := newTestFFDX(t, false)
-	defer done()
-
-	u := fftypes.NewUUID()
-	hash := fftypes.NewRandB32()
-	httpmock.RegisterResponder("HEAD", fmt.Sprintf("%s/api/v1/blobs/peer1/ns1/%s", httpURL, u),
-		func(r *http.Request) (*http.Response, error) {
-			res := &http.Response{
-				Header: http.Header{
-					"Dx-Hash": []string{hash.String()},
-					"Dx-Size": []string{"12345"},
-				},
-				StatusCode: 200,
-			}
-			return res, nil
-		})
-
-	hashReturned, size, err := h.CheckBlobReceived(context.Background(), "peer1", "ns1", *u)
-	assert.NoError(t, err)
-	assert.Equal(t, *hash, *hashReturned)
-	assert.Equal(t, int64(size), size)
-}
-
-func TestCheckBlobReceivedBadHash(t *testing.T) {
-
-	h, _, _, httpURL, done := newTestFFDX(t, false)
-	defer done()
-
-	u := fftypes.NewUUID()
-	httpmock.RegisterResponder("HEAD", fmt.Sprintf("%s/api/v1/blobs/peer1/ns1/%s", httpURL, u),
-		func(r *http.Request) (*http.Response, error) {
-			res := &http.Response{
-				Header: http.Header{
-					"Dx-Hash": []string{"!wrong"},
-				},
-				StatusCode: 200,
-			}
-			return res, nil
-		})
-
-	_, _, err := h.CheckBlobReceived(context.Background(), "peer1", "ns1", *u)
-	assert.Regexp(t, "FF10237", err)
-}
-
-func TestCheckBlobReceivedBadSize(t *testing.T) {
-
-	h, _, _, httpURL, done := newTestFFDX(t, false)
-	defer done()
-
-	u := fftypes.NewUUID()
-	hash := fftypes.NewRandB32()
-	httpmock.RegisterResponder("HEAD", fmt.Sprintf("%s/api/v1/blobs/peer1/ns1/%s", httpURL, u),
-		func(r *http.Request) (*http.Response, error) {
-			res := &http.Response{
-				Header: http.Header{
-					"Dx-Hash": []string{hash.String()},
-					"Dx-Size": []string{"bob"},
-				},
-				StatusCode: 200,
-			}
-			return res, nil
-		})
-
-	_, _, err := h.CheckBlobReceived(context.Background(), "peer1", "ns1", *u)
-	assert.Regexp(t, "FF10237", err)
-}
-
-func TestCheckBlobReceivedNotFound(t *testing.T) {
-
-	h, _, _, httpURL, done := newTestFFDX(t, false)
-	defer done()
-
-	u := fftypes.NewUUID()
-	httpmock.RegisterResponder("HEAD", fmt.Sprintf("%s/api/v1/blobs/peer1/ns1/%s", httpURL, u),
-		func(r *http.Request) (*http.Response, error) {
-			res := &http.Response{
-				StatusCode: 404,
-			}
-			return res, nil
-		})
-
-	hashReturned, _, err := h.CheckBlobReceived(context.Background(), "peer1", "ns1", *u)
-	assert.NoError(t, err)
-	assert.Nil(t, hashReturned)
-}
-
-func TestCheckBlobReceivedError(t *testing.T) {
-
-	h, _, _, httpURL, done := newTestFFDX(t, false)
-	defer done()
-
-	u := fftypes.NewUUID()
-	httpmock.RegisterResponder("HEAD", fmt.Sprintf("%s/api/v1/blobs/peer1/ns1/%s", httpURL, u),
-		func(r *http.Request) (*http.Response, error) {
-			res := &http.Response{
-				StatusCode: 500,
-			}
-			return res, nil
-		})
-
-	_, _, err := h.CheckBlobReceived(context.Background(), "peer1", "ns1", *u)
-	assert.Regexp(t, "FF10229", err)
-}
-
 func TestDownloadBlob(t *testing.T) {
 
 	h, _, _, httpURL, done := newTestFFDX(t, false)
@@ -430,7 +324,9 @@ func TestSendMessage(t *testing.T) {
 	httpmock.RegisterResponder("POST", fmt.Sprintf("%s/api/v1/messages", httpURL),
 		httpmock.NewJsonResponderOrPanic(200, fftypes.JSONObject{}))
 
-	err := h.SendMessage(context.Background(), "ns1:"+fftypes.NewUUID().String(), "peer1", []byte(`some data`))
+	peer := fftypes.JSONObject{"id": "peer1"}
+	sender := fftypes.JSONObject{"id": "sender1"}
+	err := h.SendMessage(context.Background(), "ns1:"+fftypes.NewUUID().String(), peer, sender, []byte(`some data`))
 	assert.NoError(t, err)
 }
 
@@ -441,7 +337,9 @@ func TestSendMessageError(t *testing.T) {
 	httpmock.RegisterResponder("POST", fmt.Sprintf("%s/api/v1/message", httpURL),
 		httpmock.NewJsonResponderOrPanic(500, fftypes.JSONObject{}))
 
-	err := h.SendMessage(context.Background(), "ns1:"+fftypes.NewUUID().String(), "peer1", []byte(`some data`))
+	peer := fftypes.JSONObject{"id": "peer1"}
+	sender := fftypes.JSONObject{"id": "sender1"}
+	err := h.SendMessage(context.Background(), "ns1:"+fftypes.NewUUID().String(), peer, sender, []byte(`some data`))
 	assert.Regexp(t, "FF10229", err)
 }
 
@@ -453,7 +351,9 @@ func TestTransferBlob(t *testing.T) {
 	httpmock.RegisterResponder("POST", fmt.Sprintf("%s/api/v1/transfers", httpURL),
 		httpmock.NewJsonResponderOrPanic(200, fftypes.JSONObject{}))
 
-	err := h.TransferBlob(context.Background(), "ns1:"+fftypes.NewUUID().String(), "peer1", "ns1/id1")
+	peer := fftypes.JSONObject{"id": "peer1"}
+	sender := fftypes.JSONObject{"id": "sender1"}
+	err := h.TransferBlob(context.Background(), "ns1:"+fftypes.NewUUID().String(), peer, sender, "ns1/id1")
 	assert.NoError(t, err)
 }
 
@@ -464,7 +364,9 @@ func TestTransferBlobError(t *testing.T) {
 	httpmock.RegisterResponder("POST", fmt.Sprintf("%s/api/v1/transfers", httpURL),
 		httpmock.NewJsonResponderOrPanic(500, fftypes.JSONObject{}))
 
-	err := h.TransferBlob(context.Background(), "ns1:"+fftypes.NewUUID().String(), "peer1", "ns1/id1")
+	peer := fftypes.JSONObject{"id": "peer1"}
+	sender := fftypes.JSONObject{"id": "sender1"}
+	err := h.TransferBlob(context.Background(), "ns1:"+fftypes.NewUUID().String(), peer, sender, "ns1/id1")
 	assert.Regexp(t, "FF10229", err)
 }
 
@@ -472,6 +374,7 @@ func TestBadEvents(t *testing.T) {
 
 	h, toServer, fromServer, _, done := newTestFFDX(t, false)
 	defer done()
+	h.AddNode(context.Background(), "ns1", "node1", fftypes.JSONObject{"id": "peer2"})
 
 	err := h.Start()
 	assert.NoError(t, err)
@@ -498,6 +401,10 @@ func TestBadEvents(t *testing.T) {
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"4"}`, string(msg))
 
+	fromServer <- `{"id":"5","type":"message-received","sender":"peer1","recipient":"peer2","message":"{\"batch\":{\"namespace\":\"ns1\"}}"}`
+	msg = <-toServer
+	assert.Equal(t, `{"action":"ack","id":"5"}`, string(msg))
+
 }
 
 func TestMessageEvents(t *testing.T) {
@@ -506,9 +413,10 @@ func TestMessageEvents(t *testing.T) {
 	defer done()
 
 	mcb := &dataexchangemocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
+	h.SetHandler("ns1", "node1", mcb)
 	ocb := &coremocks.OperationCallbacks{}
 	h.SetOperationHandler("ns1", ocb)
+	h.AddNode(context.Background(), "ns1", "node1", fftypes.JSONObject{"id": "peer1"})
 
 	err := h.Start()
 	assert.NoError(t, err)
@@ -546,9 +454,9 @@ func TestMessageEvents(t *testing.T) {
 	mcb.On("DXEvent", h, mock.MatchedBy(func(ev dataexchange.DXEvent) bool {
 		return ev.EventID() == "4" &&
 			ev.Type() == dataexchange.DXEventTypeMessageReceived &&
-			ev.MessageReceived().PeerID == "peer1"
+			ev.MessageReceived().PeerID == "peer2"
 	})).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":"peer2","recipient":"peer1","message":"{\"batch\":{\"namespace\":\"ns1\"}}"}`
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"4","manifest":"{\"manifest\":true}"}`, string(msg))
 
@@ -562,9 +470,10 @@ func TestBlobEvents(t *testing.T) {
 	defer done()
 
 	mcb := &dataexchangemocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
+	h.SetHandler("ns1", "node1", mcb)
 	ocb := &coremocks.OperationCallbacks{}
 	h.SetOperationHandler("ns1", ocb)
+	h.AddNode(context.Background(), "ns1", "node1", fftypes.JSONObject{"id": "peer1"})
 
 	err := h.Start()
 	assert.NoError(t, err)
@@ -596,7 +505,7 @@ func TestBlobEvents(t *testing.T) {
 			ev.Type() == dataexchange.DXEventTypePrivateBlobReceived &&
 			ev.PrivateBlobReceived().Hash.Equals(hash)
 	})).Run(acker()).Return(nil)
-	fromServer <- fmt.Sprintf(`{"id":"9","type":"blob-received","sender":"peer1","path":"ns1/%s","hash":"%s","size":12345}`, u.String(), hash.String())
+	fromServer <- fmt.Sprintf(`{"id":"9","type":"blob-received","sender":"peer2","recipient":"peer1","path":"ns1/%s","hash":"%s","size":12345}`, u.String(), hash.String())
 	msg = <-toServer
 	assert.Equal(t, `{"action":"ack","id":"9"}`, string(msg))
 
@@ -628,7 +537,7 @@ func TestEventsWithManifest(t *testing.T) {
 	assert.Equal(t, `{"action":"ack","id":"0"}`, string(msg))
 
 	mcb := &dataexchangemocks.Callbacks{}
-	h.SetHandler("ns1", mcb)
+	h.SetHandler("ns1", "node1", mcb)
 	ocb := &coremocks.OperationCallbacks{}
 	h.SetOperationHandler("ns1", ocb)
 
@@ -719,7 +628,6 @@ func TestWebsocketWithReinit(t *testing.T) {
 	u.Scheme = "http"
 	httpURL := u.String()
 	h := &FFDX{}
-	nodes := []fftypes.JSONObject{{}}
 
 	coreconfig.Reset()
 	h.InitConfig(utConfig)
@@ -754,7 +662,7 @@ func TestWebsocketWithReinit(t *testing.T) {
 	h.InitConfig(utConfig)
 	err := h.Init(context.Background(), utConfig)
 	assert.NoError(t, err)
-	h.SetNodes(nodes)
+	h.AddNode(context.Background(), "ns1", "node1", fftypes.JSONObject{})
 
 	err = h.Start()
 	assert.NoError(t, err)
@@ -769,15 +677,11 @@ func TestDXUninitialized(t *testing.T) {
 
 	h.initialized = false
 
-	_, err := h.GetEndpointInfo(context.Background())
-	assert.Regexp(t, "FF10342", err)
-
-	err = h.AddPeer(context.Background(), fftypes.JSONObject{})
-	assert.Regexp(t, "FF10342", err)
-
-	err = h.TransferBlob(context.Background(), "ns1:"+fftypes.NewUUID().String(), "peer1", "ns1/id1")
+	peer := fftypes.JSONObject{"id": "peer1"}
+	sender := fftypes.JSONObject{"id": "sender1"}
+	err := h.TransferBlob(context.Background(), "ns1:"+fftypes.NewUUID().String(), peer, sender, "ns1/id1")
 	assert.Regexp(t, "FF10342", err)
 
-	err = h.SendMessage(context.Background(), "ns1:"+fftypes.NewUUID().String(), "peer1", []byte(`some data`))
+	err = h.SendMessage(context.Background(), "ns1:"+fftypes.NewUUID().String(), peer, sender, []byte(`some data`))
 	assert.Regexp(t, "FF10342", err)
 }
diff --git a/internal/definitions/handler_network_node_test.go b/internal/definitions/handler_network_node_test.go
index 8eacff79b..4e2605599 100644
--- a/internal/definitions/handler_network_node_test.go
+++ b/internal/definitions/handler_network_node_test.go
@@ -128,7 +128,8 @@ func TestHandleDeprecatedNodeDefinitionOK(t *testing.T) {
 	})).Return(nil)
 
 	mdx := dh.exchange.(*dataexchangemocks.Plugin)
-	mdx.On("AddPeer", ctx, node.DX.Endpoint).Return(nil)
+	mdx.On("GetPeerID", node.DX.Endpoint).Return("member_0")
+	mdx.On("AddNode", ctx, "ns1", node.Name, node.DX.Endpoint).Return(nil)
 
 	dh.multiparty = true
 
diff --git a/internal/definitions/handler_test.go b/internal/definitions/handler_test.go
index 495dbea81..349e982d0 100644
--- a/internal/definitions/handler_test.go
+++ b/internal/definitions/handler_test.go
@@ -44,7 +44,7 @@ func newTestDefinitionHandler(t *testing.T) (*definitionHandler, *testDefinition
 	tokenNames := make(map[string]string)
 	tokenNames["remote1"] = "connector1"
 	mbi.On("VerifierType").Return(core.VerifierTypeEthAddress).Maybe()
-	dh, _ := newDefinitionHandler(context.Background(), "ns1", false, mdi, mbi, mdx, mdm, mim, mam, mcm, tokenNames)
+	dh, _ := newDefinitionHandler(context.Background(), "ns1", "ns1", false, mdi, mbi, mdx, mdm, mim, mam, mcm, tokenNames)
 	return dh, newTestDefinitionBatchState(t)
 }
 
@@ -70,7 +70,7 @@ func (bs *testDefinitionBatchState) assertNoFinalizers() {
 }
 
 func TestInitFail(t *testing.T) {
-	_, err := newDefinitionHandler(context.Background(), "", false, nil, nil, nil, nil, nil, nil, nil, nil)
+	_, err := newDefinitionHandler(context.Background(), "", "", false, nil, nil, nil, nil, nil, nil, nil, nil)
 	assert.Regexp(t, "FF10128", err)
 }
 
diff --git a/internal/definitions/sender_test.go b/internal/definitions/sender_test.go
index 5ba920a12..56befa796 100644
--- a/internal/definitions/sender_test.go
+++ b/internal/definitions/sender_test.go
@@ -48,13 +48,13 @@ func newTestDefinitionSender(t *testing.T) (*definitionSender, func()) {
 	tokenRemoteNames["connector1"] = "remote1"
 
 	ctx, cancel := context.WithCancel(context.Background())
-	ds, _, err := NewDefinitionSender(ctx, "ns1", false, mdi, mbi, mdx, mbm, mim, mdm, mam, mcm, tokenRemoteNames)
+	ds, _, err := NewDefinitionSender(ctx, "ns1", "ns1", false, mdi, mbi, mdx, mbm, mim, mdm, mam, mcm, tokenRemoteNames)
 	assert.NoError(t, err)
 	return ds.(*definitionSender), cancel
 }
 
 func TestInitSenderFail(t *testing.T) {
-	_, _, err := NewDefinitionSender(context.Background(), "", false, nil, nil, nil, nil, nil, nil, nil, nil, nil)
+	_, _, err := NewDefinitionSender(context.Background(), "", "", false, nil, nil, nil, nil, nil, nil, nil, nil, nil)
 	assert.Regexp(t, "FF10128", err)
 }
 
diff --git a/internal/networkmap/register_node_test.go b/internal/networkmap/register_node_test.go
index b75e37335..4b499210c 100644
--- a/internal/networkmap/register_node_test.go
+++ b/internal/networkmap/register_node_test.go
@@ -45,7 +45,7 @@ func TestRegisterNodeOk(t *testing.T) {
 	mim.On("ResolveIdentitySigner", nm.ctx, parentOrg).Return(signerRef, nil)
 
 	mdx := nm.exchange.(*dataexchangemocks.Plugin)
-	mdx.On("GetEndpointInfo", nm.ctx).Return(fftypes.JSONObject{
+	mdx.On("GetEndpointInfo", nm.ctx, "node1").Return(fftypes.JSONObject{
 		"id":       "peer1",
 		"endpoint": "details",
 	}, nil)
@@ -101,7 +101,7 @@ func TestRegisterNodePeerInfoFail(t *testing.T) {
 	mim.On("GetMultipartyRootOrg", nm.ctx).Return(parentOrg, nil)
 
 	mdx := nm.exchange.(*dataexchangemocks.Plugin)
-	mdx.On("GetEndpointInfo", nm.ctx).Return(fftypes.JSONObject{}, fmt.Errorf("pop"))
+	mdx.On("GetEndpointInfo", nm.ctx, "node1").Return(fftypes.JSONObject{}, fmt.Errorf("pop"))
 
 	mmp := nm.multiparty.(*multipartymocks.Manager)
 	mmp.On("LocalNode").Return(multiparty.LocalNode{Name: "node1"})
diff --git a/internal/orchestrator/orchestrator_test.go b/internal/orchestrator/orchestrator_test.go
index 23692daab..f4d6c11f7 100644
--- a/internal/orchestrator/orchestrator_test.go
+++ b/internal/orchestrator/orchestrator_test.go
@@ -204,13 +204,19 @@ func TestInitOK(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
 	or.namespace.RemoteName = "ns2"
+	or.config.Multiparty.Node.Name = "node1"
+	node := &core.Identity{
+		IdentityBase: core.IdentityBase{
+			Name: "node1",
+		},
+	}
 	or.mdi.On("SetHandler", "ns", mock.Anything).Return()
 	or.mbi.On("SetHandler", "ns", mock.Anything).Return()
 	or.mbi.On("SetOperationHandler", "ns", mock.Anything).Return()
-	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return([]*core.Identity{{}}, nil, nil)
-	or.mdx.On("SetHandler", "ns2", mock.Anything).Return()
+	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return([]*core.Identity{node}, nil, nil)
+	or.mdx.On("SetHandler", "ns2", "node1", mock.Anything).Return()
 	or.mdx.On("SetOperationHandler", "ns", mock.Anything).Return()
-	or.mdx.On("SetNodes", mock.Anything).Return()
+	or.mdx.On("AddNode", mock.Anything, "ns2", "node1", mock.Anything).Return(nil)
 	or.mps.On("SetHandler", "ns", mock.Anything).Return()
 	or.mti.On("SetHandler", "ns", mock.Anything).Return(nil)
 	or.mti.On("SetOperationHandler", "ns", mock.Anything).Return()
@@ -231,7 +237,7 @@ func TestInitOK(t *testing.T) {
 	assert.Equal(t, or.mmp, or.MultiParty())
 }
 
-func TestInitDataexchangeNodesFail(t *testing.T) {
+func TestInitDataexchangeLookupNodesFail(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
 	or.mdi.On("SetHandler", "ns", mock.Anything).Return()
@@ -244,6 +250,25 @@ func TestInitDataexchangeNodesFail(t *testing.T) {
 	assert.EqualError(t, err, "pop")
 }
 
+func TestInitDataexchangeAddNodesFail(t *testing.T) {
+	or := newTestOrchestrator()
+	defer or.cleanup(t)
+	node := &core.Identity{
+		IdentityBase: core.IdentityBase{
+			Name: "node1",
+		},
+	}
+	or.mdi.On("SetHandler", "ns", mock.Anything).Return()
+	or.mbi.On("SetHandler", "ns", mock.Anything).Return()
+	or.mbi.On("SetOperationHandler", "ns", mock.Anything).Return()
+	or.mps.On("SetHandler", "ns", mock.Anything).Return()
+	or.mdi.On("GetIdentities", mock.Anything, "ns", mock.Anything).Return([]*core.Identity{node}, nil, nil)
+	or.mdx.On("AddNode", mock.Anything, "ns", "node1", mock.Anything).Return(fmt.Errorf("pop"))
+	ctx := context.Background()
+	err := or.initHandlers(ctx)
+	assert.EqualError(t, err, "pop")
+}
+
 func TestInitMessagingComponentFail(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
diff --git a/internal/privatemessaging/operations_test.go b/internal/privatemessaging/operations_test.go
index c023fa4c3..4782aade4 100644
--- a/internal/privatemessaging/operations_test.go
+++ b/internal/privatemessaging/operations_test.go
@@ -49,6 +49,16 @@ func TestPrepareAndRunTransferBlob(t *testing.T) {
 			},
 		},
 	}
+	localNode := &core.Identity{
+		IdentityBase: core.IdentityBase{
+			ID: fftypes.NewUUID(),
+		},
+		IdentityProfile: core.IdentityProfile{
+			Profile: fftypes.JSONObject{
+				"id": "local1",
+			},
+		},
+	}
 	blob := &core.Blob{
 		Hash:       fftypes.NewRandB32(),
 		PayloadRef: "payload",
@@ -60,7 +70,8 @@ func TestPrepareAndRunTransferBlob(t *testing.T) {
 	mim := pm.identity.(*identitymanagermocks.Manager)
 	mim.On("CachedIdentityLookupByID", context.Background(), mock.Anything).Return(node, nil)
 	mdi.On("GetBlobMatchingHash", context.Background(), blob.Hash).Return(blob, nil)
-	mdx.On("TransferBlob", context.Background(), "ns1:"+op.ID.String(), "peer1", "payload").Return(nil)
+	mim.On("GetLocalNode", context.Background()).Return(localNode, nil)
+	mdx.On("TransferBlob", context.Background(), "ns1:"+op.ID.String(), node.Profile, localNode.Profile, "payload").Return(nil)
 
 	po, err := pm.PrepareOperation(context.Background(), op)
 	assert.NoError(t, err)
@@ -96,6 +107,16 @@ func TestPrepareAndRunBatchSend(t *testing.T) {
 			},
 		},
 	}
+	localNode := &core.Identity{
+		IdentityBase: core.IdentityBase{
+			ID: fftypes.NewUUID(),
+		},
+		IdentityProfile: core.IdentityProfile{
+			Profile: fftypes.JSONObject{
+				"id": "local1",
+			},
+		},
+	}
 	group := &core.Group{
 		Hash: fftypes.NewRandB32(),
 	}
@@ -114,10 +135,11 @@ func TestPrepareAndRunBatchSend(t *testing.T) {
 	mdm := pm.data.(*datamocks.Manager)
 	mdm.On("HydrateBatch", context.Background(), bp).Return(batch, nil)
 	mim := pm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", context.Background()).Return(localNode, nil)
 	mim.On("CachedIdentityLookupByID", context.Background(), node.ID).Return(node, nil)
 	mdi.On("GetGroupByHash", context.Background(), "ns1", group.Hash).Return(group, nil)
 	mdi.On("GetBatchByID", context.Background(), "ns1", batch.ID).Return(bp, nil)
-	mdx.On("SendMessage", context.Background(), "ns1:"+op.ID.String(), "peer1", mock.Anything).Return(nil)
+	mdx.On("SendMessage", context.Background(), "ns1:"+op.ID.String(), node.Profile, localNode.Profile, mock.Anything).Return(nil)
 
 	po, err := pm.PrepareOperation(context.Background(), op)
 	assert.NoError(t, err)
@@ -540,6 +562,18 @@ func TestRunOperationBatchSendInvalidData(t *testing.T) {
 			ID: fftypes.NewUUID(),
 		},
 	}
+	localNode := &core.Identity{
+		IdentityBase: core.IdentityBase{
+			ID: fftypes.NewUUID(),
+		},
+		IdentityProfile: core.IdentityProfile{
+			Profile: fftypes.JSONObject{
+				"id": "local1",
+			},
+		},
+	}
+	mim := pm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", context.Background()).Return(localNode, nil)
 	transport := &core.TransportWrapper{
 		Group: &core.Group{},
 		Batch: &core.Batch{
@@ -557,6 +591,54 @@ func TestRunOperationBatchSendInvalidData(t *testing.T) {
 	assert.Regexp(t, "FF10137", err)
 }
 
+func TestRunOperationBatchSendNodeFail(t *testing.T) {
+	pm, cancel := newTestPrivateMessaging(t)
+	defer cancel()
+
+	op := &core.Operation{}
+	node := &core.Identity{
+		IdentityBase: core.IdentityBase{
+			ID: fftypes.NewUUID(),
+		},
+	}
+	mim := pm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", context.Background()).Return(nil, fmt.Errorf("pop"))
+	transport := &core.TransportWrapper{
+		Group: &core.Group{},
+		Batch: &core.Batch{
+			Payload: core.BatchPayload{
+				Data: core.DataArray{
+					{Value: fftypes.JSONAnyPtr(`!json`)},
+				},
+			},
+		},
+	}
+
+	_, complete, err := pm.RunOperation(context.Background(), opSendBatch(op, node, transport))
+
+	assert.False(t, complete)
+	assert.EqualError(t, err, "pop")
+}
+
+func TestRunOperationBlobSendNodeFail(t *testing.T) {
+	pm, cancel := newTestPrivateMessaging(t)
+	defer cancel()
+
+	op := &core.Operation{}
+	node := &core.Identity{
+		IdentityBase: core.IdentityBase{
+			ID: fftypes.NewUUID(),
+		},
+	}
+	mim := pm.identity.(*identitymanagermocks.Manager)
+	mim.On("GetLocalNode", context.Background()).Return(nil, fmt.Errorf("pop"))
+
+	_, complete, err := pm.RunOperation(context.Background(), opSendBlob(op, node, &core.Blob{}))
+
+	assert.False(t, complete)
+	assert.EqualError(t, err, "pop")
+}
+
 func TestOperationUpdate(t *testing.T) {
 	pm, cancel := newTestPrivateMessaging(t)
 	defer cancel()