Repository: hyperledger/firefly
Pull request #918 (2022-08-02 17:55:21)

Add multi-tenancy E2E test
~~In a chain with #916~~

Includes additional fixes for separating messages/data on different namespaces.

This also includes changes to the NextPins table to add a namespace column.
While it's not possible to populate namespace on any existing entries, we make a best
effort at migration by setting the namespace every time a row is updated. This will allow
legacy environments to migrate over time, as long as they don't immediately add
multi-tenant nodes alongside existing ones.

codecov-commenter: # [Codecov](https://codecov.io/gh/hyperledger/firefly/pull/918?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) Report
> Merging [#918](https://codecov.io/gh/hyperledger/firefly/pull/918?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) (b231756) into [main](https://codecov.io/gh/hyperledger/firefly/commit/3ae7ba8f1defb7fa20df5a7bf128210747ee896b?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) (3ae7ba8) will **increase** coverage by `0.00%`.
> The diff coverage is `100.00%`.

> :exclamation: Current head b231756 differs from pull request most recent head 6a15ee6. Consider uploading reports for the commit 6a15ee6 to get more accurate results

```diff
@@           Coverage Diff           @@
##             main     #918   +/-   ##
=======================================
  Coverage   99.97%   99.97%           
=======================================
  Files         299      299           
  Lines       19475    19491   +16     
=======================================
+ Hits        19470    19486   +16     
  Misses          4        4           
  Partials        1        1           
```


| [Impacted Files](https://codecov.io/gh/hyperledger/firefly/pull/918?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) | Coverage Δ | |
|---|---|---|
| [internal/blockchain/ethereum/ethereum.go](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvYmxvY2tjaGFpbi9ldGhlcmV1bS9ldGhlcmV1bS5nbw==) | `100.00% <ø> (ø)` | |
| [internal/batch/batch\_processor.go](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvYmF0Y2gvYmF0Y2hfcHJvY2Vzc29yLmdv) | `100.00% <100.00%> (ø)` | |
| [internal/broadcast/operations.go](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvYnJvYWRjYXN0L29wZXJhdGlvbnMuZ28=) | `100.00% <100.00%> (ø)` | |
| [internal/data/data\_manager.go](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZGF0YS9kYXRhX21hbmFnZXIuZ28=) | `100.00% <100.00%> (ø)` | |
| [internal/database/sqlcommon/batch\_sql.go](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZGF0YWJhc2Uvc3FsY29tbW9uL2JhdGNoX3NxbC5nbw==) | `100.00% <100.00%> (ø)` | |
| [internal/database/sqlcommon/data\_sql.go](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZGF0YWJhc2Uvc3FsY29tbW9uL2RhdGFfc3FsLmdv) | `100.00% <100.00%> (ø)` | |
| [internal/database/sqlcommon/group\_sql.go](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZGF0YWJhc2Uvc3FsY29tbW9uL2dyb3VwX3NxbC5nbw==) | `100.00% <100.00%> (ø)` | |
| [internal/database/sqlcommon/message\_sql.go](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZGF0YWJhc2Uvc3FsY29tbW9uL21lc3NhZ2Vfc3FsLmdv) | `100.00% <100.00%> (ø)` | |
| [internal/database/sqlcommon/pin\_sql.go](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZGF0YWJhc2Uvc3FsY29tbW9uL3Bpbl9zcWwuZ28=) | `100.00% <100.00%> (ø)` | |
| [internal/database/sqlcommon/transaction\_sql.go](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZGF0YWJhc2Uvc3FsY29tbW9uL3RyYW5zYWN0aW9uX3NxbC5nbw==) | `100.00% <100.00%> (ø)` | |
| ... and [13 more](https://codecov.io/gh/hyperledger/firefly/pull/918/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) | |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger)


Patch:
diff --git a/db/migrations/postgres/000096_add_namespace_to_message_indexes.down.sql b/db/migrations/postgres/000096_add_namespace_to_message_indexes.down.sql
index 853061a7b..a5ec1a5a8 100644
--- a/db/migrations/postgres/000096_add_namespace_to_message_indexes.down.sql
+++ b/db/migrations/postgres/000096_add_namespace_to_message_indexes.down.sql
@@ -14,6 +14,15 @@ CREATE UNIQUE INDEX tokenpool_locator ON tokenpool(connector, locator);
 DROP INDEX ffi_id;
 CREATE UNIQUE INDEX ffi_id ON ffi(id);
 
+DROP INDEX datatypes_id;
+CREATE UNIQUE INDEX datatypes_id ON datatypes(id);
+
+DROP INDEX batches_id;
+CREATE UNIQUE INDEX batches_id ON batches(id);
+
+DROP INDEX groups_hash;
+CREATE UNIQUE INDEX groups_hash ON groups(hash);
+
 DROP INDEX messages_id;
 CREATE UNIQUE INDEX messages_id ON messages(id);
 
@@ -34,4 +43,8 @@ DROP TABLE messages_data_old;
 
 CREATE INDEX messages_data_message ON messages_data(message_id);
 CREATE INDEX messages_data_data ON messages_data(data_id);
+
+DROP INDEX nextpins_context;
+ALTER TABLE nextpins DROP COLUMN namespace;
+CREATE INDEX nextpins_hash ON nextpins(hash);
 COMMIT;
diff --git a/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql b/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql
index 84b5d7e8c..4a240b442 100644
--- a/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql
+++ b/db/migrations/postgres/000096_add_namespace_to_message_indexes.up.sql
@@ -14,6 +14,15 @@ CREATE UNIQUE INDEX tokenpool_locator ON tokenpool(namespace, connector, locator
 DROP INDEX ffi_id;
 CREATE UNIQUE INDEX ffi_id ON ffi(namespace, id);
 
+DROP INDEX datatypes_id;
+CREATE UNIQUE INDEX datatypes_id ON datatypes(namespace, id);
+
+DROP INDEX batches_id;
+CREATE UNIQUE INDEX batches_id ON batches(namespace, id);
+
+DROP INDEX groups_hash;
+CREATE UNIQUE INDEX groups_hash ON groups(namespace_local, hash);
+
 DROP INDEX messages_id;
 CREATE UNIQUE INDEX messages_id ON messages(namespace_local, id);
 
@@ -38,4 +47,8 @@ UPDATE messages_data SET namespace = msg.namespace
 
 CREATE INDEX messages_data_message ON messages_data(namespace, message_id);
 CREATE INDEX messages_data_data ON messages_data(namespace, data_id);
+
+ALTER TABLE nextpins ADD COLUMN namespace VARCHAR(64);
+DROP INDEX nextpins_hash;
+CREATE INDEX nextpins_context ON nextpins(namespace, context);
 COMMIT;
diff --git a/db/migrations/sqlite/000096_add_namespace_to_message_indexes.down.sql b/db/migrations/sqlite/000096_add_namespace_to_message_indexes.down.sql
index 6979f6f1a..4f4f7d7fe 100644
--- a/db/migrations/sqlite/000096_add_namespace_to_message_indexes.down.sql
+++ b/db/migrations/sqlite/000096_add_namespace_to_message_indexes.down.sql
@@ -13,6 +13,15 @@ CREATE UNIQUE INDEX tokenpool_locator ON tokenpool(connector, locator);
 DROP INDEX ffi_id;
 CREATE UNIQUE INDEX ffi_id ON ffi(id);
 
+DROP INDEX datatypes_id;
+CREATE UNIQUE INDEX datatypes_id ON datatypes(id);
+
+DROP INDEX batches_id;
+CREATE UNIQUE INDEX batches_id ON batches(id);
+
+DROP INDEX groups_hash;
+CREATE UNIQUE INDEX groups_hash ON groups(hash);
+
 DROP INDEX messages_id;
 CREATE UNIQUE INDEX messages_id ON messages(id);
 
@@ -33,3 +42,7 @@ DROP TABLE messages_data_old;
 
 CREATE INDEX messages_data_message ON messages_data(message_id);
 CREATE INDEX messages_data_data ON messages_data(data_id);
+
+DROP INDEX nextpins_context;
+ALTER TABLE nextpins DROP COLUMN namespace;
+CREATE INDEX nextpins_hash ON nextpins(hash);
diff --git a/db/migrations/sqlite/000096_add_namespace_to_message_indexes.up.sql b/db/migrations/sqlite/000096_add_namespace_to_message_indexes.up.sql
index aff5238e0..eaa28256d 100644
--- a/db/migrations/sqlite/000096_add_namespace_to_message_indexes.up.sql
+++ b/db/migrations/sqlite/000096_add_namespace_to_message_indexes.up.sql
@@ -13,6 +13,15 @@ CREATE UNIQUE INDEX tokenpool_locator ON tokenpool(namespace, connector, locator
 DROP INDEX ffi_id;
 CREATE UNIQUE INDEX ffi_id ON ffi(namespace, id);
 
+DROP INDEX datatypes_id;
+CREATE UNIQUE INDEX datatypes_id ON datatypes(namespace, id);
+
+DROP INDEX batches_id;
+CREATE UNIQUE INDEX batches_id ON batches(namespace, id);
+
+DROP INDEX groups_hash;
+CREATE UNIQUE INDEX groups_hash ON groups(namespace_local, hash);
+
 DROP INDEX messages_id;
 CREATE UNIQUE INDEX messages_id ON messages(namespace_local, id);
 
@@ -37,3 +46,7 @@ UPDATE messages_data SET namespace = msg.namespace
 
 CREATE INDEX messages_data_message ON messages_data(namespace, message_id);
 CREATE INDEX messages_data_data ON messages_data(namespace, data_id);
+
+ALTER TABLE nextpins ADD COLUMN namespace VARCHAR(64);
+DROP INDEX nextpins_hash;
+CREATE INDEX nextpins_context ON nextpins(namespace, context);
diff --git a/internal/batch/batch_processor.go b/internal/batch/batch_processor.go
index 0653815bf..500f419a4 100644
--- a/internal/batch/batch_processor.go
+++ b/internal/batch/batch_processor.go
@@ -531,7 +531,7 @@ func (bp *batchProcessor) flushNonceState(ctx context.Context, state *DispatchSt
 	// message cache, until after the Retry/RunAsGroup has ended in sealBatch.
 	for _, msg := range state.Messages {
 		if pins, ok := state.msgPins[*msg.Header.ID]; ok {
-			if err := bp.database.UpdateMessage(ctx, msg.Header.ID,
+			if err := bp.database.UpdateMessage(ctx, bp.bm.namespace, msg.Header.ID,
 				database.MessageQueryFactory.NewUpdate(ctx).Set("pins", pins),
 			); err != nil {
 				return err
@@ -643,7 +643,7 @@ func (bp *batchProcessor) markPayloadDispatched(state *DispatchState) error {
 					Set("confirmed", confirmTime)
 			}
 
-			if err = bp.database.UpdateMessages(ctx, filter, allMsgsUpdate); err != nil {
+			if err = bp.database.UpdateMessages(ctx, bp.bm.namespace, filter, allMsgsUpdate); err != nil {
 				return err
 			}
 
diff --git a/internal/broadcast/operations.go b/internal/broadcast/operations.go
index f79cdb6f0..e30cddf25 100644
--- a/internal/broadcast/operations.go
+++ b/internal/broadcast/operations.go
@@ -159,7 +159,7 @@ func (bm *broadcastManager) uploadBlob(ctx context.Context, data uploadBlobData)
 	}
 
 	// Update the data in the DB
-	err = bm.database.UpdateData(ctx, data.Data.ID, database.DataQueryFactory.NewUpdate(ctx).Set("blob.public", data.Data.Blob.Public))
+	err = bm.database.UpdateData(ctx, bm.namespace.LocalName, data.Data.ID, database.DataQueryFactory.NewUpdate(ctx).Set("blob.public", data.Data.Blob.Public))
 	if err != nil {
 		return nil, false, err
 	}
diff --git a/internal/coremsgs/en_error_messages.go b/internal/coremsgs/en_error_messages.go
index 0f5af182e..02bfa6842 100644
--- a/internal/coremsgs/en_error_messages.go
+++ b/internal/coremsgs/en_error_messages.go
@@ -145,7 +145,7 @@ var (
 	MsgIdentityNotFoundByString           = ffe("FF10277", "Identity could not be resolved via lookup string '%s'")
 	MsgAuthorOrgSigningKeyMismatch        = ffe("FF10279", "Author organization '%s' is not associated with signing key '%s'")
 	MsgCannotTransferToSelf               = ffe("FF10280", "From and to addresses must be different", 400)
-	MsgLocalOrgLookupFailed               = ffe("FF10281", "Unable resolve the local org '%s' by the configured signing key on the node. Please confirm the org is registered with key '%s'", 500)
+	MsgLocalOrgLookupFailed               = ffe("FF10281", "Unable to resolve the local org '%s' by the configured signing key on the node. Please confirm the org is registered with key '%s'", 500)
 	MsgFabconnectRESTErr                  = ffe("FF10284", "Error from fabconnect: %s")
 	MsgInvalidIdentity                    = ffe("FF10285", "Supplied Fabric signer identity is invalid", 400)
 	MsgFailedToDecodeCertificate          = ffe("FF10286", "Failed to decode certificate: %s", 500)
diff --git a/internal/database/sqlcommon/batch_sql.go b/internal/database/sqlcommon/batch_sql.go
index c85f635fb..e5b8f31f9 100644
--- a/internal/database/sqlcommon/batch_sql.go
+++ b/internal/database/sqlcommon/batch_sql.go
@@ -67,7 +67,7 @@ func (s *SQLCommon) UpsertBatch(ctx context.Context, batch *core.BatchPersisted)
 	batchRows, _, err := s.queryTx(ctx, batchesTable, tx,
 		sq.Select("hash").
 			From(batchesTable).
-			Where(sq.Eq{"id": batch.ID}),
+			Where(sq.Eq{"id": batch.ID, "namespace": batch.Namespace}),
 	)
 	if err != nil {
 		return err
@@ -91,7 +91,6 @@ func (s *SQLCommon) UpsertBatch(ctx context.Context, batch *core.BatchPersisted)
 		if _, err = s.updateTx(ctx, batchesTable, tx,
 			sq.Update(batchesTable).
 				Set("btype", string(batch.Type)).
-				Set("namespace", batch.Namespace).
 				Set("author", batch.Author).
 				Set("key", batch.Key).
 				Set("group_hash", batch.Group).
@@ -102,7 +101,7 @@ func (s *SQLCommon) UpsertBatch(ctx context.Context, batch *core.BatchPersisted)
 				Set("tx_type", batch.TX.Type).
 				Set("tx_id", batch.TX.ID).
 				Set("node_id", batch.Node).
-				Where(sq.Eq{"id": batch.ID}),
+				Where(sq.Eq{"id": batch.ID, "namespace": batch.Namespace}),
 			func() {
 				s.callbacks.UUIDCollectionNSEvent(database.CollectionBatches, core.ChangeEventTypeUpdated, batch.Namespace, batch.ID)
 			},
@@ -214,7 +213,7 @@ func (s *SQLCommon) GetBatches(ctx context.Context, namespace string, filter dat
 
 }
 
-func (s *SQLCommon) UpdateBatch(ctx context.Context, id *fftypes.UUID, update database.Update) (err error) {
+func (s *SQLCommon) UpdateBatch(ctx context.Context, namespace string, id *fftypes.UUID, update database.Update) (err error) {
 
 	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
 	if err != nil {
@@ -226,7 +225,7 @@ func (s *SQLCommon) UpdateBatch(ctx context.Context, id *fftypes.UUID, update da
 	if err != nil {
 		return err
 	}
-	query = query.Where(sq.Eq{"id": id})
+	query = query.Where(sq.Eq{"id": id, "namespace": namespace})
 
 	_, err = s.updateTx(ctx, batchesTable, tx, query, nil /* no change events on filter update */)
 	if err != nil {
diff --git a/internal/database/sqlcommon/contractapis_sql.go b/internal/database/sqlcommon/contractapis_sql.go
index 3f7609f9a..d7bb989b7 100644
--- a/internal/database/sqlcommon/contractapis_sql.go
+++ b/internal/database/sqlcommon/contractapis_sql.go
@@ -56,7 +56,10 @@ func (s *SQLCommon) UpsertContractAPI(ctx context.Context, api *core.ContractAPI
 	rows, _, err := s.queryTx(ctx, contractapisTable, tx,
 		sq.Select("id").
 			From(contractapisTable).
-			Where(sq.And{sq.Eq{"namespace": api.Namespace}, sq.Eq{"name": api.Name}}),
+			Where(sq.Eq{
+				"namespace": api.Namespace,
+				"name":      api.Name,
+			}),
 	)
 	if err != nil {
 		return err
@@ -81,7 +84,6 @@ func (s *SQLCommon) UpsertContractAPI(ctx context.Context, api *core.ContractAPI
 				Set("interface_id", api.Interface.ID).
 				Set("location", api.Location).
 				Set("name", api.Name).
-				Set("namespace", api.Namespace).
 				Set("message_id", api.Message),
 			func() {
 				s.callbacks.UUIDCollectionNSEvent(database.CollectionContractAPIs, core.ChangeEventTypeUpdated, api.Namespace, api.ID)
diff --git a/internal/database/sqlcommon/data_sql.go b/internal/database/sqlcommon/data_sql.go
index 82daae384..4f23bb5be 100644
--- a/internal/database/sqlcommon/data_sql.go
+++ b/internal/database/sqlcommon/data_sql.go
@@ -70,7 +70,6 @@ func (s *SQLCommon) attemptDataUpdate(ctx context.Context, tx *txWrapper, data *
 	return s.updateTx(ctx, dataTable, tx,
 		sq.Update(dataTable).
 			Set("validator", string(data.Validator)).
-			Set("namespace", data.Namespace).
 			Set("datatype_name", datatype.Name).
 			Set("datatype_version", datatype.Version).
 			Set("hash", data.Hash).
@@ -82,8 +81,9 @@ func (s *SQLCommon) attemptDataUpdate(ctx context.Context, tx *txWrapper, data *
 			Set("value_size", data.ValueSize).
 			Set("value", data.Value).
 			Where(sq.Eq{
-				"id":   data.ID,
-				"hash": data.Hash,
+				"id":        data.ID,
+				"hash":      data.Hash,
+				"namespace": data.Namespace,
 			}),
 		func() {
 			s.callbacks.UUIDCollectionNSEvent(database.CollectionData, core.ChangeEventTypeUpdated, data.Namespace, data.ID)
@@ -150,7 +150,7 @@ func (s *SQLCommon) UpsertData(ctx context.Context, data *core.Data, optimizatio
 		dataRows, _, err := s.queryTx(ctx, dataTable, tx,
 			sq.Select("hash").
 				From(dataTable).
-				Where(sq.Eq{"id": data.ID}),
+				Where(sq.Eq{"id": data.ID, "namespace": data.Namespace}),
 		)
 		if err != nil {
 			return err
@@ -344,7 +344,7 @@ func (s *SQLCommon) GetDataRefs(ctx context.Context, namespace string, filter da
 
 }
 
-func (s *SQLCommon) UpdateData(ctx context.Context, id *fftypes.UUID, update database.Update) (err error) {
+func (s *SQLCommon) UpdateData(ctx context.Context, namespace string, id *fftypes.UUID, update database.Update) (err error) {
 
 	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
 	if err != nil {
@@ -356,7 +356,7 @@ func (s *SQLCommon) UpdateData(ctx context.Context, id *fftypes.UUID, update dat
 	if err != nil {
 		return err
 	}
-	query = query.Where(sq.Eq{"id": id})
+	query = query.Where(sq.Eq{"id": id, "namespace": namespace})
 
 	_, err = s.updateTx(ctx, dataTable, tx, query, nil /* no change events for filter based updates */)
 	if err != nil {
diff --git a/internal/database/sqlcommon/datatype_sql.go b/internal/database/sqlcommon/datatype_sql.go
index a45696c0b..48ffefe29 100644
--- a/internal/database/sqlcommon/datatype_sql.go
+++ b/internal/database/sqlcommon/datatype_sql.go
@@ -62,7 +62,10 @@ func (s *SQLCommon) UpsertDatatype(ctx context.Context, datatype *core.Datatype,
 		datatypeRows, _, err := s.queryTx(ctx, datatypesTable, tx,
 			sq.Select("id").
 				From(datatypesTable).
-				Where(sq.Eq{"id": datatype.ID}),
+				Where(sq.Eq{
+					"namespace": datatype.Namespace,
+					"id":        datatype.ID,
+				}),
 		)
 		if err != nil {
 			return err
@@ -78,7 +81,6 @@ func (s *SQLCommon) UpsertDatatype(ctx context.Context, datatype *core.Datatype,
 			sq.Update(datatypesTable).
 				Set("message_id", datatype.Message).
 				Set("validator", string(datatype.Validator)).
-				Set("namespace", datatype.Namespace).
 				Set("name", datatype.Name).
 				Set("version", datatype.Version).
 				Set("hash", datatype.Hash).
diff --git a/internal/database/sqlcommon/ffi_sql.go b/internal/database/sqlcommon/ffi_sql.go
index d348154f3..43bb91a4e 100644
--- a/internal/database/sqlcommon/ffi_sql.go
+++ b/internal/database/sqlcommon/ffi_sql.go
@@ -55,7 +55,10 @@ func (s *SQLCommon) UpsertFFI(ctx context.Context, ffi *fftypes.FFI) (err error)
 	rows, _, err := s.queryTx(ctx, ffiTable, tx,
 		sq.Select("id").
 			From(ffiTable).
-			Where(sq.And{sq.Eq{"id": ffi.ID}}),
+			Where(sq.Eq{
+				"namespace": ffi.Namespace,
+				"id":        ffi.ID,
+			}),
 	)
 	if err != nil {
 		return err
@@ -66,7 +69,6 @@ func (s *SQLCommon) UpsertFFI(ctx context.Context, ffi *fftypes.FFI) (err error)
 	if existing {
 		if _, err = s.updateTx(ctx, ffiTable, tx,
 			sq.Update(ffiTable).
-				Set("namespace", ffi.Namespace).
 				Set("name", ffi.Name).
 				Set("version", ffi.Version).
 				Set("description", ffi.Description).
diff --git a/internal/database/sqlcommon/group_sql.go b/internal/database/sqlcommon/group_sql.go
index 24fc59605..2917648e8 100644
--- a/internal/database/sqlcommon/group_sql.go
+++ b/internal/database/sqlcommon/group_sql.go
@@ -69,7 +69,7 @@ func (s *SQLCommon) UpsertGroup(ctx context.Context, group *core.Group, optimiza
 		groupRows, _, err := s.queryTx(ctx, groupsTable, tx,
 			sq.Select("hash").
 				From(groupsTable).
-				Where(sq.Eq{"hash": group.Hash}),
+				Where(sq.Eq{"hash": group.Hash, "namespace_local": group.LocalNamespace}),
 		)
 		if err != nil {
 			return err
@@ -108,7 +108,7 @@ func (s *SQLCommon) attemptGroupUpdate(ctx context.Context, tx *txWrapper, group
 			Set("name", group.Name).
 			Set("hash", group.Hash).
 			Set("created", group.Created).
-			Where(sq.Eq{"hash": group.Hash}),
+			Where(sq.Eq{"hash": group.Hash, "namespace_local": group.LocalNamespace}),
 		func() {
 			s.callbacks.HashCollectionNSEvent(database.CollectionGroups, core.ChangeEventTypeUpdated, group.LocalNamespace, group.Hash)
 		},
diff --git a/internal/database/sqlcommon/identity_sql.go b/internal/database/sqlcommon/identity_sql.go
index e2311a91c..84b8aff65 100644
--- a/internal/database/sqlcommon/identity_sql.go
+++ b/internal/database/sqlcommon/identity_sql.go
@@ -63,7 +63,6 @@ func (s *SQLCommon) attemptIdentityUpdate(ctx context.Context, tx *txWrapper, id
 			Set("did", identity.DID).
 			Set("parent", identity.Parent).
 			Set("itype", identity.Type).
-			Set("namespace", identity.Namespace).
 			Set("name", identity.Name).
 			Set("description", identity.Description).
 			Set("profile", identity.Profile).
diff --git a/internal/database/sqlcommon/message_sql.go b/internal/database/sqlcommon/message_sql.go
index 5df652e46..5fa7a4021 100644
--- a/internal/database/sqlcommon/message_sql.go
+++ b/internal/database/sqlcommon/message_sql.go
@@ -70,8 +70,6 @@ func (s *SQLCommon) attemptMessageUpdate(ctx context.Context, tx *txWrapper, mes
 			Set("author", message.Header.Author).
 			Set("key", message.Header.Key).
 			Set("created", message.Header.Created).
-			Set("namespace", message.Header.Namespace).
-			Set("namespace_local", message.LocalNamespace).
 			Set("topics", message.Header.Topics).
 			Set("tag", message.Header.Tag).
 			Set("group_hash", message.Header.Group).
@@ -83,8 +81,10 @@ func (s *SQLCommon) attemptMessageUpdate(ctx context.Context, tx *txWrapper, mes
 			Set("tx_type", message.Header.TxType).
 			Set("batch_id", message.BatchID).
 			Where(sq.Eq{
-				"id":   message.Header.ID,
-				"hash": message.Hash,
+				"id":              message.Header.ID,
+				"hash":            message.Hash,
+				"namespace_local": message.LocalNamespace,
+				"namespace":       message.Header.Namespace,
 			}),
 		func() {
 			s.callbacks.OrderedUUIDCollectionNSEvent(database.CollectionMessages, core.ChangeEventTypeUpdated, message.LocalNamespace, message.Header.ID, -1 /* not applicable on update */)
@@ -150,7 +150,7 @@ func (s *SQLCommon) UpsertMessage(ctx context.Context, message *core.Message, op
 		msgRows, _, err := s.queryTx(ctx, messagesTable, tx,
 			sq.Select("hash", sequenceColumn).
 				From(messagesTable).
-				Where(sq.Eq{"id": message.Header.ID}),
+				Where(sq.Eq{"id": message.Header.ID, "namespace_local": message.LocalNamespace}),
 		)
 		if err != nil {
 			return err
@@ -275,7 +275,7 @@ func (s *SQLCommon) ReplaceMessage(ctx context.Context, message *core.Message) (
 	if err := s.deleteTx(ctx, messagesTable, tx,
 		sq.Delete(messagesTable).
 			Where(sq.And{
-				sq.Eq{"id": message.Header.ID},
+				sq.Eq{"id": message.Header.ID, "namespace_local": message.LocalNamespace},
 			}),
 		nil, // no change event
 	); err != nil {
@@ -299,7 +299,7 @@ func (s *SQLCommon) updateMessageDataRefs(ctx context.Context, tx *txWrapper, me
 		if err := s.deleteTx(ctx, messagesDataJoinTable, tx,
 			sq.Delete(messagesDataJoinTable).
 				Where(sq.And{
-					sq.Eq{"message_id": message.Header.ID},
+					sq.Eq{"message_id": message.Header.ID, "namespace": message.LocalNamespace},
 				}),
 			nil, // no change event
 		); err != nil && err != database.DeleteRecordNotFound {
@@ -318,12 +318,14 @@ func (s *SQLCommon) updateMessageDataRefs(ctx context.Context, tx *txWrapper, me
 		if _, err := s.insertTx(ctx, messagesDataJoinTable, tx,
 			sq.Insert(messagesDataJoinTable).
 				Columns(
+					"namespace",
 					"message_id",
 					"data_id",
 					"data_hash",
 					"data_idx",
 				).
 				Values(
+					message.LocalNamespace,
 					message.Header.ID,
 					msgDataRef.ID,
 					msgDataRef.Hash,
@@ -344,7 +346,7 @@ func (s *SQLCommon) updateMessageDataRefs(ctx context.Context, tx *txWrapper, me
 // way for a single-query option. So a two-query option ended up being simplest.
 // See commit e304161a30b8044a42b5bac3fcfca7e7bd8f8ab7 for the abandoned changeset
 // that implemented LEFT JOIN
-func (s *SQLCommon) loadDataRefs(ctx context.Context, msgs []*core.Message) error {
+func (s *SQLCommon) loadDataRefs(ctx context.Context, namespace string, msgs []*core.Message) error {
 
 	msgIDs := make([]string, len(msgs))
 	for i, m := range msgs {
@@ -361,7 +363,7 @@ func (s *SQLCommon) loadDataRefs(ctx context.Context, msgs []*core.Message) erro
 			"data_idx",
 		).
 			From(messagesDataJoinTable).
-			Where(sq.Eq{"message_id": msgIDs}).
+			Where(sq.Eq{"message_id": msgIDs, "namespace": namespace}).
 			OrderBy("data_idx"),
 	)
 	if err != nil {
@@ -451,14 +453,14 @@ func (s *SQLCommon) GetMessageByID(ctx context.Context, namespace string, id *ff
 	}
 
 	rows.Close()
-	if err = s.loadDataRefs(ctx, []*core.Message{msg}); err != nil {
+	if err = s.loadDataRefs(ctx, namespace, []*core.Message{msg}); err != nil {
 		return nil, err
 	}
 
 	return msg, nil
 }
 
-func (s *SQLCommon) getMessagesQuery(ctx context.Context, query sq.SelectBuilder, fop sq.Sqlizer, fi *database.FilterInfo, allowCount bool) (message []*core.Message, fr *database.FilterResult, err error) {
+func (s *SQLCommon) getMessagesQuery(ctx context.Context, namespace string, query sq.SelectBuilder, fop sq.Sqlizer, fi *database.FilterInfo, allowCount bool) (message []*core.Message, fr *database.FilterResult, err error) {
 	if fi.Count && !allowCount {
 		return nil, nil, i18n.NewError(ctx, coremsgs.MsgFilterCountNotSupported)
 	}
@@ -480,7 +482,7 @@ func (s *SQLCommon) getMessagesQuery(ctx context.Context, query sq.SelectBuilder
 
 	rows.Close()
 	if len(msgs) > 0 {
-		if err = s.loadDataRefs(ctx, msgs); err != nil {
+		if err = s.loadDataRefs(ctx, namespace, msgs); err != nil {
 			return nil, nil, err
 		}
 	}
@@ -557,7 +559,7 @@ func (s *SQLCommon) GetMessages(ctx context.Context, namespace string, filter da
 	if err != nil {
 		return nil, nil, err
 	}
-	return s.getMessagesQuery(ctx, query, fop, fi, true)
+	return s.getMessagesQuery(ctx, namespace, query, fop, fi, true)
 }
 
 func (s *SQLCommon) GetMessagesForData(ctx context.Context, namespace string, dataID *fftypes.UUID, filter database.Filter) (message []*core.Message, fr *database.FilterResult, err error) {
@@ -575,14 +577,14 @@ func (s *SQLCommon) GetMessagesForData(ctx context.Context, namespace string, da
 	}
 
 	query = query.LeftJoin("messages AS m ON m.id = md.message_id")
-	return s.getMessagesQuery(ctx, query, fop, fi, false)
+	return s.getMessagesQuery(ctx, namespace, query, fop, fi, false)
 }
 
-func (s *SQLCommon) UpdateMessage(ctx context.Context, msgid *fftypes.UUID, update database.Update) (err error) {
-	return s.UpdateMessages(ctx, database.MessageQueryFactory.NewFilter(ctx).Eq("id", msgid), update)
+func (s *SQLCommon) UpdateMessage(ctx context.Context, namespace string, msgid *fftypes.UUID, update database.Update) (err error) {
+	return s.UpdateMessages(ctx, namespace, database.MessageQueryFactory.NewFilter(ctx).Eq("id", msgid), update)
 }
 
-func (s *SQLCommon) UpdateMessages(ctx context.Context, filter database.Filter, update database.Update) (err error) {
+func (s *SQLCommon) UpdateMessages(ctx context.Context, namespace string, filter database.Filter, update database.Update) (err error) {
 
 	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
 	if err != nil {
@@ -590,7 +592,7 @@ func (s *SQLCommon) UpdateMessages(ctx context.Context, filter database.Filter,
 	}
 	defer s.rollbackTx(ctx, tx, autoCommit)
 
-	query, err := s.buildUpdate(sq.Update(messagesTable), update, msgFilterFieldMap)
+	query, err := s.buildUpdate(sq.Update(messagesTable).Where(sq.Eq{"namespace_local": namespace}), update, msgFilterFieldMap)
 	if err != nil {
 		return err
 	}
diff --git a/internal/database/sqlcommon/nextpin_sql.go b/internal/database/sqlcommon/nextpin_sql.go
index dbd732596..fa5a47ebb 100644
--- a/internal/database/sqlcommon/nextpin_sql.go
+++ b/internal/database/sqlcommon/nextpin_sql.go
@@ -19,12 +19,10 @@ package sqlcommon
 import (
 	"context"
 	"database/sql"
-	"fmt"
 
 	sq "github.com/Masterminds/squirrel"
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/i18n"
-	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/pkg/core"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -32,12 +30,12 @@ import (
 
 var (
 	nextpinColumns = []string{
+		"namespace",
 		"context",
 		"identity",
 		"hash",
 		"nonce",
 	}
-	nextpinFilterFieldMap = map[string]string{}
 )
 
 const nextpinsTable = "nextpins"
@@ -53,6 +51,7 @@ func (s *SQLCommon) InsertNextPin(ctx context.Context, nextpin *core.NextPin) (e
 		sq.Insert(nextpinsTable).
 			Columns(nextpinColumns...).
 			Values(
+				nextpin.Namespace,
 				nextpin.Context,
 				nextpin.Identity,
 				nextpin.Hash,
@@ -71,6 +70,7 @@ func (s *SQLCommon) InsertNextPin(ctx context.Context, nextpin *core.NextPin) (e
 func (s *SQLCommon) nextpinResult(ctx context.Context, row *sql.Rows) (*core.NextPin, error) {
 	nextpin := core.NextPin{}
 	err := row.Scan(
+		&nextpin.Namespace,
 		&nextpin.Context,
 		&nextpin.Identity,
 		&nextpin.Hash,
@@ -83,58 +83,18 @@ func (s *SQLCommon) nextpinResult(ctx context.Context, row *sql.Rows) (*core.Nex
 	return &nextpin, nil
 }
 
-func (s *SQLCommon) getNextPinPred(ctx context.Context, desc string, pred interface{}) (message *core.NextPin, err error) {
-	cols := append([]string{}, nextpinColumns...)
-	cols = append(cols, sequenceColumn)
-	rows, _, err := s.query(ctx, nextpinsTable,
-		sq.Select(cols...).
-			From(nextpinsTable).
-			Where(pred).
-			Limit(1),
-	)
-	if err != nil {
-		return nil, err
-	}
-	defer rows.Close()
-
-	if !rows.Next() {
-		log.L(ctx).Debugf("NextPin '%s' not found", desc)
-		return nil, nil
-	}
-
-	nextpin, err := s.nextpinResult(ctx, rows)
-	if err != nil {
-		return nil, err
-	}
-
-	return nextpin, nil
-}
-
-func (s *SQLCommon) GetNextPinByContextAndIdentity(ctx context.Context, context *fftypes.Bytes32, identity string) (message *core.NextPin, err error) {
-	return s.getNextPinPred(ctx, fmt.Sprintf("%s:%s", context, identity), sq.Eq{
-		"context":  context,
-		"identity": identity,
-	})
-}
-
-func (s *SQLCommon) GetNextPinByHash(ctx context.Context, hash *fftypes.Bytes32) (message *core.NextPin, err error) {
-	return s.getNextPinPred(ctx, hash.String(), sq.Eq{
-		"hash": hash,
-	})
-}
-
-func (s *SQLCommon) GetNextPins(ctx context.Context, filter database.Filter) (message []*core.NextPin, fr *database.FilterResult, err error) {
+func (s *SQLCommon) GetNextPinsForContext(ctx context.Context, namespace string, context *fftypes.Bytes32) (message []*core.NextPin, err error) {
 
 	cols := append([]string{}, nextpinColumns...)
 	cols = append(cols, sequenceColumn)
-	query, fop, fi, err := s.filterSelect(ctx, "", sq.Select(cols...).From(nextpinsTable), filter, nextpinFilterFieldMap, []interface{}{"sequence"})
-	if err != nil {
-		return nil, nil, err
-	}
 
-	rows, tx, err := s.query(ctx, nextpinsTable, query)
+	rows, _, err := s.query(ctx, nextpinsTable, sq.Select(cols...).From(nextpinsTable).
+		Where(sq.And{
+			sq.Eq{"context": context},
+			sq.Or{sq.Eq{"namespace": namespace}, sq.Eq{"namespace": nil}}, // old entries will have NULL for namespace
+		}))
 	if err != nil {
-		return nil, nil, err
+		return nil, err
 	}
 	defer rows.Close()
 
@@ -142,16 +102,16 @@ func (s *SQLCommon) GetNextPins(ctx context.Context, filter database.Filter) (me
 	for rows.Next() {
 		d, err := s.nextpinResult(ctx, rows)
 		if err != nil {
-			return nil, nil, err
+			return nil, err
 		}
 		nextpin = append(nextpin, d)
 	}
 
-	return nextpin, s.queryRes(ctx, nextpinsTable, tx, fop, fi), err
+	return nextpin, err
 
 }
 
-func (s *SQLCommon) UpdateNextPin(ctx context.Context, sequence int64, update database.Update) (err error) {
+func (s *SQLCommon) UpdateNextPin(ctx context.Context, namespace string, sequence int64, update database.Update) (err error) {
 
 	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
 	if err != nil {
@@ -163,6 +123,7 @@ func (s *SQLCommon) UpdateNextPin(ctx context.Context, sequence int64, update da
 	if err != nil {
 		return err
 	}
+	query = query.Set("namespace", namespace) // always populate namespace (to migrate the table over time)
 	query = query.Where(sq.Eq{sequenceColumn: sequence})
 
 	_, err = s.updateTx(ctx, nextpinsTable, tx, query, nil /* no change events for next pins */)
@@ -172,21 +133,3 @@ func (s *SQLCommon) UpdateNextPin(ctx context.Context, sequence int64, update da
 
 	return s.commitTx(ctx, tx, autoCommit)
 }
-
-func (s *SQLCommon) DeleteNextPin(ctx context.Context, sequence int64) (err error) {
-
-	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
-	if err != nil {
-		return err
-	}
-	defer s.rollbackTx(ctx, tx, autoCommit)
-
-	err = s.deleteTx(ctx, nextpinsTable, tx, sq.Delete(nextpinsTable).Where(sq.Eq{
-		sequenceColumn: sequence,
-	}), nil /* no change events for next pins */)
-	if err != nil {
-		return err
-	}
-
-	return s.commitTx(ctx, tx, autoCommit)
-}
diff --git a/internal/database/sqlcommon/pin_sql.go b/internal/database/sqlcommon/pin_sql.go
index 0bfda45d4..a900a759c 100644
--- a/internal/database/sqlcommon/pin_sql.go
+++ b/internal/database/sqlcommon/pin_sql.go
@@ -61,9 +61,10 @@ func (s *SQLCommon) UpsertPin(ctx context.Context, pin *core.Pin) (err error) {
 		sq.Select(sequenceColumn, "masked", "dispatched").
 			From(pinsTable).
 			Where(sq.Eq{
-				"hash":     pin.Hash,
-				"batch_id": pin.Batch,
-				"idx":      pin.Index,
+				"hash":      pin.Hash,
+				"batch_id":  pin.Batch,
+				"idx":       pin.Index,
+				"namespace": pin.Namespace,
 			}))
 	if err != nil {
 		return err
@@ -198,7 +199,7 @@ func (s *SQLCommon) GetPins(ctx context.Context, namespace string, filter databa
 
 }
 
-func (s *SQLCommon) UpdatePins(ctx context.Context, filter database.Filter, update database.Update) (err error) {
+func (s *SQLCommon) UpdatePins(ctx context.Context, namespace string, filter database.Filter, update database.Update) (err error) {
 
 	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
 	if err != nil {
@@ -206,7 +207,7 @@ func (s *SQLCommon) UpdatePins(ctx context.Context, filter database.Filter, upda
 	}
 	defer s.rollbackTx(ctx, tx, autoCommit)
 
-	query, err := s.buildUpdate(sq.Update(pinsTable), update, pinFilterFieldMap)
+	query, err := s.buildUpdate(sq.Update(pinsTable).Where(sq.Eq{"namespace": namespace}), update, pinFilterFieldMap)
 	if err != nil {
 		return err
 	}
diff --git a/internal/database/sqlcommon/subscription_sql.go b/internal/database/sqlcommon/subscription_sql.go
index 3c2c12e58..cd21de9ba 100644
--- a/internal/database/sqlcommon/subscription_sql.go
+++ b/internal/database/sqlcommon/subscription_sql.go
@@ -88,7 +88,6 @@ func (s *SQLCommon) UpsertSubscription(ctx context.Context, subscription *core.S
 		if _, err = s.updateTx(ctx, subscriptionsTable, tx,
 			sq.Update(subscriptionsTable).
 				// Note we do not update ID
-				Set("namespace", subscription.Namespace).
 				Set("name", subscription.Name).
 				Set("transport", subscription.Transport).
 				Set("filters", subscription.Filter).
diff --git a/internal/database/sqlcommon/tokenapproval_sql.go b/internal/database/sqlcommon/tokenapproval_sql.go
index 33dd09c9c..67ac09365 100644
--- a/internal/database/sqlcommon/tokenapproval_sql.go
+++ b/internal/database/sqlcommon/tokenapproval_sql.go
@@ -73,7 +73,10 @@ func (s *SQLCommon) UpsertTokenApproval(ctx context.Context, approval *core.Toke
 	rows, _, err := s.queryTx(ctx, tokenapprovalTable, tx,
 		sq.Select("seq").
 			From(tokenapprovalTable).
-			Where(sq.Eq{"protocol_id": approval.ProtocolID}),
+			Where(sq.Eq{
+				"namespace":   approval.Namespace,
+				"protocol_id": approval.ProtocolID,
+			}),
 	)
 	if err != nil {
 		return err
@@ -91,7 +94,6 @@ func (s *SQLCommon) UpsertTokenApproval(ctx context.Context, approval *core.Toke
 				Set("operator_key", approval.Operator).
 				Set("pool_id", approval.Pool).
 				Set("connector", approval.Connector).
-				Set("namespace", approval.Namespace).
 				Set("approved", approval.Approved).
 				Set("info", approval.Info).
 				Set("tx_type", approval.TX.Type).
diff --git a/internal/database/sqlcommon/tokenbalance_sql.go b/internal/database/sqlcommon/tokenbalance_sql.go
index e12155e6a..b35572f7e 100644
--- a/internal/database/sqlcommon/tokenbalance_sql.go
+++ b/internal/database/sqlcommon/tokenbalance_sql.go
@@ -49,33 +49,34 @@ var (
 )
 
 func (s *SQLCommon) addTokenBalance(ctx context.Context, tx *txWrapper, transfer *core.TokenTransfer, key string, negate bool) error {
-	account, err := s.GetTokenBalance(ctx, transfer.Namespace, transfer.Pool, transfer.TokenIndex, key)
+	balance, err := s.GetTokenBalance(ctx, transfer.Namespace, transfer.Pool, transfer.TokenIndex, key)
 	if err != nil {
 		return err
 	}
 
-	var balance *fftypes.FFBigInt
-	if account != nil {
-		balance = &account.Balance
+	var total *fftypes.FFBigInt
+	if balance != nil {
+		total = &balance.Balance
 	} else {
-		balance = &fftypes.FFBigInt{}
+		total = &fftypes.FFBigInt{}
 	}
 	if negate {
-		balance.Int().Sub(balance.Int(), transfer.Amount.Int())
+		total.Int().Sub(total.Int(), transfer.Amount.Int())
 	} else {
-		balance.Int().Add(balance.Int(), transfer.Amount.Int())
+		total.Int().Add(total.Int(), transfer.Amount.Int())
 	}
 
-	if account != nil {
+	if balance != nil {
 		if _, err = s.updateTx(ctx, tokenbalanceTable, tx,
 			sq.Update(tokenbalanceTable).
 				Set("uri", transfer.URI).
-				Set("balance", balance).
+				Set("balance", total).
 				Set("updated", fftypes.Now()).
-				Where(sq.And{
-					sq.Eq{"pool_id": account.Pool},
-					sq.Eq{"token_index": account.TokenIndex},
-					sq.Eq{"key": account.Key},
+				Where(sq.Eq{
+					"namespace":   balance.Namespace,
+					"pool_id":     balance.Pool,
+					"token_index": balance.TokenIndex,
+					"key":         balance.Key,
 				}),
 			nil,
 		); err != nil {
@@ -92,7 +93,7 @@ func (s *SQLCommon) addTokenBalance(ctx context.Context, tx *txWrapper, transfer
 					transfer.Connector,
 					transfer.Namespace,
 					key,
-					balance,
+					total,
 					fftypes.Now(),
 				),
 			nil,
diff --git a/internal/database/sqlcommon/tokenpool_sql.go b/internal/database/sqlcommon/tokenpool_sql.go
index d66a7e735..e0c49f2e8 100644
--- a/internal/database/sqlcommon/tokenpool_sql.go
+++ b/internal/database/sqlcommon/tokenpool_sql.go
@@ -66,7 +66,10 @@ func (s *SQLCommon) UpsertTokenPool(ctx context.Context, pool *core.TokenPool) (
 	rows, _, err := s.queryTx(ctx, tokenpoolTable, tx,
 		sq.Select("id").
 			From(tokenpoolTable).
-			Where(sq.And{sq.Eq{"namespace": pool.Namespace}, sq.Eq{"name": pool.Name}}),
+			Where(sq.Eq{
+				"namespace": pool.Namespace,
+				"name":      pool.Name,
+			}),
 	)
 	if err != nil {
 		return err
@@ -87,7 +90,6 @@ func (s *SQLCommon) UpsertTokenPool(ctx context.Context, pool *core.TokenPool) (
 	if existing {
 		if _, err = s.updateTx(ctx, tokenpoolTable, tx,
 			sq.Update(tokenpoolTable).
-				Set("namespace", pool.Namespace).
 				Set("name", pool.Name).
 				Set("standard", pool.Standard).
 				Set("locator", pool.Locator).
diff --git a/internal/database/sqlcommon/tokentransfer_sql.go b/internal/database/sqlcommon/tokentransfer_sql.go
index d66ad13ab..de2342365 100644
--- a/internal/database/sqlcommon/tokentransfer_sql.go
+++ b/internal/database/sqlcommon/tokentransfer_sql.go
@@ -78,7 +78,10 @@ func (s *SQLCommon) UpsertTokenTransfer(ctx context.Context, transfer *core.Toke
 	rows, _, err := s.queryTx(ctx, tokentransferTable, tx,
 		sq.Select("seq").
 			From(tokentransferTable).
-			Where(sq.Eq{"protocol_id": transfer.ProtocolID}),
+			Where(sq.Eq{
+				"protocol_id": transfer.ProtocolID,
+				"namespace":   transfer.Namespace,
+			}),
 	)
 	if err != nil {
 		return err
@@ -95,7 +98,6 @@ func (s *SQLCommon) UpsertTokenTransfer(ctx context.Context, transfer *core.Toke
 				Set("token_index", transfer.TokenIndex).
 				Set("uri", transfer.URI).
 				Set("connector", transfer.Connector).
-				Set("namespace", transfer.Namespace).
 				Set("key", transfer.Key).
 				Set("from_key", transfer.From).
 				Set("to_key", transfer.To).
diff --git a/internal/database/sqlcommon/transaction_sql.go b/internal/database/sqlcommon/transaction_sql.go
index d3d84b611..f9f9fbe74 100644
--- a/internal/database/sqlcommon/transaction_sql.go
+++ b/internal/database/sqlcommon/transaction_sql.go
@@ -139,7 +139,7 @@ func (s *SQLCommon) GetTransactions(ctx context.Context, namespace string, filte
 
 }
 
-func (s *SQLCommon) UpdateTransaction(ctx context.Context, id *fftypes.UUID, update database.Update) (err error) {
+func (s *SQLCommon) UpdateTransaction(ctx context.Context, namespace string, id *fftypes.UUID, update database.Update) (err error) {
 
 	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
 	if err != nil {
@@ -151,7 +151,7 @@ func (s *SQLCommon) UpdateTransaction(ctx context.Context, id *fftypes.UUID, upd
 	if err != nil {
 		return err
 	}
-	query = query.Where(sq.Eq{"id": id})
+	query = query.Where(sq.Eq{"id": id, "namespace": namespace})
 
 	_, err = s.updateTx(ctx, transactionsTable, tx, query, nil /* no change evnents for filter based updates */)
 	if err != nil {
diff --git a/internal/database/sqlcommon/verifier_sql.go b/internal/database/sqlcommon/verifier_sql.go
index 2392d42de..d62239d02 100644
--- a/internal/database/sqlcommon/verifier_sql.go
+++ b/internal/database/sqlcommon/verifier_sql.go
@@ -50,7 +50,6 @@ func (s *SQLCommon) attemptVerifierUpdate(ctx context.Context, tx *txWrapper, ve
 		sq.Update(verifiersTable).
 			Set("identity", verifier.Identity).
 			Set("vtype", verifier.Type).
-			Set("namespace", verifier.Namespace).
 			Set("value", verifier.Value).
 			Where(sq.Eq{
 				"hash": verifier.Hash,
@@ -100,7 +99,10 @@ func (s *SQLCommon) UpsertVerifier(ctx context.Context, verifier *core.Verifier,
 		msgRows, _, err := s.queryTx(ctx, verifiersTable, tx,
 			sq.Select("hash").
 				From(verifiersTable).
-				Where(sq.Eq{"hash": verifier.Hash}),
+				Where(sq.Eq{
+					"namespace": verifier.Namespace,
+					"hash":      verifier.Hash,
+				}),
 		)
 		if err != nil {
 			return err
diff --git a/internal/events/aggregator.go b/internal/events/aggregator.go
index 8c6d966c5..5fb50f209 100644
--- a/internal/events/aggregator.go
+++ b/internal/events/aggregator.go
@@ -488,7 +488,7 @@ func (ag *aggregator) processMessage(ctx context.Context, manifest *core.BatchMa
 	// - dispatched=false: we need to prevent dispatch of any subsequent messages on the same topic in the batch
 	if dispatched {
 		for _, np := range nextPins {
-			np.IncrementNextPin(ctx)
+			np.IncrementNextPin(ctx, ag.namespace)
 		}
 		state.markMessageDispatched(manifest.ID, msg, msgBaseIndex, newState)
 	} else {
diff --git a/internal/events/aggregator_batch_state.go b/internal/events/aggregator_batch_state.go
index 5448c49f5..085f50b59 100644
--- a/internal/events/aggregator_batch_state.go
+++ b/internal/events/aggregator_batch_state.go
@@ -165,7 +165,7 @@ func (bs *batchState) checkMaskedContextReady(ctx context.Context, msg *core.Mes
 	h.Write([]byte(topic))
 	h.Write((*msg.Header.Group)[:])
 	contextUnmasked := fftypes.HashResult(h)
-	npg, err := bs.stateForMaskedContext(ctx, msg.Header.Group, topic, *contextUnmasked)
+	npg, err := bs.stateForMaskedContext(ctx, msg.Header.Group, topic, contextUnmasked)
 	if err != nil {
 		return nil, err
 	}
@@ -236,7 +236,7 @@ func (bs *batchState) flushPins(ctx context.Context) error {
 				update := database.NextPinQueryFactory.NewUpdate(ctx).
 					Set("nonce", np.Nonce).
 					Set("hash", np.Hash)
-				if err := bs.database.UpdateNextPin(ctx, np.Sequence, update); err != nil {
+				if err := bs.database.UpdateNextPin(ctx, bs.namespace, np.Sequence, update); err != nil {
 					return err
 				}
 			}
@@ -273,7 +273,7 @@ func (bs *batchState) flushPins(ctx context.Context) error {
 			))
 		}
 		update := database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true)
-		if err := bs.database.UpdatePins(ctx, filter, update); err != nil {
+		if err := bs.database.UpdatePins(ctx, bs.namespace, filter, update); err != nil {
 			return err
 		}
 	}
@@ -291,7 +291,7 @@ func (bs *batchState) flushPins(ctx context.Context) error {
 		setConfirmed := database.MessageQueryFactory.NewUpdate(ctx).
 			Set("confirmed", confirmTime).
 			Set("state", msgState)
-		if err := bs.database.UpdateMessages(ctx, filter, setConfirmed); err != nil {
+		if err := bs.database.UpdateMessages(ctx, bs.namespace, filter, setConfirmed); err != nil {
 			return err
 		}
 	}
@@ -299,7 +299,7 @@ func (bs *batchState) flushPins(ctx context.Context) error {
 	return nil
 }
 
-func (nps *nextPinState) IncrementNextPin(ctx context.Context) {
+func (nps *nextPinState) IncrementNextPin(ctx context.Context, namespace string) {
 	npg := nps.nextPinGroup
 	np := nps.nextPin
 	for i, existingPin := range npg.nextPins {
@@ -307,11 +307,12 @@ func (nps *nextPinState) IncrementNextPin(ctx context.Context) {
 			// We are spending this one, replace it in the list
 			newNonce := np.Nonce + 1
 			newNextPin := &core.NextPin{
-				Context:  np.Context,
-				Identity: np.Identity,
-				Nonce:    newNonce,
-				Hash:     npg.calcPinHash(np.Identity, newNonce),
-				Sequence: np.Sequence, // used for update in Flush
+				Namespace: namespace,
+				Context:   np.Context,
+				Identity:  np.Identity,
+				Nonce:     newNonce,
+				Hash:      npg.calcPinHash(np.Identity, newNonce),
+				Sequence:  np.Sequence, // used for update in Flush
 			}
 			npg.nextPins[i] = newNextPin
 			log.L(ctx).Debugf("Incrementing NextPin=%s - Nonce=%d Topic=%s Group=%s NewNextPin=%s", np.Hash, np.Nonce, npg.topic, npg.groupID.String(), newNextPin.Hash)
@@ -332,14 +333,13 @@ func (npg *nextPinGroupState) calcPinHash(identity string, nonce int64) *fftypes
 	return fftypes.HashResult(h)
 }
 
-func (bs *batchState) stateForMaskedContext(ctx context.Context, groupID *fftypes.Bytes32, topic string, contextUnmasked fftypes.Bytes32) (*nextPinGroupState, error) {
+func (bs *batchState) stateForMaskedContext(ctx context.Context, groupID *fftypes.Bytes32, topic string, contextUnmasked *fftypes.Bytes32) (*nextPinGroupState, error) {
 
-	if npg, exists := bs.maskedContexts[contextUnmasked]; exists {
+	if npg, exists := bs.maskedContexts[*contextUnmasked]; exists {
 		return npg, nil
 	}
 
-	filter := database.NextPinQueryFactory.NewFilter(ctx).Eq("context", contextUnmasked)
-	nextPins, _, err := bs.database.GetNextPins(ctx, filter)
+	nextPins, err := bs.database.GetNextPinsForContext(ctx, bs.namespace, contextUnmasked)
 	if err != nil {
 		return nil, err
 	}
@@ -354,7 +354,7 @@ func (bs *batchState) stateForMaskedContext(ctx context.Context, groupID *fftype
 		identitiesChanged: make(map[string]bool),
 		nextPins:          nextPins,
 	}
-	bs.maskedContexts[contextUnmasked] = npg
+	bs.maskedContexts[*contextUnmasked] = npg
 	return npg, nil
 
 }
@@ -383,10 +383,11 @@ func (bs *batchState) attemptContextInit(ctx context.Context, msg *core.Message,
 	for i, member := range group.Members {
 		zeroHash := npg.calcPinHash(member.Identity, 0)
 		np := &core.NextPin{
-			Context:  contextUnmasked,
-			Identity: member.Identity,
-			Hash:     zeroHash,
-			Nonce:    0,
+			Namespace: bs.namespace,
+			Context:   contextUnmasked,
+			Identity:  member.Identity,
+			Hash:      zeroHash,
+			Nonce:     0,
 		}
 		if *pin == *zeroHash {
 			if member.Identity != msg.Header.Author {
diff --git a/internal/events/dx_callbacks.go b/internal/events/dx_callbacks.go
index 381fe7d97..65bdd477d 100644
--- a/internal/events/dx_callbacks.go
+++ b/internal/events/dx_callbacks.go
@@ -157,7 +157,7 @@ func (em *eventManager) markUnpinnedMessagesConfirmed(ctx context.Context, batch
 		Set("state", core.MessageStateConfirmed).
 		Set("confirmed", fftypes.Now())
 
-	if err := em.database.UpdateMessages(ctx, filter, update); err != nil {
+	if err := em.database.UpdateMessages(ctx, em.namespace.LocalName, filter, update); err != nil {
 		return err
 	}
 
diff --git a/internal/events/ss_callbacks.go b/internal/events/ss_callbacks.go
index 78d3bb744..e1ee63f85 100644
--- a/internal/events/ss_callbacks.go
+++ b/internal/events/ss_callbacks.go
@@ -31,7 +31,7 @@ func (em *eventManager) SharedStorageBatchDownloaded(ss sharedstorage.Plugin, pa
 	l := log.L(em.ctx)
 
 	if em.multiparty == nil {
-		l.Errorf("Ignoring private batch from non-multiparty network!")
+		l.Errorf("Ignoring batch from non-multiparty network!")
 		return nil, nil
 	}
 
diff --git a/internal/privatemessaging/groupmanager.go b/internal/privatemessaging/groupmanager.go
index 5815716f5..ebffe30f9 100644
--- a/internal/privatemessaging/groupmanager.go
+++ b/internal/privatemessaging/groupmanager.go
@@ -102,6 +102,7 @@ func (gm *groupManager) groupInit(ctx context.Context, signer *core.SignerRef, g
 	if err != nil {
 		return i18n.WrapError(ctx, err, coremsgs.MsgSerializationFailed)
 	}
+	group.LocalNamespace = gm.namespace.LocalName
 
 	// In the case of groups, we actually write the unconfirmed group directly to our database.
 	// So it can be used straight away.
@@ -229,6 +230,7 @@ func (gm *groupManager) ResolveInitGroup(ctx context.Context, msg *core.Message)
 			return nil, nil
 		}
 		newGroup.Message = msg.Header.ID
+		newGroup.LocalNamespace = gm.namespace.LocalName
 		err = gm.database.UpsertGroup(ctx, &newGroup, database.UpsertOptimizationNew /* we think we're first to create this */)
 		if err != nil {
 			return nil, err
diff --git a/internal/privatemessaging/privatemessaging.go b/internal/privatemessaging/privatemessaging.go
index 322c27304..3e3cff64e 100644
--- a/internal/privatemessaging/privatemessaging.go
+++ b/internal/privatemessaging/privatemessaging.go
@@ -263,8 +263,8 @@ func (pm *privateMessaging) sendData(ctx context.Context, tw *core.TransportWrap
 	l := log.L(ctx)
 	batch := tw.Batch
 
-	// Lookup the local org
-	localOrg, err := pm.identity.GetMultipartyRootOrg(ctx)
+	// Lookup the local node
+	localNode, err := pm.identity.GetLocalNode(ctx)
 	if err != nil {
 		return err
 	}
@@ -272,7 +272,7 @@ func (pm *privateMessaging) sendData(ctx context.Context, tw *core.TransportWrap
 	// Write it to the dataexchange for each member
 	for i, node := range nodes {
 
-		if node.Parent.Equals(localOrg.ID) {
+		if node.ID.Equals(localNode.ID) {
 			l.Debugf("Skipping send of batch for local node %s for group=%s node=%s (%d/%d)", batch.ID, batch.Group, node.ID, i+1, len(nodes))
 			continue
 		}
diff --git a/internal/privatemessaging/recipients.go b/internal/privatemessaging/recipients.go
index 81c375b54..633f70529 100644
--- a/internal/privatemessaging/recipients.go
+++ b/internal/privatemessaging/recipients.go
@@ -162,9 +162,8 @@ func (pm *privateMessaging) findOrGenerateGroup(ctx context.Context, in *core.Me
 	// exists. If it doesn't, we go ahead and create it. If it does - we don't return
 	// this candidate - we return the existing group.
 	newCandidate := &core.Group{
-		LocalNamespace: pm.namespace.LocalName,
-		GroupIdentity:  *gi,
-		Created:        fftypes.Now(),
+		GroupIdentity: *gi,
+		Created:       fftypes.Now(),
 	}
 	newCandidate.Seal()
 
diff --git a/internal/txcommon/txcommon.go b/internal/txcommon/txcommon.go
index 2a3544b41..3f5dbc52d 100644
--- a/internal/txcommon/txcommon.go
+++ b/internal/txcommon/txcommon.go
@@ -128,7 +128,7 @@ func (t *transactionHelper) PersistTransaction(ctx context.Context, id *fftypes.
 			return true, nil
 		}
 
-		if err = t.database.UpdateTransaction(ctx, tx.ID, database.TransactionQueryFactory.NewUpdate(ctx).Set("blockchainids", newBlockchainIDs)); err != nil {
+		if err = t.database.UpdateTransaction(ctx, t.namespace, tx.ID, database.TransactionQueryFactory.NewUpdate(ctx).Set("blockchainids", newBlockchainIDs)); err != nil {
 			return false, err
 		}
 	} else {
@@ -158,7 +158,7 @@ func (t *transactionHelper) AddBlockchainTX(ctx context.Context, tx *core.Transa
 		return nil
 	}
 
-	err := t.database.UpdateTransaction(ctx, tx.ID, database.TransactionQueryFactory.NewUpdate(ctx).Set("blockchainids", tx.BlockchainIDs))
+	err := t.database.UpdateTransaction(ctx, t.namespace, tx.ID, database.TransactionQueryFactory.NewUpdate(ctx).Set("blockchainids", tx.BlockchainIDs))
 	if err != nil {
 		return err
 	}
diff --git a/mocks/databasemocks/plugin.go b/mocks/databasemocks/plugin.go
index dfa4aa140..4c4630f3b 100644
--- a/mocks/databasemocks/plugin.go
+++ b/mocks/databasemocks/plugin.go
@@ -65,20 +65,6 @@ func (_m *Plugin) DeleteContractListenerByID(ctx context.Context, namespace stri
 	return r0
 }
 
-// DeleteNextPin provides a mock function with given fields: ctx, sequence
-func (_m *Plugin) DeleteNextPin(ctx context.Context, sequence int64) error {
-	ret := _m.Called(ctx, sequence)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, int64) error); ok {
-		r0 = rf(ctx, sequence)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
 // DeleteNonce provides a mock function with given fields: ctx, hash
 func (_m *Plugin) DeleteNonce(ctx context.Context, hash *fftypes.Bytes32) error {
 	ret := _m.Called(ctx, hash)
@@ -1254,45 +1240,22 @@ func (_m *Plugin) GetNamespace(ctx context.Context, name string) (*core.Namespac
 	return r0, r1
 }
 
-// GetNextPinByContextAndIdentity provides a mock function with given fields: ctx, _a1, identity
-func (_m *Plugin) GetNextPinByContextAndIdentity(ctx context.Context, _a1 *fftypes.Bytes32, identity string) (*core.NextPin, error) {
-	ret := _m.Called(ctx, _a1, identity)
-
-	var r0 *core.NextPin
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.Bytes32, string) *core.NextPin); ok {
-		r0 = rf(ctx, _a1, identity)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*core.NextPin)
-		}
-	}
+// GetNextPinsForContext provides a mock function with given fields: ctx, namespace, _a2
+func (_m *Plugin) GetNextPinsForContext(ctx context.Context, namespace string, _a2 *fftypes.Bytes32) ([]*core.NextPin, error) {
+	ret := _m.Called(ctx, namespace, _a2)
 
-	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.Bytes32, string) error); ok {
-		r1 = rf(ctx, _a1, identity)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
-// GetNextPinByHash provides a mock function with given fields: ctx, hash
-func (_m *Plugin) GetNextPinByHash(ctx context.Context, hash *fftypes.Bytes32) (*core.NextPin, error) {
-	ret := _m.Called(ctx, hash)
-
-	var r0 *core.NextPin
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.Bytes32) *core.NextPin); ok {
-		r0 = rf(ctx, hash)
+	var r0 []*core.NextPin
+	if rf, ok := ret.Get(0).(func(context.Context, string, *fftypes.Bytes32) []*core.NextPin); ok {
+		r0 = rf(ctx, namespace, _a2)
 	} else {
 		if ret.Get(0) != nil {
-			r0 = ret.Get(0).(*core.NextPin)
+			r0 = ret.Get(0).([]*core.NextPin)
 		}
 	}
 
 	var r1 error
-	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.Bytes32) error); ok {
-		r1 = rf(ctx, hash)
+	if rf, ok := ret.Get(1).(func(context.Context, string, *fftypes.Bytes32) error); ok {
+		r1 = rf(ctx, namespace, _a2)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -1300,38 +1263,6 @@ func (_m *Plugin) GetNextPinByHash(ctx context.Context, hash *fftypes.Bytes32) (
 	return r0, r1
 }
 
-// GetNextPins provides a mock function with given fields: ctx, filter
-func (_m *Plugin) GetNextPins(ctx context.Context, filter database.Filter) ([]*core.NextPin, *database.FilterResult, error) {
-	ret := _m.Called(ctx, filter)
-
-	var r0 []*core.NextPin
-	if rf, ok := ret.Get(0).(func(context.Context, database.Filter) []*core.NextPin); ok {
-		r0 = rf(ctx, filter)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).([]*core.NextPin)
-		}
-	}
-
-	var r1 *database.FilterResult
-	if rf, ok := ret.Get(1).(func(context.Context, database.Filter) *database.FilterResult); ok {
-		r1 = rf(ctx, filter)
-	} else {
-		if ret.Get(1) != nil {
-			r1 = ret.Get(1).(*database.FilterResult)
-		}
-	}
-
-	var r2 error
-	if rf, ok := ret.Get(2).(func(context.Context, database.Filter) error); ok {
-		r2 = rf(ctx, filter)
-	} else {
-		r2 = ret.Error(2)
-	}
-
-	return r0, r1, r2
-}
-
 // GetNonce provides a mock function with given fields: ctx, hash
 func (_m *Plugin) GetNonce(ctx context.Context, hash *fftypes.Bytes32) (*core.Nonce, error) {
 	ret := _m.Called(ctx, hash)
@@ -2378,13 +2309,13 @@ func (_m *Plugin) SetHandler(namespace string, handler database.Callbacks) {
 	_m.Called(namespace, handler)
 }
 
-// UpdateBatch provides a mock function with given fields: ctx, id, update
-func (_m *Plugin) UpdateBatch(ctx context.Context, id *fftypes.UUID, update database.Update) error {
-	ret := _m.Called(ctx, id, update)
+// UpdateBatch provides a mock function with given fields: ctx, namespace, id, update
+func (_m *Plugin) UpdateBatch(ctx context.Context, namespace string, id *fftypes.UUID, update database.Update) error {
+	ret := _m.Called(ctx, namespace, id, update)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, database.Update) error); ok {
-		r0 = rf(ctx, id, update)
+	if rf, ok := ret.Get(0).(func(context.Context, string, *fftypes.UUID, database.Update) error); ok {
+		r0 = rf(ctx, namespace, id, update)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -2392,13 +2323,13 @@ func (_m *Plugin) UpdateBatch(ctx context.Context, id *fftypes.UUID, update data
 	return r0
 }
 
-// UpdateData provides a mock function with given fields: ctx, id, update
-func (_m *Plugin) UpdateData(ctx context.Context, id *fftypes.UUID, update database.Update) error {
-	ret := _m.Called(ctx, id, update)
+// UpdateData provides a mock function with given fields: ctx, namespace, id, update
+func (_m *Plugin) UpdateData(ctx context.Context, namespace string, id *fftypes.UUID, update database.Update) error {
+	ret := _m.Called(ctx, namespace, id, update)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, database.Update) error); ok {
-		r0 = rf(ctx, id, update)
+	if rf, ok := ret.Get(0).(func(context.Context, string, *fftypes.UUID, database.Update) error); ok {
+		r0 = rf(ctx, namespace, id, update)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -2406,13 +2337,13 @@ func (_m *Plugin) UpdateData(ctx context.Context, id *fftypes.UUID, update datab
 	return r0
 }
 
-// UpdateMessage provides a mock function with given fields: ctx, id, update
-func (_m *Plugin) UpdateMessage(ctx context.Context, id *fftypes.UUID, update database.Update) error {
-	ret := _m.Called(ctx, id, update)
+// UpdateMessage provides a mock function with given fields: ctx, namespace, id, update
+func (_m *Plugin) UpdateMessage(ctx context.Context, namespace string, id *fftypes.UUID, update database.Update) error {
+	ret := _m.Called(ctx, namespace, id, update)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, database.Update) error); ok {
-		r0 = rf(ctx, id, update)
+	if rf, ok := ret.Get(0).(func(context.Context, string, *fftypes.UUID, database.Update) error); ok {
+		r0 = rf(ctx, namespace, id, update)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -2420,13 +2351,13 @@ func (_m *Plugin) UpdateMessage(ctx context.Context, id *fftypes.UUID, update da
 	return r0
 }
 
-// UpdateMessages provides a mock function with given fields: ctx, filter, update
-func (_m *Plugin) UpdateMessages(ctx context.Context, filter database.Filter, update database.Update) error {
-	ret := _m.Called(ctx, filter, update)
+// UpdateMessages provides a mock function with given fields: ctx, namespace, filter, update
+func (_m *Plugin) UpdateMessages(ctx context.Context, namespace string, filter database.Filter, update database.Update) error {
+	ret := _m.Called(ctx, namespace, filter, update)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, database.Filter, database.Update) error); ok {
-		r0 = rf(ctx, filter, update)
+	if rf, ok := ret.Get(0).(func(context.Context, string, database.Filter, database.Update) error); ok {
+		r0 = rf(ctx, namespace, filter, update)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -2434,13 +2365,13 @@ func (_m *Plugin) UpdateMessages(ctx context.Context, filter database.Filter, up
 	return r0
 }
 
-// UpdateNextPin provides a mock function with given fields: ctx, sequence, update
-func (_m *Plugin) UpdateNextPin(ctx context.Context, sequence int64, update database.Update) error {
-	ret := _m.Called(ctx, sequence, update)
+// UpdateNextPin provides a mock function with given fields: ctx, namespace, sequence, update
+func (_m *Plugin) UpdateNextPin(ctx context.Context, namespace string, sequence int64, update database.Update) error {
+	ret := _m.Called(ctx, namespace, sequence, update)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, int64, database.Update) error); ok {
-		r0 = rf(ctx, sequence, update)
+	if rf, ok := ret.Get(0).(func(context.Context, string, int64, database.Update) error); ok {
+		r0 = rf(ctx, namespace, sequence, update)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -2490,13 +2421,13 @@ func (_m *Plugin) UpdateOperation(ctx context.Context, namespace string, id *fft
 	return r0
 }
 
-// UpdatePins provides a mock function with given fields: ctx, filter, update
-func (_m *Plugin) UpdatePins(ctx context.Context, filter database.Filter, update database.Update) error {
-	ret := _m.Called(ctx, filter, update)
+// UpdatePins provides a mock function with given fields: ctx, namespace, filter, update
+func (_m *Plugin) UpdatePins(ctx context.Context, namespace string, filter database.Filter, update database.Update) error {
+	ret := _m.Called(ctx, namespace, filter, update)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, database.Filter, database.Update) error); ok {
-		r0 = rf(ctx, filter, update)
+	if rf, ok := ret.Get(0).(func(context.Context, string, database.Filter, database.Update) error); ok {
+		r0 = rf(ctx, namespace, filter, update)
 	} else {
 		r0 = ret.Error(0)
 	}
@@ -2546,13 +2477,13 @@ func (_m *Plugin) UpdateTokenBalances(ctx context.Context, transfer *core.TokenT
 	return r0
 }
 
-// UpdateTransaction provides a mock function with given fields: ctx, id, update
-func (_m *Plugin) UpdateTransaction(ctx context.Context, id *fftypes.UUID, update database.Update) error {
-	ret := _m.Called(ctx, id, update)
+// UpdateTransaction provides a mock function with given fields: ctx, namespace, id, update
+func (_m *Plugin) UpdateTransaction(ctx context.Context, namespace string, id *fftypes.UUID, update database.Update) error {
+	ret := _m.Called(ctx, namespace, id, update)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, database.Update) error); ok {
-		r0 = rf(ctx, id, update)
+	if rf, ok := ret.Get(0).(func(context.Context, string, *fftypes.UUID, database.Update) error); ok {
+		r0 = rf(ctx, namespace, id, update)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/pkg/core/nextpin.go b/pkg/core/nextpin.go
index e5d371976..c49bad14f 100644
--- a/pkg/core/nextpin.go
+++ b/pkg/core/nextpin.go
@@ -19,9 +19,10 @@ package core
 import "github.com/hyperledger/firefly-common/pkg/fftypes"
 
 type NextPin struct {
-	Context  *fftypes.Bytes32 `json:"context"`
-	Identity string           `json:"identity"`
-	Hash     *fftypes.Bytes32 `json:"hash"`
-	Nonce    int64            `json:"nonce"`
-	Sequence int64            `json:"_"` // Local database sequence used internally for update efficiency
+	Namespace string           `json:"namespace"`
+	Context   *fftypes.Bytes32 `json:"context"`
+	Identity  string           `json:"identity"`
+	Hash      *fftypes.Bytes32 `json:"hash"`
+	Nonce     int64            `json:"nonce"`
+	Sequence  int64            `json:"_"` // Local database sequence used internally for update efficiency
 }
diff --git a/pkg/database/plugin.go b/pkg/database/plugin.go
index 3beb80d50..cd15af04a 100644
--- a/pkg/database/plugin.go
+++ b/pkg/database/plugin.go
@@ -84,14 +84,14 @@ type iMessageCollection interface {
 	InsertMessages(ctx context.Context, messages []*core.Message, hooks ...PostCompletionHook) (err error)
 
 	// UpdateMessage - Update message
-	UpdateMessage(ctx context.Context, id *fftypes.UUID, update Update) (err error)
+	UpdateMessage(ctx context.Context, namespace string, id *fftypes.UUID, update Update) (err error)
 
 	// ReplaceMessage updates the message, and assigns it a new sequence number at the front of the list.
 	// A new event is raised for the message, with the new sequence number - as if it was brand new.
 	ReplaceMessage(ctx context.Context, message *core.Message) (err error)
 
 	// UpdateMessages - Update messages
-	UpdateMessages(ctx context.Context, filter Filter, update Update) (err error)
+	UpdateMessages(ctx context.Context, namespace string, filter Filter, update Update) (err error)
 
 	// GetMessageByID - Get a message by ID
 	GetMessageByID(ctx context.Context, namespace string, id *fftypes.UUID) (message *core.Message, err error)
@@ -122,7 +122,7 @@ type iDataCollection interface {
 	InsertDataArray(ctx context.Context, data core.DataArray) (err error)
 
 	// UpdateData - Update data
-	UpdateData(ctx context.Context, id *fftypes.UUID, update Update) (err error)
+	UpdateData(ctx context.Context, namespace string, id *fftypes.UUID, update Update) (err error)
 
 	// GetDataByID - Get a data record by ID
 	GetDataByID(ctx context.Context, namespace string, id *fftypes.UUID, withValue bool) (message *core.Data, err error)
@@ -139,7 +139,7 @@ type iBatchCollection interface {
 	UpsertBatch(ctx context.Context, data *core.BatchPersisted) (err error)
 
 	// UpdateBatch - Update data
-	UpdateBatch(ctx context.Context, id *fftypes.UUID, update Update) (err error)
+	UpdateBatch(ctx context.Context, namespace string, id *fftypes.UUID, update Update) (err error)
 
 	// GetBatchByID - Get a batch by ID
 	GetBatchByID(ctx context.Context, namespace string, id *fftypes.UUID) (message *core.BatchPersisted, err error)
@@ -153,7 +153,7 @@ type iTransactionCollection interface {
 	InsertTransaction(ctx context.Context, data *core.Transaction) (err error)
 
 	// UpdateTransaction - Update transaction
-	UpdateTransaction(ctx context.Context, id *fftypes.UUID, update Update) (err error)
+	UpdateTransaction(ctx context.Context, namespace string, id *fftypes.UUID, update Update) (err error)
 
 	// GetTransactionByID - Get a transaction by ID
 	GetTransactionByID(ctx context.Context, namespace string, id *fftypes.UUID) (message *core.Transaction, err error)
@@ -204,7 +204,7 @@ type iPinCollection interface {
 	GetPins(ctx context.Context, namespace string, filter Filter) (offset []*core.Pin, res *FilterResult, err error)
 
 	// UpdatePins - Updates pins
-	UpdatePins(ctx context.Context, filter Filter, update Update) (err error)
+	UpdatePins(ctx context.Context, namespace string, filter Filter, update Update) (err error)
 }
 
 type iOperationCollection interface {
@@ -322,20 +322,11 @@ type iNextPinCollection interface {
 	// InsertNextPin - insert a nextpin
 	InsertNextPin(ctx context.Context, nextpin *core.NextPin) (err error)
 
-	// GetNextPinByContextAndIdentity - lookup nextpin by context+identity
-	GetNextPinByContextAndIdentity(ctx context.Context, context *fftypes.Bytes32, identity string) (message *core.NextPin, err error)
-
-	// GetNextPinByHash - lookup nextpin by its hash
-	GetNextPinByHash(ctx context.Context, hash *fftypes.Bytes32) (message *core.NextPin, err error)
-
 	// GetNextPins - get nextpins
-	GetNextPins(ctx context.Context, filter Filter) (message []*core.NextPin, res *FilterResult, err error)
+	GetNextPinsForContext(ctx context.Context, namespace string, context *fftypes.Bytes32) (message []*core.NextPin, err error)
 
 	// UpdateNextPin - update a next hash using its local database ID
-	UpdateNextPin(ctx context.Context, sequence int64, update Update) (err error)
-
-	// DeleteNextPin - delete a next hash, using its local database ID
-	DeleteNextPin(ctx context.Context, sequence int64) (err error)
+	UpdateNextPin(ctx context.Context, namespace string, sequence int64, update Update) (err error)
 }
 
 type iBlobCollection interface {
@@ -830,7 +821,7 @@ var VerifierQueryFactory = &queryFields{
 	"created":  &TimeField{},
 }
 
-// GroupQueryFactory filter fields for nodes
+// GroupQueryFactory filter fields for groups
 var GroupQueryFactory = &queryFields{
 	"hash":        &Bytes32Field{},
 	"message":     &UUIDField{},
@@ -839,13 +830,13 @@ var GroupQueryFactory = &queryFields{
 	"created":     &TimeField{},
 }
 
-// NonceQueryFactory filter fields for nodes
+// NonceQueryFactory filter fields for nonces
 var NonceQueryFactory = &queryFields{
 	"hash":  &StringField{},
 	"nonce": &Int64Field{},
 }
 
-// NextPinQueryFactory filter fields for nodes
+// NextPinQueryFactory filter fields for next pins
 var NextPinQueryFactory = &queryFields{
 	"context":  &Bytes32Field{},
 	"identity": &StringField{},


Test patch:
diff --git a/internal/batch/batch_manager_test.go b/internal/batch/batch_manager_test.go
index 4440af911..3394af264 100644
--- a/internal/batch/batch_manager_test.go
+++ b/internal/batch/batch_manager_test.go
@@ -125,14 +125,14 @@ func TestE2EDispatchBroadcast(t *testing.T) {
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).Return([]*core.IDAndSequence{}, nil)
 	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	mdi.On("UpdateBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
-	mdi.On("UpdateMessage", mock.Anything, mock.Anything, mock.Anything).Return(nil) // pins
+	mdi.On("UpdateMessage", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil) // pins
 	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	rag.RunFn = func(a mock.Arguments) {
 		ctx := a.Get(0).(context.Context)
 		fn := a.Get(1).(func(context.Context) error)
 		fn(ctx)
 	}
-	mdi.On("UpdateMessages", mock.Anything, mock.MatchedBy(func(f database.Filter) bool {
+	mdi.On("UpdateMessages", mock.Anything, "ns1", mock.MatchedBy(func(f database.Filter) bool {
 		fi, err := f.Finalize()
 		assert.NoError(t, err)
 		assert.Equal(t, fmt.Sprintf("( id IN ['%s'] ) && ( state == 'ready' )", msg.Header.ID.String()), fi.String())
@@ -241,7 +241,7 @@ func TestE2EDispatchPrivateUnpinned(t *testing.T) {
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).Return([]*core.IDAndSequence{{ID: *msg.Header.ID}}, nil).Once()
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).Return([]*core.IDAndSequence{}, nil)
-	mdi.On("UpdateMessage", mock.Anything, mock.Anything, mock.Anything).Return(nil) // pins
+	mdi.On("UpdateMessage", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil) // pins
 	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	mdi.On("UpdateBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything, mock.Anything).Return(nil)
@@ -250,7 +250,7 @@ func TestE2EDispatchPrivateUnpinned(t *testing.T) {
 		fn := a.Get(1).(func(context.Context) error)
 		fn(ctx)
 	}
-	mdi.On("UpdateMessages", mock.Anything, mock.MatchedBy(func(f database.Filter) bool {
+	mdi.On("UpdateMessages", mock.Anything, "ns1", mock.MatchedBy(func(f database.Filter) bool {
 		fi, err := f.Finalize()
 		assert.NoError(t, err)
 		assert.Equal(t, fmt.Sprintf("( id IN ['%s'] ) && ( state == 'ready' )", msg.Header.ID.String()), fi.String())
@@ -414,7 +414,7 @@ func TestMessageSequencerUpdateMessagesFail(t *testing.T) {
 	mdi.On("InsertTransaction", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil) // transaction submit
 	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
-	mdi.On("UpdateMessages", mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("fizzle"))
+	mdi.On("UpdateMessages", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(fmt.Errorf("fizzle"))
 	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything, mock.Anything)
 	rag.RunFn = func(a mock.Arguments) {
 		ctx := a.Get(0).(context.Context)
diff --git a/internal/batch/batch_processor_test.go b/internal/batch/batch_processor_test.go
index 4463db68f..a2a6c9fdc 100644
--- a/internal/batch/batch_processor_test.go
+++ b/internal/batch/batch_processor_test.go
@@ -81,7 +81,7 @@ func TestUnfilledBatch(t *testing.T) {
 	defer cancel()
 
 	mockRunAsGroupPassthrough(mdi)
-	mdi.On("UpdateMessages", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdateMessages", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 
 	mth := bp.txHelper.(*txcommonmocks.Helper)
@@ -126,7 +126,7 @@ func TestBatchSizeOverflow(t *testing.T) {
 	defer cancel()
 	bp.conf.BatchMaxBytes = batchSizeEstimateBase + (&core.Message{}).EstimateSize(false) + 100
 	mockRunAsGroupPassthrough(mdi)
-	mdi.On("UpdateMessages", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdateMessages", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 
 	mth := bp.txHelper.(*txcommonmocks.Helper)
@@ -401,7 +401,7 @@ func TestMarkMessageDispatchedUnpinnedOK(t *testing.T) {
 	bp.conf.txType = core.TransactionTypeUnpinned
 
 	mockRunAsGroupPassthrough(mdi)
-	mdi.On("UpdateMessages", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdateMessages", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(fmt.Errorf("pop")).Once()
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil)
@@ -454,7 +454,7 @@ func TestMaskContextsRetryAfterPinsAssigned(t *testing.T) {
 	mdi.On("UpdateNonce", mock.Anything, mock.MatchedBy(func(dbNonce *core.Nonce) bool {
 		return dbNonce.Nonce == 12347 // twice incremented
 	})).Return(nil).Once()
-	mdi.On("UpdateMessage", mock.Anything, mock.Anything, mock.Anything).Return(nil).Twice()
+	mdi.On("UpdateMessage", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil).Twice()
 	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 
 	mdm := bp.data.(*datamocks.Manager)
@@ -510,7 +510,7 @@ func TestMaskContextsUpdateMessageFail(t *testing.T) {
 	mockRunAsGroupPassthrough(mdi)
 	mdi.On("GetNonce", mock.Anything, mock.Anything).Return(nil, nil)
 	mdi.On("InsertNonce", mock.Anything, mock.Anything).Return(nil)
-	mdi.On("UpdateMessage", mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("pop")).Once()
+	mdi.On("UpdateMessage", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(fmt.Errorf("pop")).Once()
 
 	msg := &core.Message{
 		Header: core.MessageHeader{
diff --git a/internal/broadcast/operations_test.go b/internal/broadcast/operations_test.go
index 7b5ece3bc..046c8fb99 100644
--- a/internal/broadcast/operations_test.go
+++ b/internal/broadcast/operations_test.go
@@ -258,7 +258,7 @@ func TestPrepareAndRunUploadBlob(t *testing.T) {
 	mdi.On("GetBlobMatchingHash", mock.Anything, blob.Hash).Return(blob, nil)
 	mps.On("UploadData", context.Background(), mock.Anything).Return("123", nil)
 	mdx.On("DownloadBlob", context.Background(), mock.Anything).Return(reader, nil)
-	mdi.On("UpdateData", context.Background(), data.ID, mock.MatchedBy(func(update database.Update) bool {
+	mdi.On("UpdateData", context.Background(), "ns1", data.ID, mock.MatchedBy(func(update database.Update) bool {
 		info, _ := update.Finalize()
 		assert.Equal(t, 1, len(info.SetOperations))
 		assert.Equal(t, "blob.public", info.SetOperations[0].Field)
@@ -425,7 +425,7 @@ func TestRunOperationUploadBlobUpdateFail(t *testing.T) {
 	reader := ioutil.NopCloser(strings.NewReader("some data"))
 	mdx.On("DownloadBlob", context.Background(), mock.Anything).Return(reader, nil)
 	mps.On("UploadData", context.Background(), mock.Anything).Return("123", nil)
-	mdi.On("UpdateData", context.Background(), data.ID, mock.Anything).Return(fmt.Errorf("pop"))
+	mdi.On("UpdateData", context.Background(), "ns1", data.ID, mock.Anything).Return(fmt.Errorf("pop"))
 
 	_, complete, err := bm.RunOperation(context.Background(), opUploadBlob(op, data, blob))
 
diff --git a/internal/database/sqlcommon/batch_sql_test.go b/internal/database/sqlcommon/batch_sql_test.go
index 429038f78..969113e03 100644
--- a/internal/database/sqlcommon/batch_sql_test.go
+++ b/internal/database/sqlcommon/batch_sql_test.go
@@ -147,7 +147,7 @@ func TestBatch2EWithDB(t *testing.T) {
 	// Update
 	author2 := "0x222222"
 	up := database.BatchQueryFactory.NewUpdate(ctx).Set("author", author2)
-	err = s.UpdateBatch(ctx, batchID, up)
+	err = s.UpdateBatch(ctx, "ns1", batchID, up)
 	assert.NoError(t, err)
 
 	// Test find updated value
@@ -276,7 +276,7 @@ func TestBatchUpdateBeginFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
 	u := database.BatchQueryFactory.NewUpdate(context.Background()).Set("id", "anything")
-	err := s.UpdateBatch(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateBatch(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10114", err)
 }
 
@@ -284,7 +284,7 @@ func TestBatchUpdateBuildQueryFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin()
 	u := database.BatchQueryFactory.NewUpdate(context.Background()).Set("id", map[bool]bool{true: false})
-	err := s.UpdateBatch(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateBatch(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF00143.*id", err)
 }
 
@@ -294,6 +294,6 @@ func TestBatchUpdateFail(t *testing.T) {
 	mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	u := database.BatchQueryFactory.NewUpdate(context.Background()).Set("id", fftypes.NewUUID())
-	err := s.UpdateBatch(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateBatch(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10117", err)
 }
diff --git a/internal/database/sqlcommon/data_sql_test.go b/internal/database/sqlcommon/data_sql_test.go
index ffad60756..3e8a1f1b8 100644
--- a/internal/database/sqlcommon/data_sql_test.go
+++ b/internal/database/sqlcommon/data_sql_test.go
@@ -150,7 +150,7 @@ func TestDataE2EWithDB(t *testing.T) {
 	// Update
 	v2 := "2.0.0"
 	up := database.DataQueryFactory.NewUpdate(ctx).Set("datatype.version", v2)
-	err = s.UpdateData(ctx, dataID, up)
+	err = s.UpdateData(ctx, "ns1", dataID, up)
 	assert.NoError(t, err)
 
 	// Test find updated value
@@ -359,7 +359,7 @@ func TestDataUpdateBeginFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
 	u := database.DataQueryFactory.NewUpdate(context.Background()).Set("id", "anything")
-	err := s.UpdateData(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateData(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10114", err)
 }
 
@@ -367,7 +367,7 @@ func TestDataUpdateBuildQueryFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin()
 	u := database.DataQueryFactory.NewUpdate(context.Background()).Set("id", map[bool]bool{true: false})
-	err := s.UpdateData(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateData(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF00143.*id", err)
 }
 
@@ -377,6 +377,6 @@ func TestDataUpdateFail(t *testing.T) {
 	mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	u := database.DataQueryFactory.NewUpdate(context.Background()).Set("id", fftypes.NewUUID())
-	err := s.UpdateData(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateData(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10117", err)
 }
diff --git a/internal/database/sqlcommon/message_sql_test.go b/internal/database/sqlcommon/message_sql_test.go
index 9b78a820b..2246225a7 100644
--- a/internal/database/sqlcommon/message_sql_test.go
+++ b/internal/database/sqlcommon/message_sql_test.go
@@ -205,7 +205,7 @@ func TestUpsertE2EWithDB(t *testing.T) {
 	up := database.MessageQueryFactory.NewUpdate(ctx).
 		Set("group", gid2).
 		Set("batch", bid2)
-	err = s.UpdateMessage(ctx, msgID, up)
+	err = s.UpdateMessage(ctx, "ns12345", msgID, up)
 	assert.NoError(t, err)
 
 	// Test find updated value
@@ -480,7 +480,7 @@ func TestLoadMessageDataRefsQueryFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	msgID := fftypes.NewUUID()
 	mock.ExpectQuery("SELECT .*").WillReturnError(fmt.Errorf("pop"))
-	err := s.loadDataRefs(context.Background(), []*core.Message{
+	err := s.loadDataRefs(context.Background(), "ns1", []*core.Message{
 		{
 			Header: core.MessageHeader{ID: msgID},
 		},
@@ -493,7 +493,7 @@ func TestLoadMessageDataRefsScanFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	msgID := fftypes.NewUUID()
 	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{"data_id"}).AddRow("only one"))
-	err := s.loadDataRefs(context.Background(), []*core.Message{
+	err := s.loadDataRefs(context.Background(), "ns1", []*core.Message{
 		{
 			Header: core.MessageHeader{ID: msgID},
 		},
@@ -507,7 +507,7 @@ func TestLoadMessageDataRefsEmpty(t *testing.T) {
 	msgID := fftypes.NewUUID()
 	msg := &core.Message{Header: core.MessageHeader{ID: msgID}}
 	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{"data_id", "data_hash"}))
-	err := s.loadDataRefs(context.Background(), []*core.Message{msg})
+	err := s.loadDataRefs(context.Background(), "ns1", []*core.Message{msg})
 	assert.NoError(t, err)
 	assert.Equal(t, core.DataRefs{}, msg.Data)
 	assert.NoError(t, mock.ExpectationsWereMet())
@@ -607,7 +607,7 @@ func TestMessageUpdateBeginFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
 	u := database.MessageQueryFactory.NewUpdate(context.Background()).Set("id", "anything")
-	err := s.UpdateMessage(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateMessage(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10114", err)
 }
 
@@ -641,7 +641,7 @@ func TestMessageUpdateBuildQueryFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin()
 	u := database.MessageQueryFactory.NewUpdate(context.Background()).Set("id", map[bool]bool{true: false})
-	err := s.UpdateMessage(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateMessage(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF00143.*id", err)
 }
 
@@ -650,7 +650,7 @@ func TestMessagesUpdateBuildFilterFail(t *testing.T) {
 	mock.ExpectBegin()
 	f := database.MessageQueryFactory.NewFilter(context.Background()).Eq("id", map[bool]bool{true: false})
 	u := database.MessageQueryFactory.NewUpdate(context.Background()).Set("type", core.MessageTypeBroadcast)
-	err := s.UpdateMessages(context.Background(), f, u)
+	err := s.UpdateMessages(context.Background(), "ns1", f, u)
 	assert.Regexp(t, "FF00143.*id", err)
 }
 
@@ -660,7 +660,7 @@ func TestMessageUpdateFail(t *testing.T) {
 	mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	u := database.MessageQueryFactory.NewUpdate(context.Background()).Set("group", fftypes.NewRandB32())
-	err := s.UpdateMessage(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateMessage(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10117", err)
 }
 
diff --git a/internal/database/sqlcommon/nextpin_sql_test.go b/internal/database/sqlcommon/nextpin_sql_test.go
index 12ee58ae4..f05d909e0 100644
--- a/internal/database/sqlcommon/nextpin_sql_test.go
+++ b/internal/database/sqlcommon/nextpin_sql_test.go
@@ -39,61 +39,40 @@ func TestNextPinsE2EWithDB(t *testing.T) {
 
 	// Create a new nextpin entry
 	nextpin := &core.NextPin{
-		Context:  fftypes.NewRandB32(),
-		Identity: "0x12345",
-		Hash:     fftypes.NewRandB32(),
-		Nonce:    int64(12345),
+		Namespace: "ns",
+		Context:   fftypes.NewRandB32(),
+		Identity:  "0x12345",
+		Hash:      fftypes.NewRandB32(),
+		Nonce:     int64(12345),
 	}
 	err := s.InsertNextPin(ctx, nextpin)
 	assert.NoError(t, err)
 
 	// Check we get the exact same nextpin back
-	nextpinRead, err := s.GetNextPinByContextAndIdentity(ctx, nextpin.Context, nextpin.Identity)
+	nextpinRead, err := s.GetNextPinsForContext(ctx, "ns", nextpin.Context)
 	assert.NoError(t, err)
-	assert.NotNil(t, nextpinRead)
-	nextpinJson, _ := json.Marshal(&nextpin)
-	nextpinReadJson, _ := json.Marshal(&nextpinRead)
+	assert.Len(t, nextpinRead, 1)
+	nextpinJson, _ := json.Marshal(nextpin)
+	nextpinReadJson, _ := json.Marshal(nextpinRead[0])
 	assert.Equal(t, string(nextpinJson), string(nextpinReadJson))
 
-	// Attempt with wrong ID
 	var nextpinUpdated core.NextPin
 	nextpinUpdated = *nextpin
 	nextpinUpdated.Nonce = 1111111
 	nextpinUpdated.Hash = fftypes.NewRandB32()
-	err = s.UpdateNextPin(context.Background(), nextpin.Sequence, database.NextPinQueryFactory.NewUpdate(ctx).
+	err = s.UpdateNextPin(context.Background(), "ns", nextpin.Sequence, database.NextPinQueryFactory.NewUpdate(ctx).
 		Set("hash", nextpinUpdated.Hash).
 		Set("nonce", nextpinUpdated.Nonce),
 	)
+	nextpinJson, _ = json.Marshal(nextpinUpdated)
 
-	// Check we get the exact same data back - note the removal of one of the nextpin elements
-	nextpinRead, err = s.GetNextPinByHash(ctx, nextpinUpdated.Hash)
+	// Check we get the exact same data back
+	nextpinRead, err = s.GetNextPinsForContext(ctx, "ns", nextpin.Context)
 	assert.NoError(t, err)
-	nextpinJson, _ = json.Marshal(&nextpinUpdated)
-	nextpinReadJson, _ = json.Marshal(&nextpinRead)
+	assert.Len(t, nextpinRead, 1)
+	nextpinReadJson, _ = json.Marshal(nextpinRead[0])
 	assert.Equal(t, string(nextpinJson), string(nextpinReadJson))
 
-	// Query back the nextpin
-	fb := database.NextPinQueryFactory.NewFilter(ctx)
-	filter := fb.And(
-		fb.Eq("context", nextpinUpdated.Context),
-		fb.Eq("hash", nextpinUpdated.Hash),
-		fb.Eq("identity", nextpinUpdated.Identity),
-		fb.Eq("nonce", nextpinUpdated.Nonce),
-	)
-	nextpinRes, res, err := s.GetNextPins(ctx, filter.Count(true))
-	assert.NoError(t, err)
-	assert.Equal(t, 1, len(nextpinRes))
-	assert.Equal(t, int64(1), *res.TotalCount)
-	nextpinReadJson, _ = json.Marshal(nextpinRes[0])
-	assert.Equal(t, string(nextpinJson), string(nextpinReadJson))
-
-	// Test delete
-	err = s.DeleteNextPin(ctx, nextpinUpdated.Sequence)
-	assert.NoError(t, err)
-	nextpins, _, err := s.GetNextPins(ctx, filter)
-	assert.NoError(t, err)
-	assert.Equal(t, 0, len(nextpins))
-
 }
 
 func TestUpsertNextPinFailBegin(t *testing.T) {
@@ -124,52 +103,18 @@ func TestUpsertNextPinFailCommit(t *testing.T) {
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
-func TestGetNextPinByIDSelectFail(t *testing.T) {
-	s, mock := newMockProvider().init()
-	mock.ExpectQuery("SELECT .*").WillReturnError(fmt.Errorf("pop"))
-	_, err := s.GetNextPinByContextAndIdentity(context.Background(), fftypes.NewRandB32(), "0x12345")
-	assert.Regexp(t, "FF10115", err)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
-func TestGetNextPinByIDNotFound(t *testing.T) {
-	s, mock := newMockProvider().init()
-	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{}))
-	msg, err := s.GetNextPinByContextAndIdentity(context.Background(), fftypes.NewRandB32(), "0x12345")
-	assert.NoError(t, err)
-	assert.Nil(t, msg)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
-func TestGetNextPinByIDScanFail(t *testing.T) {
-	s, mock := newMockProvider().init()
-	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{"context"}).AddRow("only one"))
-	_, err := s.GetNextPinByContextAndIdentity(context.Background(), fftypes.NewRandB32(), "0x12345")
-	assert.Regexp(t, "FF10121", err)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
 func TestGetNextPinQueryFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectQuery("SELECT .*").WillReturnError(fmt.Errorf("pop"))
-	f := database.NextPinQueryFactory.NewFilter(context.Background()).Eq("context", "")
-	_, _, err := s.GetNextPins(context.Background(), f)
+	_, err := s.GetNextPinsForContext(context.Background(), "ns", fftypes.NewRandB32())
 	assert.Regexp(t, "FF10115", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
-func TestGetNextPinBuildQueryFail(t *testing.T) {
-	s, _ := newMockProvider().init()
-	f := database.NextPinQueryFactory.NewFilter(context.Background()).Eq("context", map[bool]bool{true: false})
-	_, _, err := s.GetNextPins(context.Background(), f)
-	assert.Regexp(t, "FF00143.*type", err)
-}
-
 func TestGetNextPinReadMessageFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{"context"}).AddRow("only one"))
-	f := database.NextPinQueryFactory.NewFilter(context.Background()).Eq("context", "")
-	_, _, err := s.GetNextPins(context.Background(), f)
+	_, err := s.GetNextPinsForContext(context.Background(), "ns", fftypes.NewRandB32())
 	assert.Regexp(t, "FF10121", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
@@ -178,7 +123,7 @@ func TestNextPinUpdateBeginFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
 	u := database.NextPinQueryFactory.NewUpdate(context.Background()).Set("context", "anything")
-	err := s.UpdateNextPin(context.Background(), 12345, u)
+	err := s.UpdateNextPin(context.Background(), "ns", 12345, u)
 	assert.Regexp(t, "FF10114", err)
 }
 
@@ -186,7 +131,7 @@ func TestNextPinUpdateBuildQueryFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin()
 	u := database.NextPinQueryFactory.NewUpdate(context.Background()).Set("context", map[bool]bool{true: false})
-	err := s.UpdateNextPin(context.Background(), 12345, u)
+	err := s.UpdateNextPin(context.Background(), "ns", 12345, u)
 	assert.Regexp(t, "FF00143.*context", err)
 }
 
@@ -196,22 +141,6 @@ func TestNextPinUpdateFail(t *testing.T) {
 	mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	u := database.NextPinQueryFactory.NewUpdate(context.Background()).Set("context", fftypes.NewRandB32())
-	err := s.UpdateNextPin(context.Background(), 12345, u)
+	err := s.UpdateNextPin(context.Background(), "ns", 12345, u)
 	assert.Regexp(t, "FF10117", err)
 }
-
-func TestNextPinDeleteBeginFail(t *testing.T) {
-	s, mock := newMockProvider().init()
-	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
-	err := s.DeleteNextPin(context.Background(), 12345)
-	assert.Regexp(t, "FF10114", err)
-}
-
-func TestNextPinDeleteFail(t *testing.T) {
-	s, mock := newMockProvider().init()
-	mock.ExpectBegin()
-	mock.ExpectExec("DELETE .*").WillReturnError(fmt.Errorf("pop"))
-	mock.ExpectRollback()
-	err := s.DeleteNextPin(context.Background(), 12345)
-	assert.Regexp(t, "FF10118", err)
-}
diff --git a/internal/database/sqlcommon/pin_sql_test.go b/internal/database/sqlcommon/pin_sql_test.go
index ece43f68b..e291d7bfb 100644
--- a/internal/database/sqlcommon/pin_sql_test.go
+++ b/internal/database/sqlcommon/pin_sql_test.go
@@ -69,7 +69,7 @@ func TestPinsE2EWithDB(t *testing.T) {
 	assert.Equal(t, int64(1), *res.TotalCount)
 
 	// Set it dispatched
-	err = s.UpdatePins(ctx, database.PinQueryFactory.NewFilter(ctx).Eq("sequence", pin.Sequence), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
+	err = s.UpdatePins(ctx, "ns", database.PinQueryFactory.NewFilter(ctx).Eq("sequence", pin.Sequence), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
 	assert.NoError(t, err)
 
 	// Double insert, checking no error and we keep the dispatched flag
@@ -220,7 +220,7 @@ func TestUpdatePinsBeginFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
 	ctx := context.Background()
-	err := s.UpdatePins(ctx, database.PinQueryFactory.NewFilter(ctx).Eq("sequence", 1), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
+	err := s.UpdatePins(ctx, "ns1", database.PinQueryFactory.NewFilter(ctx).Eq("sequence", 1), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
 	assert.Regexp(t, "FF10114", err)
 }
 
@@ -230,7 +230,7 @@ func TestUpdatePinsUpdateFail(t *testing.T) {
 	mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	ctx := context.Background()
-	err := s.UpdatePins(ctx, database.PinQueryFactory.NewFilter(ctx).Eq("sequence", 1), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
+	err := s.UpdatePins(ctx, "ns1", database.PinQueryFactory.NewFilter(ctx).Eq("sequence", 1), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
 	assert.Regexp(t, "FF10117", err)
 }
 
@@ -239,7 +239,7 @@ func TestUpdatePinsBadFilter(t *testing.T) {
 	mock.ExpectBegin()
 	mock.ExpectRollback()
 	ctx := context.Background()
-	err := s.UpdatePins(ctx, database.PinQueryFactory.NewFilter(ctx).Eq("sequence", 1), database.PinQueryFactory.NewUpdate(ctx).Set("bad", true))
+	err := s.UpdatePins(ctx, "ns1", database.PinQueryFactory.NewFilter(ctx).Eq("sequence", 1), database.PinQueryFactory.NewUpdate(ctx).Set("bad", true))
 	assert.Regexp(t, "FF00142", err)
 }
 
@@ -248,6 +248,6 @@ func TestUpdatePinsBadUpdate(t *testing.T) {
 	mock.ExpectBegin()
 	mock.ExpectRollback()
 	ctx := context.Background()
-	err := s.UpdatePins(ctx, database.PinQueryFactory.NewFilter(ctx).Eq("bad", 1), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
+	err := s.UpdatePins(ctx, "ns1", database.PinQueryFactory.NewFilter(ctx).Eq("bad", 1), database.PinQueryFactory.NewUpdate(ctx).Set("dispatched", true))
 	assert.Regexp(t, "FF00142", err)
 }
diff --git a/internal/database/sqlcommon/transaction_sql_test.go b/internal/database/sqlcommon/transaction_sql_test.go
index dd0e5ec84..2a3207a8f 100644
--- a/internal/database/sqlcommon/transaction_sql_test.go
+++ b/internal/database/sqlcommon/transaction_sql_test.go
@@ -84,7 +84,7 @@ func TestTransactionE2EWithDB(t *testing.T) {
 	// Update
 	up := database.TransactionQueryFactory.NewUpdate(ctx).
 		Set("blockchainids", core.FFStringArray{"0x12345", "0x23456"})
-	err = s.UpdateTransaction(ctx, transaction.ID, up)
+	err = s.UpdateTransaction(ctx, "ns1", transaction.ID, up)
 	assert.NoError(t, err)
 
 	// Test find updated value
@@ -184,7 +184,7 @@ func TestTransactionUpdateBeginFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
 	u := database.TransactionQueryFactory.NewUpdate(context.Background()).Set("id", "anything")
-	err := s.UpdateTransaction(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateTransaction(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10114", err)
 }
 
@@ -192,7 +192,7 @@ func TestTransactionUpdateBuildQueryFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin()
 	u := database.TransactionQueryFactory.NewUpdate(context.Background()).Set("id", map[bool]bool{true: false})
-	err := s.UpdateTransaction(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateTransaction(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF00143.*id", err)
 }
 
@@ -202,6 +202,6 @@ func TestTransactionUpdateFail(t *testing.T) {
 	mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	u := database.TransactionQueryFactory.NewUpdate(context.Background()).Set("id", fftypes.NewUUID())
-	err := s.UpdateTransaction(context.Background(), fftypes.NewUUID(), u)
+	err := s.UpdateTransaction(context.Background(), "ns1", fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10117", err)
 }
diff --git a/internal/events/aggregator_batch_state_test.go b/internal/events/aggregator_batch_state_test.go
index 7735e55ed..4391092aa 100644
--- a/internal/events/aggregator_batch_state_test.go
+++ b/internal/events/aggregator_batch_state_test.go
@@ -34,7 +34,7 @@ func TestFlushPinsFailUpdatePins(t *testing.T) {
 	bs := newBatchState(ag)
 
 	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("UpdatePins", ag.ctx, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	mdi.On("UpdatePins", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 
 	bs.markMessageDispatched(fftypes.NewUUID(), &core.Message{
 		Header: core.MessageHeader{
@@ -55,8 +55,8 @@ func TestFlushPinsFailUpdateMessages(t *testing.T) {
 	msgID := fftypes.NewUUID()
 
 	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("UpdatePins", ag.ctx, mock.Anything, mock.Anything).Return(nil)
-	mdi.On("UpdateMessages", ag.ctx, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	mdi.On("UpdatePins", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdateMessages", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 	mdm := ag.data.(*datamocks.Manager)
 	mdm.On("UpdateMessageStateIfCached", ag.ctx, msgID, core.MessageStateConfirmed, mock.Anything).Return()
 
diff --git a/internal/events/aggregator_test.go b/internal/events/aggregator_test.go
index 886185c72..970e22ad9 100644
--- a/internal/events/aggregator_test.go
+++ b/internal/events/aggregator_test.go
@@ -193,7 +193,7 @@ func TestAggregationMaskedZeroNonceMatch(t *testing.T) {
 	// Get the batch
 	mdi.On("GetBatchByID", ag.ctx, "ns1", batchID).Return(bp, nil)
 	// Look for existing nextpins - none found, first on context
-	mdi.On("GetNextPins", ag.ctx, mock.Anything).Return([]*core.NextPin{}, nil, nil).Once()
+	mdi.On("GetNextPinsForContext", ag.ctx, "ns1", contextUnmasked).Return([]*core.NextPin{}, nil).Once()
 	// Get the group members
 	mpm.On("ResolveInitGroup", ag.ctx, mock.Anything).Return(&core.Group{
 		GroupIdentity: core.GroupIdentity{
@@ -225,9 +225,9 @@ func TestAggregationMaskedZeroNonceMatch(t *testing.T) {
 		return *e.Reference == *msgID && e.Type == core.EventTypeMessageConfirmed
 	})).Return(nil)
 	// Set the pin to dispatched
-	mdi.On("UpdatePins", ag.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdatePins", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
 	// Update the message
-	mdi.On("UpdateMessages", ag.ctx, mock.Anything, mock.MatchedBy(func(u database.Update) bool {
+	mdi.On("UpdateMessages", ag.ctx, "ns1", mock.Anything, mock.MatchedBy(func(u database.Update) bool {
 		update, err := u.Finalize()
 		assert.NoError(t, err)
 		assert.Len(t, update.SetOperations, 2)
@@ -339,10 +339,10 @@ func TestAggregationMaskedNextSequenceMatch(t *testing.T) {
 	// Get the batch
 	mdi.On("GetBatchByID", ag.ctx, "ns1", batchID).Return(bp, nil)
 	// Look for existing nextpins - none found, first on context
-	mdi.On("GetNextPins", ag.ctx, mock.Anything).Return([]*core.NextPin{
+	mdi.On("GetNextPinsForContext", ag.ctx, "ns1", contextUnmasked).Return([]*core.NextPin{
 		{Context: contextUnmasked, Identity: member1org.DID, Hash: member1Nonce100, Nonce: 100, Sequence: 929},
 		{Context: contextUnmasked, Identity: member2org.DID, Hash: member2Nonce500, Nonce: 500, Sequence: 424},
-	}, nil, nil).Once()
+	}, nil).Once()
 	// Validate the message is ok
 	mdm.On("GetMessageWithDataCached", ag.ctx, batch.Payload.Messages[0].Header.ID, data.CRORequirePins).Return(batch.Payload.Messages[0], core.DataArray{}, true, nil)
 	mdm.On("ValidateAll", ag.ctx, mock.Anything).Return(true, nil)
@@ -352,7 +352,7 @@ func TestAggregationMaskedNextSequenceMatch(t *testing.T) {
 		return *e.Reference == *msgID && e.Type == core.EventTypeMessageConfirmed
 	})).Return(nil)
 	// Update member2 to nonce 1
-	mdi.On("UpdateNextPin", ag.ctx, mock.MatchedBy(func(seq int64) bool {
+	mdi.On("UpdateNextPin", ag.ctx, "ns1", mock.MatchedBy(func(seq int64) bool {
 		return seq == 424
 	}), mock.MatchedBy(func(update database.Update) bool {
 		ui, _ := update.Finalize()
@@ -365,9 +365,9 @@ func TestAggregationMaskedNextSequenceMatch(t *testing.T) {
 		return true
 	})).Return(nil)
 	// Set the pin to dispatched
-	mdi.On("UpdatePins", ag.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdatePins", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
 	// Update the message
-	mdi.On("UpdateMessages", ag.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdateMessages", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
 
 	_, err := ag.processPinsEventsHandler([]core.LocallySequenced{
 		&core.Pin{
@@ -452,9 +452,9 @@ func TestAggregationBroadcast(t *testing.T) {
 		return *e.Reference == *msgID && e.Type == core.EventTypeMessageConfirmed
 	})).Return(nil)
 	// Set the pin to dispatched
-	mdi.On("UpdatePins", ag.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdatePins", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
 	// Update the message
-	mdi.On("UpdateMessages", ag.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdateMessages", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
 
 	err := ag.processPins(ag.ctx, []*core.Pin{
 		{
@@ -547,9 +547,9 @@ func TestAggregationMigratedBroadcast(t *testing.T) {
 		return *e.Reference == *msgID && e.Type == core.EventTypeMessageConfirmed
 	})).Return(nil)
 	// Set the pin to dispatched
-	mdi.On("UpdatePins", ag.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdatePins", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
 	// Update the message
-	mdi.On("UpdateMessages", ag.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdateMessages", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
 
 	err = ag.processPins(ag.ctx, []*core.Pin{
 		{
@@ -999,7 +999,7 @@ func TestProcessMsgFailGetNextPins(t *testing.T) {
 	defer cancel()
 
 	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("GetNextPins", ag.ctx, mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
+	mdi.On("GetNextPinsForContext", ag.ctx, "ns1", mock.Anything).Return(nil, fmt.Errorf("pop"))
 
 	msg := &core.Message{
 		Header: core.MessageHeader{
@@ -1094,14 +1094,14 @@ func TestProcessMsgFailPinUpdate(t *testing.T) {
 		Type:  core.VerifierTypeEthAddress,
 		Value: "0x12345",
 	}).Return(org1, nil)
-	mdi.On("GetNextPins", ag.ctx, mock.Anything).Return([]*core.NextPin{
+	mdi.On("GetNextPinsForContext", ag.ctx, "ns1", mock.Anything).Return([]*core.NextPin{
 		{Context: fftypes.NewRandB32(), Hash: pin, Identity: org1.DID},
-	}, nil, nil)
+	}, nil)
 	mdm.On("GetMessageWithDataCached", ag.ctx, mock.Anything, data.CRORequirePins).Return(msg, nil, true, nil)
 	mdm.On("ValidateAll", ag.ctx, mock.Anything).Return(false, nil)
 	mdi.On("InsertEvent", ag.ctx, mock.Anything).Return(nil)
-	mdi.On("UpdateMessages", ag.ctx, mock.Anything, mock.Anything).Return(nil)
-	mdi.On("UpdateNextPin", ag.ctx, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	mdi.On("UpdateMessages", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdateNextPin", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 
 	err := ag.processMessage(ag.ctx, &core.BatchManifest{
 		ID: fftypes.NewUUID(),
@@ -1125,9 +1125,9 @@ func TestCheckMaskedContextReadyMismatchedAuthor(t *testing.T) {
 	pin := fftypes.NewRandB32()
 
 	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("GetNextPins", ag.ctx, mock.Anything).Return([]*core.NextPin{
+	mdi.On("GetNextPinsForContext", ag.ctx, "ns1", mock.Anything).Return([]*core.NextPin{
 		{Context: fftypes.NewRandB32(), Hash: pin},
-	}, nil, nil)
+	}, nil)
 
 	bs := newBatchState(ag)
 	_, err := bs.checkMaskedContextReady(ag.ctx, &core.Message{
@@ -1524,7 +1524,7 @@ func TestDefinitionBroadcastActionRejectCustomCorrelator(t *testing.T) {
 	mdm.On("GetMessageData", ag.ctx, mock.Anything, true).Return(core.DataArray{}, true, nil)
 
 	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("UpdateMessages", ag.ctx, mock.Anything, mock.MatchedBy(func(u database.Update) bool {
+	mdi.On("UpdateMessages", ag.ctx, "ns1", mock.Anything, mock.MatchedBy(func(u database.Update) bool {
 		update, err := u.Finalize()
 		assert.NoError(t, err)
 		assert.Len(t, update.SetOperations, 2)
@@ -1577,7 +1577,7 @@ func TestDefinitionBroadcastInvalidSigner(t *testing.T) {
 	mdm.On("GetMessageData", ag.ctx, mock.Anything, true).Return(core.DataArray{}, true, nil)
 
 	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("UpdateMessages", ag.ctx, mock.Anything, mock.MatchedBy(func(u database.Update) bool {
+	mdi.On("UpdateMessages", ag.ctx, "ns1", mock.Anything, mock.MatchedBy(func(u database.Update) bool {
 		update, err := u.Finalize()
 		assert.NoError(t, err)
 		assert.Len(t, update.SetOperations, 2)
@@ -1662,9 +1662,9 @@ func TestDispatchPrivateQueuesLaterDispatch(t *testing.T) {
 	context := fftypes.HashResult(h)
 
 	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("GetNextPins", ag.ctx, mock.Anything).Return([]*core.NextPin{
+	mdi.On("GetNextPinsForContext", ag.ctx, "ns1", mock.Anything).Return([]*core.NextPin{
 		{Context: context, Nonce: 1 /* match member1NonceOne */, Identity: org1.DID, Hash: member1NonceOne},
-	}, nil, nil)
+	}, nil)
 
 	msg1.Pins = core.FFStringArray{member1NonceOne.String()}
 	msg2.Pins = core.FFStringArray{member1NonceTwo.String()}
@@ -1705,9 +1705,9 @@ func TestDispatchPrivateNextPinIncremented(t *testing.T) {
 	context := fftypes.HashResult(h)
 
 	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("GetNextPins", ag.ctx, mock.Anything).Return([]*core.NextPin{
+	mdi.On("GetNextPinsForContext", ag.ctx, "ns1", mock.Anything).Return([]*core.NextPin{
 		{Context: context, Nonce: 1 /* match member1NonceOne */, Identity: org1.DID, Hash: member1NonceOne},
-	}, nil, nil)
+	}, nil)
 
 	msg1.Pins = core.FFStringArray{member1NonceOne.String()}
 	msg2.Pins = core.FFStringArray{member1NonceTwo.String()}
@@ -1906,7 +1906,7 @@ func TestRewindOffchainBatchesNoBatches(t *testing.T) {
 	defer cancel()
 
 	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("UpdateMessages", ag.ctx, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	mdi.On("UpdateMessages", ag.ctx, "ns1", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 
 	rewind, offset := ag.rewindOffchainBatches()
 	assert.False(t, rewind)
diff --git a/internal/events/dx_callbacks_test.go b/internal/events/dx_callbacks_test.go
index ad45bbd46..090e74251 100644
--- a/internal/events/dx_callbacks_test.go
+++ b/internal/events/dx_callbacks_test.go
@@ -656,7 +656,7 @@ func TestMessageReceiveUnpinnedBatchOk(t *testing.T) {
 	mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
 		args[2].(database.PostCompletionHook)()
 	})
-	mdi.On("UpdateMessages", em.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdateMessages", em.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertEvent", em.ctx, mock.Anything).Return(nil)
 	mdm := em.data.(*datamocks.Manager)
 	mdm.On("UpdateMessageCache", mock.Anything, mock.Anything).Return()
@@ -697,7 +697,7 @@ func TestMessageReceiveUnpinnedBatchConfirmMessagesFail(t *testing.T) {
 	mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
 		args[2].(database.PostCompletionHook)()
 	})
-	mdi.On("UpdateMessages", em.ctx, mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	mdi.On("UpdateMessages", em.ctx, "ns1", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
 	mdm := em.data.(*datamocks.Manager)
 	mdm.On("UpdateMessageCache", mock.Anything, mock.Anything).Return()
 
@@ -738,7 +738,7 @@ func TestMessageReceiveUnpinnedBatchPersistEventFail(t *testing.T) {
 	mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
 		args[2].(database.PostCompletionHook)()
 	})
-	mdi.On("UpdateMessages", em.ctx, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("UpdateMessages", em.ctx, "ns1", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertEvent", em.ctx, mock.Anything).Return(fmt.Errorf("pop"))
 	mdm := em.data.(*datamocks.Manager)
 	mdm.On("UpdateMessageCache", mock.Anything, mock.Anything).Return()
diff --git a/internal/operations/operation_updater_test.go b/internal/operations/operation_updater_test.go
index a271ecb8e..3b9549057 100644
--- a/internal/operations/operation_updater_test.go
+++ b/internal/operations/operation_updater_test.go
@@ -166,7 +166,7 @@ func TestSubmitUpdateWorkerE2ESuccess(t *testing.T) {
 	}, nil, nil)
 	mdi.On("GetTransactions", mock.Anything, mock.Anything, mock.Anything).Return([]*core.Transaction{tx1}, nil, nil)
 	mdi.On("ResolveOperation", mock.Anything, "ns1", opID1, core.OpStatusSucceeded, mock.Anything, fftypes.JSONObject(nil)).Return(nil)
-	mdi.On("UpdateTransaction", mock.Anything, tx1.ID, mock.Anything).Return(nil)
+	mdi.On("UpdateTransaction", mock.Anything, "ns1", tx1.ID, mock.Anything).Return(nil)
 	err1Str := "err1"
 	mdi.On("ResolveOperation", mock.Anything, "ns1", opID2, core.OpStatusFailed, &err1Str, fftypes.JSONObject{"test": true}).Return(nil)
 	err2Str := "err2"
@@ -312,7 +312,7 @@ func TestDoUpdateFailTransactionUpdate(t *testing.T) {
 	opID1 := fftypes.NewUUID()
 	txID1 := fftypes.NewUUID()
 	mdi := ou.database.(*databasemocks.Plugin)
-	mdi.On("UpdateTransaction", mock.Anything, txID1, mock.Anything).Return(fmt.Errorf("pop"))
+	mdi.On("UpdateTransaction", mock.Anything, "ns1", txID1, mock.Anything).Return(fmt.Errorf("pop"))
 
 	ou.initQueues()
 
diff --git a/internal/privatemessaging/message_test.go b/internal/privatemessaging/message_test.go
index cafe5d2ea..3f0294626 100644
--- a/internal/privatemessaging/message_test.go
+++ b/internal/privatemessaging/message_test.go
@@ -640,7 +640,7 @@ func TestDispatchedUnpinnedMessageOK(t *testing.T) {
 	node2 := newTestNode("node2", newTestOrg("remoteorg"))
 
 	mim := pm.identity.(*identitymanagermocks.Manager)
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(node1, nil)
 
 	mdx := pm.exchange.(*dataexchangemocks.Plugin)
 	mdx.On("SendMessage", pm.ctx, mock.Anything, "node2-peer", mock.Anything).Return(nil)
@@ -700,6 +700,7 @@ func TestSendDataTransferBlobsFail(t *testing.T) {
 
 	localOrg := newTestOrg("localorg")
 	groupID := fftypes.NewRandB32()
+	node1 := newTestNode("node1", localOrg)
 	node2 := newTestNode("node2", newTestOrg("remoteorg"))
 	nodes := []*core.Identity{node2}
 
@@ -709,6 +710,7 @@ func TestSendDataTransferBlobsFail(t *testing.T) {
 		return true
 	})).Return(nil)
 	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(node1, nil)
 
 	mdi := pm.database.(*databasemocks.Plugin)
 	mdi.On("GetBlobMatchingHash", pm.ctx, mock.Anything).Return(nil, fmt.Errorf("pop"))
@@ -753,11 +755,12 @@ func TestSendDataTransferFail(t *testing.T) {
 
 	localOrg := newTestOrg("localorg")
 	groupID := fftypes.NewRandB32()
+	node1 := newTestNode("node1", localOrg)
 	node2 := newTestNode("node2", newTestOrg("remoteorg"))
 	nodes := []*core.Identity{node2}
 
 	mim := pm.identity.(*identitymanagermocks.Manager)
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(node1, nil)
 
 	mom := pm.operations.(*operationmocks.Manager)
 	mom.On("AddOrReuseOperation", pm.ctx, mock.Anything).Return(nil)
@@ -802,6 +805,7 @@ func TestSendDataTransferInsertOperationFail(t *testing.T) {
 
 	localOrg := newTestOrg("localorg")
 	groupID := fftypes.NewRandB32()
+	node1 := newTestNode("node1", localOrg)
 	node2 := newTestNode("node2", newTestOrg("remoteorg"))
 	nodes := []*core.Identity{node2}
 
@@ -811,6 +815,7 @@ func TestSendDataTransferInsertOperationFail(t *testing.T) {
 		return true
 	})).Return(nil)
 	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(node1, nil)
 
 	mom := pm.operations.(*operationmocks.Manager)
 	mom.On("AddOrReuseOperation", pm.ctx, mock.Anything).Return(fmt.Errorf("pop"))
diff --git a/internal/privatemessaging/privatemessaging_test.go b/internal/privatemessaging/privatemessaging_test.go
index 8b1786475..4a7c9e07d 100644
--- a/internal/privatemessaging/privatemessaging_test.go
+++ b/internal/privatemessaging/privatemessaging_test.go
@@ -135,7 +135,7 @@ func TestDispatchBatchWithBlobs(t *testing.T) {
 	mim := pm.identity.(*identitymanagermocks.Manager)
 	mom := pm.operations.(*operationmocks.Manager)
 
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(node1, nil)
 	mdi.On("GetGroupByHash", pm.ctx, "ns1", groupID).Return(&core.Group{
 		Hash: fftypes.NewRandB32(),
 		GroupIdentity: core.GroupIdentity{
@@ -271,7 +271,7 @@ func TestSendAndSubmitBatchUnregisteredNode(t *testing.T) {
 		},
 	}, nil)
 
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(nil, fmt.Errorf("pop"))
+	mim.On("GetLocalNode", pm.ctx).Return(nil, fmt.Errorf("pop"))
 
 	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
 		Persisted: core.BatchPersisted{
@@ -321,7 +321,7 @@ func TestSendSubmitInsertOperationFail(t *testing.T) {
 	node2 := newTestNode("node2", newTestOrg("remoteorg"))
 
 	mim := pm.identity.(*identitymanagermocks.Manager)
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(node1, nil)
 
 	mdi := pm.database.(*databasemocks.Plugin)
 	mim.On("CachedIdentityLookupByID", pm.ctx, node1.ID).Return(node1, nil).Once()
@@ -367,7 +367,7 @@ func TestSendSubmitBlobTransferFail(t *testing.T) {
 	blob1 := fftypes.NewRandB32()
 
 	mim := pm.identity.(*identitymanagermocks.Manager)
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(node1, nil)
 
 	mdi := pm.database.(*databasemocks.Plugin)
 	mim.On("CachedIdentityLookupByID", pm.ctx, node1.ID).Return(node1, nil).Once()
@@ -427,7 +427,7 @@ func TestWriteTransactionSubmitBatchPinFail(t *testing.T) {
 	blob1 := fftypes.NewRandB32()
 
 	mim := pm.identity.(*identitymanagermocks.Manager)
-	mim.On("GetMultipartyRootOrg", pm.ctx).Return(localOrg, nil)
+	mim.On("GetLocalNode", pm.ctx).Return(node1, nil)
 
 	mdi := pm.database.(*databasemocks.Plugin)
 	mim.On("CachedIdentityLookupByID", pm.ctx, node1.ID).Return(node1, nil).Once()
diff --git a/internal/txcommon/txcommon_test.go b/internal/txcommon/txcommon_test.go
index 642a78502..1b10851f2 100644
--- a/internal/txcommon/txcommon_test.go
+++ b/internal/txcommon/txcommon_test.go
@@ -174,7 +174,7 @@ func TestPersistTransactionExistingAddBlockchainID(t *testing.T) {
 		Created:       fftypes.Now(),
 		BlockchainIDs: core.FFStringArray{"0x111111"},
 	}, nil)
-	mdi.On("UpdateTransaction", ctx, txid, mock.Anything).Return(nil)
+	mdi.On("UpdateTransaction", ctx, "ns1", txid, mock.Anything).Return(nil)
 
 	valid, err := txHelper.PersistTransaction(ctx, txid, core.TransactionTypeBatchPin, "0x222222")
 	assert.NoError(t, err)
@@ -199,7 +199,7 @@ func TestPersistTransactionExistingUpdateFail(t *testing.T) {
 		Created:       fftypes.Now(),
 		BlockchainIDs: core.FFStringArray{"0x111111"},
 	}, nil)
-	mdi.On("UpdateTransaction", ctx, txid, mock.Anything).Return(fmt.Errorf("pop"))
+	mdi.On("UpdateTransaction", ctx, "ns1", txid, mock.Anything).Return(fmt.Errorf("pop"))
 
 	valid, err := txHelper.PersistTransaction(ctx, txid, core.TransactionTypeBatchPin, "0x222222")
 	assert.Regexp(t, "pop", err)
@@ -313,7 +313,7 @@ func TestAddBlockchainTX(t *testing.T) {
 		Created:       fftypes.Now(),
 		BlockchainIDs: core.FFStringArray{"0x111111"},
 	}
-	mdi.On("UpdateTransaction", ctx, tx.ID, mock.MatchedBy(func(u database.Update) bool {
+	mdi.On("UpdateTransaction", ctx, "ns1", tx.ID, mock.MatchedBy(func(u database.Update) bool {
 		info, _ := u.Finalize()
 		assert.Equal(t, 1, len(info.SetOperations))
 		assert.Equal(t, "blockchainids", info.SetOperations[0].Field)
@@ -343,7 +343,7 @@ func TestAddBlockchainTXUpdateFail(t *testing.T) {
 		Created:       fftypes.Now(),
 		BlockchainIDs: core.FFStringArray{"0x111111"},
 	}
-	mdi.On("UpdateTransaction", ctx, tx.ID, mock.Anything).Return(fmt.Errorf("pop"))
+	mdi.On("UpdateTransaction", ctx, "ns1", tx.ID, mock.Anything).Return(fmt.Errorf("pop"))
 
 	err := txHelper.AddBlockchainTX(ctx, tx, "abc")
 	assert.EqualError(t, err, "pop")
diff --git a/test/e2e/client/restclient.go b/test/e2e/client/restclient.go
index 6d34bc7c8..6c3d051d0 100644
--- a/test/e2e/client/restclient.go
+++ b/test/e2e/client/restclient.go
@@ -161,7 +161,7 @@ func (client *FireFlyClient) GetMessageEvents(t *testing.T, startTime time.Time,
 	return events
 }
 
-func (client *FireFlyClient) GetMessages(t *testing.T, startTime time.Time, msgType core.MessageType, topic string, expectedStatus int) (msgs []*core.Message) {
+func (client *FireFlyClient) GetMessages(t *testing.T, startTime time.Time, msgType core.MessageType, topic string) (msgs []*core.Message) {
 	path := client.namespaced(urlGetMessages)
 	resp, err := client.Client.R().
 		SetQueryParam("type", string(msgType)).
@@ -171,21 +171,32 @@ func (client *FireFlyClient) GetMessages(t *testing.T, startTime time.Time, msgT
 		SetResult(&msgs).
 		Get(path)
 	require.NoError(t, err)
-	require.Equal(t, expectedStatus, resp.StatusCode(), "GET %s [%d]: %s", path, resp.StatusCode(), resp.String())
+	require.Equal(t, 200, resp.StatusCode(), "GET %s [%d]: %s", path, resp.StatusCode(), resp.String())
 	return msgs
 }
 
-func (client *FireFlyClient) GetData(t *testing.T, startTime time.Time, expectedStatus int) (data core.DataArray) {
+func (client *FireFlyClient) GetData(t *testing.T, startTime time.Time) (data core.DataArray) {
 	path := client.namespaced(urlGetData)
 	resp, err := client.Client.R().
 		SetQueryParam("created", fmt.Sprintf(">%d", startTime.UnixNano())).
 		SetResult(&data).
 		Get(path)
 	require.NoError(t, err)
-	require.Equal(t, expectedStatus, resp.StatusCode(), "GET %s [%d]: %s", path, resp.StatusCode(), resp.String())
+	require.Equal(t, 200, resp.StatusCode(), "GET %s [%d]: %s", path, resp.StatusCode(), resp.String())
 	return data
 }
 
+func (client *FireFlyClient) GetDataByID(t *testing.T, id *fftypes.UUID) *core.Data {
+	var data core.Data
+	path := client.namespaced(urlGetData + "/" + id.String())
+	resp, err := client.Client.R().
+		SetResult(&data).
+		Get(path)
+	require.NoError(t, err)
+	require.Equal(t, 200, resp.StatusCode(), "GET %s [%d]: %s", path, resp.StatusCode(), resp.String())
+	return &data
+}
+
 func (client *FireFlyClient) GetDataForMessage(t *testing.T, startTime time.Time, msgID *fftypes.UUID) (data core.DataArray) {
 	path := client.namespaced(urlGetMessages)
 	path += "/" + msgID.String() + "/data"
@@ -429,15 +440,8 @@ func (client *FireFlyClient) BroadcastBlobMessage(t *testing.T, topic string) (*
 	return data, res, err
 }
 
-func (client *FireFlyClient) PrivateBlobMessageDatatypeTagged(t *testing.T, topic string, orgNames []string, startTime time.Time) (*core.Data, *resty.Response, error) {
+func (client *FireFlyClient) PrivateBlobMessageDatatypeTagged(t *testing.T, topic string, members []core.MemberInput, startTime time.Time) (*core.Data, *resty.Response, error) {
 	data := client.CreateBlob(t, &core.DatatypeRef{Name: "myblob"})
-	members := make([]core.MemberInput, len(orgNames))
-	for i, oName := range orgNames {
-		// We let FireFly resolve the friendly name of the org to the identity
-		members[i] = core.MemberInput{
-			Identity: oName,
-		}
-	}
 	res, err := client.Client.R().
 		SetBody(core.MessageInOut{
 			Message: core.Message{
@@ -457,18 +461,11 @@ func (client *FireFlyClient) PrivateBlobMessageDatatypeTagged(t *testing.T, topi
 	return data, res, err
 }
 
-func (client *FireFlyClient) PrivateMessage(topic string, data *core.DataRefOrValue, orgNames []string, tag string, txType core.TransactionType, confirm bool, startTime time.Time) (*resty.Response, error) {
-	return client.PrivateMessageWithKey("", topic, data, orgNames, tag, txType, confirm, startTime)
+func (client *FireFlyClient) PrivateMessage(topic string, data *core.DataRefOrValue, members []core.MemberInput, tag string, txType core.TransactionType, confirm bool, startTime time.Time) (*resty.Response, error) {
+	return client.PrivateMessageWithKey("", topic, data, members, tag, txType, confirm, startTime)
 }
 
-func (client *FireFlyClient) PrivateMessageWithKey(key, topic string, data *core.DataRefOrValue, orgNames []string, tag string, txType core.TransactionType, confirm bool, startTime time.Time) (*resty.Response, error) {
-	members := make([]core.MemberInput, len(orgNames))
-	for i, oName := range orgNames {
-		// We let FireFly resolve the friendly name of the org to the identity
-		members[i] = core.MemberInput{
-			Identity: oName,
-		}
-	}
+func (client *FireFlyClient) PrivateMessageWithKey(key, topic string, data *core.DataRefOrValue, members []core.MemberInput, tag string, txType core.TransactionType, confirm bool, startTime time.Time) (*resty.Response, error) {
 	msg := core.MessageInOut{
 		Message: core.Message{
 			Header: core.MessageHeader{
diff --git a/test/e2e/e2e.go b/test/e2e/e2e.go
index 93cb06d44..1c4ec688a 100644
--- a/test/e2e/e2e.go
+++ b/test/e2e/e2e.go
@@ -61,7 +61,7 @@ var WidgetSchemaJSON = []byte(`{
 func PollForUp(t *testing.T, client *client.FireFlyClient) {
 	var resp *resty.Response
 	var err error
-	for i := 0; i < 3; i++ {
+	for i := 0; i < 5; i++ {
 		_, resp, err = client.GetStatus()
 		if err == nil && resp.StatusCode() == 200 {
 			break
diff --git a/test/e2e/multiparty/identity.go b/test/e2e/multiparty/identity.go
index 3e7efdee9..f76aeeac0 100644
--- a/test/e2e/multiparty/identity.go
+++ b/test/e2e/multiparty/identity.go
@@ -127,9 +127,14 @@ func (suite *IdentityTestSuite) TestCustomChildIdentityPrivate() {
 		e2e.WaitForIdentityConfirmed(suite.T(), received2)
 	}
 
+	members := []core.MemberInput{
+		{Identity: custom1.DID},
+		{Identity: custom2.DID},
+	}
+
 	resp, err := suite.testState.client1.PrivateMessageWithKey(org1key, "topic1", &core.DataRefOrValue{
 		Value: fftypes.JSONAnyPtr(`"test private custom identity"`),
-	}, []string{custom1.DID, custom2.DID}, "tag1", core.TransactionTypeBatchPin, true, suite.testState.startTime)
+	}, members, "tag1", core.TransactionTypeBatchPin, true, suite.testState.startTime)
 	require.NoError(suite.T(), err)
 	assert.Equal(suite.T(), 200, resp.StatusCode())
 
diff --git a/test/e2e/multiparty/multi_tenancy.go b/test/e2e/multiparty/multi_tenancy.go
new file mode 100644
index 000000000..29c0243e9
--- /dev/null
+++ b/test/e2e/multiparty/multi_tenancy.go
@@ -0,0 +1,206 @@
+// 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 multiparty
+
+import (
+	"fmt"
+
+	"github.com/hyperledger/firefly-common/pkg/fftypes"
+	"github.com/hyperledger/firefly/pkg/core"
+	"github.com/hyperledger/firefly/test/e2e"
+	"github.com/hyperledger/firefly/test/e2e/client"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/stretchr/testify/suite"
+)
+
+type NamespaceAliasSuite struct {
+	suite.Suite
+	testState *testState
+}
+
+func (suite *NamespaceAliasSuite) SetupSuite() {
+	suite.testState = beforeE2ETest(suite.T())
+}
+
+func (suite *NamespaceAliasSuite) BeforeTest(suiteName, testName string) {
+	suite.testState = beforeE2ETest(suite.T())
+}
+
+func (suite *NamespaceAliasSuite) AfterTest(suiteName, testName string) {
+	e2e.VerifyAllOperationsSucceeded(suite.T(), []*client.FireFlyClient{suite.testState.client1, suite.testState.client2}, suite.testState.startTime)
+}
+
+func newNamespace(remoteName, contractAddress string) (ns, org, node map[string]interface{}) {
+	org = make(map[string]interface{})
+	node = make(map[string]interface{})
+	ns = map[string]interface{}{
+		"remotename": remoteName,
+		"multiparty": map[string]interface{}{
+			"enabled": true,
+			"org":     org,
+			"node":    node,
+			"contract": []map[string]interface{}{
+				{
+					"location":   map[string]interface{}{"address": contractAddress},
+					"firstevent": "0",
+				},
+			},
+		},
+	}
+	return ns, org, node
+}
+
+func (suite *NamespaceAliasSuite) TestMultiTenancy() {
+	defer suite.testState.Done()
+
+	address := deployContract(suite.T(), suite.testState.stackName, "firefly/Firefly.json")
+	namespace := e2e.RandomName(suite.T())
+	nsAlice := namespace + "-A"
+	nsBob := namespace + "-B"
+	nsCharlie := namespace + "-C"
+	suite.T().Logf("Test namespace: %s", namespace)
+
+	data := &core.DataRefOrValue{Value: fftypes.JSONAnyPtr(`"test"`)}
+
+	// Add 3 local namespaces that all map to the same remote namespace
+	// (2 on the first node, 1 on the second)
+	data1 := e2e.ReadConfig(suite.T(), suite.testState.configFile1)
+	ns, org, node := newNamespace(namespace, address)
+	ns["name"] = nsAlice
+	org["name"] = suite.testState.org1.Name
+	org["key"] = suite.testState.org1key.Value
+	node["name"] = "alice"
+	e2e.AddNamespace(data1, ns)
+	ns, org, node = newNamespace(namespace, address)
+	ns["name"] = nsBob
+	org["name"] = suite.testState.org1.Name
+	org["key"] = suite.testState.org1key.Value
+	node["name"] = "bob"
+	e2e.AddNamespace(data1, ns)
+	e2e.WriteConfig(suite.T(), suite.testState.configFile1, data1)
+
+	data2 := e2e.ReadConfig(suite.T(), suite.testState.configFile2)
+	ns, org, node = newNamespace(namespace, address)
+	ns["name"] = nsCharlie
+	org["name"] = suite.testState.org2.Name
+	org["key"] = suite.testState.org2key.Value
+	node["name"] = "charlie"
+	e2e.AddNamespace(data2, ns)
+	e2e.WriteConfig(suite.T(), suite.testState.configFile2, data2)
+
+	admin1 := client.NewResty(suite.T())
+	admin2 := client.NewResty(suite.T())
+	admin1.SetBaseURL(suite.testState.adminHost1 + "/spi/v1")
+	admin2.SetBaseURL(suite.testState.adminHost2 + "/spi/v1")
+
+	// Reset both nodes to pick up the new namespace
+	e2e.ResetFireFly(suite.T(), admin1)
+	e2e.ResetFireFly(suite.T(), admin2)
+	e2e.PollForUp(suite.T(), suite.testState.client1)
+	e2e.PollForUp(suite.T(), suite.testState.client2)
+
+	clientAlice := client.NewFireFly(suite.T(), suite.testState.client1.Hostname, nsAlice)
+	clientBob := client.NewFireFly(suite.T(), suite.testState.client1.Hostname, nsBob)
+	clientCharlie := client.NewFireFly(suite.T(), suite.testState.client2.Hostname, nsCharlie)
+
+	eventNames := "message_confirmed"
+	queryString := fmt.Sprintf("namespace=%s&ephemeral&autoack&filter.events=%s", nsAlice, eventNames)
+	receivedAlice := e2e.WsReader(clientAlice.WebSocket(suite.T(), queryString, nil))
+	queryString = fmt.Sprintf("namespace=%s&ephemeral&autoack&filter.events=%s", nsBob, eventNames)
+	receivedBob := e2e.WsReader(clientBob.WebSocket(suite.T(), queryString, nil))
+	queryString = fmt.Sprintf("namespace=%s&ephemeral&autoack&filter.events=%s", nsCharlie, eventNames)
+	receivedCharlie := e2e.WsReader(clientCharlie.WebSocket(suite.T(), queryString, nil))
+
+	// Register org/node identities on the new namespace
+	clientAlice.RegisterSelfOrg(suite.T(), true)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedBob, core.MessageTypeDefinition)
+	clientAlice.RegisterSelfNode(suite.T(), true)
+	clientBob.RegisterSelfNode(suite.T(), true)
+	clientCharlie.RegisterSelfOrg(suite.T(), true)
+	clientCharlie.RegisterSelfNode(suite.T(), true)
+
+	// Verify that a broadcast on the new namespace succeeds
+	resp, err := clientAlice.BroadcastMessage(suite.T(), "topic", data, false)
+	require.NoError(suite.T(), err)
+	assert.Equal(suite.T(), 202, resp.StatusCode())
+	e2e.WaitForMessageConfirmed(suite.T(), receivedAlice, core.MessageTypeBroadcast)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedBob, core.MessageTypeBroadcast)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedCharlie, core.MessageTypeBroadcast)
+
+	toAlice := []core.MemberInput{
+		{Identity: suite.testState.org1.Name, Node: "did:firefly:node/alice"},
+	}
+	toBob := []core.MemberInput{
+		{Identity: suite.testState.org1.Name, Node: "did:firefly:node/bob"},
+	}
+	toCharlie := []core.MemberInput{
+		{Identity: suite.testState.org2.Name, Node: "did:firefly:node/charlie"},
+	}
+	toAll := []core.MemberInput{
+		{Identity: suite.testState.org1.Name, Node: "did:firefly:node/alice"},
+		{Identity: suite.testState.org1.Name, Node: "did:firefly:node/bob"},
+		{Identity: suite.testState.org2.Name, Node: "did:firefly:node/charlie"},
+	}
+
+	// Verify that private messages on the new namespace succeed
+	// Alice -> Charlie
+	resp, err = clientAlice.PrivateMessage("topic", data, toCharlie, "tag1", core.TransactionTypeBatchPin, false, suite.testState.startTime)
+	require.NoError(suite.T(), err)
+	assert.Equal(suite.T(), 202, resp.StatusCode())
+	e2e.WaitForMessageConfirmed(suite.T(), receivedAlice, core.MessageTypePrivate)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedCharlie, core.MessageTypePrivate)
+	// Charlie -> Alice
+	resp, err = clientCharlie.PrivateMessage("topic", data, toAlice, "tag2", core.TransactionTypeBatchPin, false, suite.testState.startTime)
+	require.NoError(suite.T(), err)
+	assert.Equal(suite.T(), 202, resp.StatusCode())
+	e2e.WaitForMessageConfirmed(suite.T(), receivedAlice, core.MessageTypePrivate)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedCharlie, core.MessageTypePrivate)
+	// Charlie -> Alice+Bob
+	resp, err = clientCharlie.PrivateMessage("topic", data, toAll, "tag3", core.TransactionTypeBatchPin, false, suite.testState.startTime)
+	require.NoError(suite.T(), err)
+	assert.Equal(suite.T(), 202, resp.StatusCode())
+	e2e.WaitForMessageConfirmed(suite.T(), receivedAlice, core.MessageTypePrivate)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedBob, core.MessageTypePrivate)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedCharlie, core.MessageTypePrivate)
+	// Alice -> Bob
+	resp, err = clientAlice.PrivateMessage("topic", data, toBob, "tag4", core.TransactionTypeBatchPin, false, suite.testState.startTime)
+	require.NoError(suite.T(), err)
+	assert.Equal(suite.T(), 202, resp.StatusCode())
+	e2e.WaitForMessageConfirmed(suite.T(), receivedAlice, core.MessageTypePrivate)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedBob, core.MessageTypePrivate)
+	// Bob -> Alice+Charlie (blob transfer)
+	dataRef, resp, err := clientBob.PrivateBlobMessageDatatypeTagged(suite.T(), "topic", toAll, suite.testState.startTime)
+	require.NoError(suite.T(), err)
+	assert.Equal(suite.T(), 202, resp.StatusCode())
+	assert.NotNil(suite.T(), dataRef.Blob.Hash)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedAlice, core.MessageTypePrivate)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedBob, core.MessageTypePrivate)
+	e2e.WaitForMessageConfirmed(suite.T(), receivedCharlie, core.MessageTypePrivate)
+
+	dataRecv := clientAlice.GetDataByID(suite.T(), dataRef.ID)
+	assert.Equal(suite.T(), dataRef.Blob.Hash, dataRecv.Blob.Hash)
+	dataRecv = clientCharlie.GetDataByID(suite.T(), dataRef.ID)
+	assert.Equal(suite.T(), dataRef.Blob.Hash, dataRecv.Blob.Hash)
+
+	messages := clientAlice.GetMessages(suite.T(), suite.testState.startTime, core.MessageTypePrivate, "topic")
+	assert.Len(suite.T(), messages, 5)
+	messages = clientBob.GetMessages(suite.T(), suite.testState.startTime, core.MessageTypePrivate, "topic")
+	assert.Len(suite.T(), messages, 3)
+	messages = clientCharlie.GetMessages(suite.T(), suite.testState.startTime, core.MessageTypePrivate, "topic")
+	assert.Len(suite.T(), messages, 4)
+}
diff --git a/test/e2e/multiparty/namespace_alias.go b/test/e2e/multiparty/namespace_alias.go
deleted file mode 100644
index c203221e2..000000000
--- a/test/e2e/multiparty/namespace_alias.go
+++ /dev/null
@@ -1,119 +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 multiparty
-
-import (
-	"fmt"
-
-	"github.com/hyperledger/firefly-common/pkg/fftypes"
-	"github.com/hyperledger/firefly/pkg/core"
-	"github.com/hyperledger/firefly/test/e2e"
-	"github.com/hyperledger/firefly/test/e2e/client"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"github.com/stretchr/testify/suite"
-)
-
-type NamespaceAliasSuite struct {
-	suite.Suite
-	testState *testState
-}
-
-func (suite *NamespaceAliasSuite) SetupSuite() {
-	suite.testState = beforeE2ETest(suite.T())
-}
-
-func (suite *NamespaceAliasSuite) BeforeTest(suiteName, testName string) {
-	suite.testState = beforeE2ETest(suite.T())
-}
-
-func (suite *NamespaceAliasSuite) AfterTest(suiteName, testName string) {
-	e2e.VerifyAllOperationsSucceeded(suite.T(), []*client.FireFlyClient{suite.testState.client1, suite.testState.client2}, suite.testState.startTime)
-}
-
-func (suite *NamespaceAliasSuite) TestNamespaceMapping() {
-	defer suite.testState.Done()
-
-	address := deployContract(suite.T(), suite.testState.stackName, "firefly/Firefly.json")
-	localNamespace1 := e2e.RandomName(suite.T())
-	localNamespace2 := e2e.RandomName(suite.T())
-	remoteNamespace := e2e.RandomName(suite.T())
-	suite.T().Logf("Test namespace: local1=%s, local2=%s, remote=%s", localNamespace1, localNamespace2, remoteNamespace)
-
-	org := map[string]interface{}{}
-	namespaceInfo := map[string]interface{}{
-		"remotename": remoteNamespace,
-		"multiparty": map[string]interface{}{
-			"enabled": true,
-			"org":     org,
-			"contract": []map[string]interface{}{
-				{
-					"location": map[string]interface{}{"address": address},
-				},
-			},
-		},
-	}
-	data := &core.DataRefOrValue{Value: fftypes.JSONAnyPtr(`"test"`)}
-
-	// Add the new namespace to both config files
-	data1 := e2e.ReadConfig(suite.T(), suite.testState.configFile1)
-	namespaceInfo["name"] = localNamespace1
-	org["name"] = suite.testState.org1.Name
-	org["key"] = suite.testState.org1key.Value
-	e2e.AddNamespace(data1, namespaceInfo)
-	e2e.WriteConfig(suite.T(), suite.testState.configFile1, data1)
-
-	data2 := e2e.ReadConfig(suite.T(), suite.testState.configFile2)
-	namespaceInfo["name"] = localNamespace2
-	org["name"] = suite.testState.org2.Name
-	org["key"] = suite.testState.org2key.Value
-	e2e.AddNamespace(data2, namespaceInfo)
-	e2e.WriteConfig(suite.T(), suite.testState.configFile2, data2)
-
-	admin1 := client.NewResty(suite.T())
-	admin2 := client.NewResty(suite.T())
-	admin1.SetBaseURL(suite.testState.adminHost1 + "/spi/v1")
-	admin2.SetBaseURL(suite.testState.adminHost2 + "/spi/v1")
-
-	// Reset both nodes to pick up the new namespace
-	e2e.ResetFireFly(suite.T(), admin1)
-	e2e.ResetFireFly(suite.T(), admin2)
-	e2e.PollForUp(suite.T(), suite.testState.client1)
-	e2e.PollForUp(suite.T(), suite.testState.client2)
-
-	client1 := client.NewFireFly(suite.T(), suite.testState.client1.Hostname, localNamespace1)
-	client2 := client.NewFireFly(suite.T(), suite.testState.client2.Hostname, localNamespace2)
-
-	eventNames := "message_confirmed|blockchain_event_received"
-	queryString := fmt.Sprintf("namespace=%s&ephemeral&autoack&filter.events=%s", localNamespace1, eventNames)
-	received1 := e2e.WsReader(client1.WebSocket(suite.T(), queryString, nil))
-	queryString = fmt.Sprintf("namespace=%s&ephemeral&autoack&filter.events=%s", localNamespace2, eventNames)
-	received2 := e2e.WsReader(client2.WebSocket(suite.T(), queryString, nil))
-
-	// Register org/node identities on the new namespace
-	client1.RegisterSelfOrg(suite.T(), true)
-	client1.RegisterSelfNode(suite.T(), true)
-	client2.RegisterSelfOrg(suite.T(), true)
-	client2.RegisterSelfNode(suite.T(), true)
-
-	// Verify that a broadcast on the new namespace succeeds
-	resp, err := client1.BroadcastMessage(suite.T(), "topic", data, false)
-	require.NoError(suite.T(), err)
-	assert.Equal(suite.T(), 202, resp.StatusCode())
-	e2e.WaitForMessageConfirmed(suite.T(), received1, core.MessageTypeBroadcast)
-	e2e.WaitForMessageConfirmed(suite.T(), received2, core.MessageTypeBroadcast)
-}
diff --git a/test/e2e/multiparty/onchain_offchain.go b/test/e2e/multiparty/onchain_offchain.go
index 8e62a1202..0119ee550 100644
--- a/test/e2e/multiparty/onchain_offchain.go
+++ b/test/e2e/multiparty/onchain_offchain.go
@@ -159,11 +159,13 @@ func (suite *OnChainOffChainTestSuite) TestStrongDatatypesPrivate() {
 		},
 	}
 
+	members := []core.MemberInput{
+		{Identity: suite.testState.org1.Name},
+		{Identity: suite.testState.org2.Name},
+	}
+
 	// Should be rejected as datatype not known
-	resp, err := suite.testState.client1.PrivateMessage("topic1", &data, []string{
-		suite.testState.org1.Name,
-		suite.testState.org2.Name,
-	}, "", core.TransactionTypeBatchPin, true, suite.testState.startTime)
+	resp, err := suite.testState.client1.PrivateMessage("topic1", &data, members, "", core.TransactionTypeBatchPin, true, suite.testState.startTime)
 	require.NoError(suite.T(), err)
 	assert.Equal(suite.T(), 400, resp.StatusCode())
 	assert.Contains(suite.T(), resp.String(), "FF10195") // datatype not found
@@ -175,10 +177,7 @@ func (suite *OnChainOffChainTestSuite) TestStrongDatatypesPrivate() {
 	}
 	suite.testState.client1.CreateDatatype(suite.T(), dt, true)
 
-	resp, err = suite.testState.client1.PrivateMessage("topic1", &data, []string{
-		suite.testState.org1.Name,
-		suite.testState.org2.Name,
-	}, "", core.TransactionTypeBatchPin, false, suite.testState.startTime)
+	resp, err = suite.testState.client1.PrivateMessage("topic1", &data, members, "", core.TransactionTypeBatchPin, false, suite.testState.startTime)
 	require.NoError(suite.T(), err)
 	assert.Equal(suite.T(), 400, resp.StatusCode())
 	assert.Contains(suite.T(), resp.String(), "FF10198") // does not conform
@@ -188,10 +187,7 @@ func (suite *OnChainOffChainTestSuite) TestStrongDatatypesPrivate() {
 		"name": "mywidget"
 	}`)
 
-	resp, err = suite.testState.client1.PrivateMessage("topic1", &data, []string{
-		suite.testState.org1.Name,
-		suite.testState.org2.Name,
-	}, "", core.TransactionTypeBatchPin, true, suite.testState.startTime)
+	resp, err = suite.testState.client1.PrivateMessage("topic1", &data, members, "", core.TransactionTypeBatchPin, true, suite.testState.startTime)
 	require.NoError(suite.T(), err)
 	assert.Equal(suite.T(), 200, resp.StatusCode())
 
@@ -206,6 +202,11 @@ func (suite *OnChainOffChainTestSuite) TestE2EPrivate() {
 	received1 := e2e.WsReader(suite.testState.ws1)
 	received2 := e2e.WsReader(suite.testState.ws2)
 
+	members := []core.MemberInput{
+		{Identity: suite.testState.org1.Name},
+		{Identity: suite.testState.org2.Name},
+	}
+
 	// Send 10 messages, that should get batched, across two topics
 	totalMessages := 10
 	topics := []string{"topicA", "topicB"}
@@ -219,10 +220,7 @@ func (suite *OnChainOffChainTestSuite) TestE2EPrivate() {
 
 		expectedData[topic] = append(expectedData[topic], data)
 
-		resp, err := suite.testState.client1.PrivateMessage(topic, data, []string{
-			suite.testState.org1.Name,
-			suite.testState.org2.Name,
-		}, "", core.TransactionTypeBatchPin, false, suite.testState.startTime)
+		resp, err := suite.testState.client1.PrivateMessage(topic, data, members, "", core.TransactionTypeBatchPin, false, suite.testState.startTime)
 		require.NoError(suite.T(), err)
 		assert.Equal(suite.T(), 202, resp.StatusCode())
 	}
@@ -279,10 +277,12 @@ func (suite *OnChainOffChainTestSuite) TestE2EPrivateBlobDatatypeTagged() {
 
 	var resp *resty.Response
 
-	data, resp, err := suite.testState.client1.PrivateBlobMessageDatatypeTagged(suite.T(), "topic1", []string{
-		suite.testState.org1.Name,
-		suite.testState.org2.Name,
-	}, suite.testState.startTime)
+	members := []core.MemberInput{
+		{Identity: suite.testState.org1.Name},
+		{Identity: suite.testState.org2.Name},
+	}
+
+	data, resp, err := suite.testState.client1.PrivateBlobMessageDatatypeTagged(suite.T(), "topic1", members, suite.testState.startTime)
 	require.NoError(suite.T(), err)
 	assert.Equal(suite.T(), 202, resp.StatusCode())
 	assert.Empty(suite.T(), data.Blob.Name)
@@ -330,11 +330,13 @@ func (suite *OnChainOffChainTestSuite) TestE2EWebhookExchange() {
 		Value: fftypes.JSONAnyPtr(`{}`),
 	}
 
+	members := []core.MemberInput{
+		{Identity: suite.testState.org1.Name},
+		{Identity: suite.testState.org2.Name},
+	}
+
 	var resp *resty.Response
-	resp, err := suite.testState.client1.PrivateMessage("topic1", &data, []string{
-		suite.testState.org1.Name,
-		suite.testState.org2.Name,
-	}, "myrequest", core.TransactionTypeBatchPin, false, suite.testState.startTime)
+	resp, err := suite.testState.client1.PrivateMessage("topic1", &data, members, "myrequest", core.TransactionTypeBatchPin, false, suite.testState.startTime)
 	require.NoError(suite.T(), err)
 	assert.Equal(suite.T(), 202, resp.StatusCode())
 
diff --git a/test/e2e/runners/ethereum_multiparty_test.go b/test/e2e/runners/ethereum_multiparty_test.go
index 2c1221fb8..25b297cfa 100644
--- a/test/e2e/runners/ethereum_multiparty_test.go
+++ b/test/e2e/runners/ethereum_multiparty_test.go
@@ -24,8 +24,8 @@ import (
 )
 
 func TestEthereumMultipartyE2ESuite(t *testing.T) {
-	suite.Run(t, new(multiparty.IdentityTestSuite))
 	suite.Run(t, new(multiparty.OnChainOffChainTestSuite))
+	suite.Run(t, new(multiparty.IdentityTestSuite))
 	suite.Run(t, new(multiparty.TokensTestSuite))
 	suite.Run(t, new(multiparty.EthereumContractTestSuite))
 	suite.Run(t, new(multiparty.ContractMigrationTestSuite))
diff --git a/test/e2e/runners/fabric_multiparty_test.go b/test/e2e/runners/fabric_multiparty_test.go
index e42349656..8c848ce18 100644
--- a/test/e2e/runners/fabric_multiparty_test.go
+++ b/test/e2e/runners/fabric_multiparty_test.go
@@ -24,7 +24,7 @@ import (
 )
 
 func TestFabricMultipartyE2ESuite(t *testing.T) {
-	suite.Run(t, new(multiparty.IdentityTestSuite))
 	suite.Run(t, new(multiparty.OnChainOffChainTestSuite))
+	suite.Run(t, new(multiparty.IdentityTestSuite))
 	suite.Run(t, new(multiparty.FabricContractTestSuite))
 }