Repository: hyperledger/firefly
Pull request #1203 (2023-03-06 04:27:39)

[tokens] Only Use broadcastName for Multi-Party Namespaces
We saw if `broadcastName` was set on a tokens plugin, token pools would fail to create within gateway namespaces:
```
[2023-03-01T20:25:10.568Z] ERROR Failed to activate token pool 'f2de67bb-5adf-4b98-ac06-7d25a19805ad': FF10272: Unknown tokens plugin 'erc20-erc721' ns=cok pid=1 role=event-manager
[2023-03-01T20:25:10.568Z] DEBUG fftokens updating operation cok:fb435cab-a2a5-4d1a-af28-f0bf84840c0b status=Failed error=FF10272: Unknown tokens plugin 'erc20-erc721' ns=cok pid=1
[2023-03-01T20:25:10.568Z] DEBUG SQL-> begin dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.568Z] DEBUG SQL<- begin dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.569Z] DEBUG SQL-> update operations dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.569Z] DEBUG SQL<- update operations affected=1 dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.569Z] DEBUG SQL-> lock cok dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.570Z] DEBUG SQL<- lock cok dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.570Z] DEBUG SQL-> insert events dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.570Z] DEBUG SQL<- insert events sequences=[9] dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.570Z] DEBUG SQL-> commit dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.573Z] DEBUG SQL<- commit dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.573Z]  INFO Emitted token_pool_op_failed event 47472d4f-70f7-4f3e-827d-a3e70c76cf72 for cok:fb435cab-a2a5-4d1a-af28-f0bf84840c0b (correlator=f2de67bb-5adf-4b98-ac06-7d25a19805ad,topic=f2de67bb-5adf-4b98-ac06-7d25a19805ad) dbtx=N1fXw8pi ns=cok pid=1
[2023-03-01T20:25:10.573Z]  INFO Inflight request 'f2de67bb-5adf-4b98-ac06-7d25a19805ad' resolved with timeout after 2068.43ms ns=cok pid=1 role=sync-async-bridge
[2023-03-01T20:25:10.573Z]  INFO <-- POST /api/v1/namespaces/cok/tokens/pools [400] (2071.90ms): FF10272: Unknown tokens plugin 'erc20-erc721' httpreq=hzG60xJW pid=1 req=Rd7exmVy
```

This is bc the definition handler doesn't know how to map the remote name back to the local name within gateway namespaces: https://github.com/hyperledger/firefly/blob/3b728534d50c0cee6ed27890114b5a4c44fcec5c/internal/definitions/handler_tokenpool.go#L36-L42

This fixes that by only using the `broadcastName` for multi-party namespaces when sending the token pool definition.


Patch:
diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
index 75050966e..8170150f4 100644
--- a/docs/swagger/swagger.yaml
+++ b/docs/swagger/swagger.yaml
@@ -23385,7 +23385,8 @@ paths:
   /namespaces/{ns}/pins/rewind:
     post:
       description: Force a rewind of the event aggregator to a previous position,
-        to re-evaluate all unconfirmed pins since that point
+        to re-evaluate (and possibly dispatch) that pin and others after it. Only
+        accepts a sequence or batch ID for a currently undispatched pin
       operationId: postPinsRewindNamespace
       parameters:
       - description: The namespace which scopes this request
@@ -30931,7 +30932,8 @@ paths:
   /pins/rewind:
     post:
       description: Force a rewind of the event aggregator to a previous position,
-        to re-evaluate all unconfirmed pins since that point
+        to re-evaluate (and possibly dispatch) that pin and others after it. Only
+        accepts a sequence or batch ID for a currently undispatched pin
       operationId: postPinsRewind
       parameters:
       - description: Server-side request timeout (milliseconds, or set a custom suffix
diff --git a/go.mod b/go.mod
index 730777f41..a1b923b9c 100644
--- a/go.mod
+++ b/go.mod
@@ -14,7 +14,7 @@ require (
 	github.com/golang-migrate/migrate/v4 v4.15.2
 	github.com/gorilla/mux v1.8.0
 	github.com/gorilla/websocket v1.5.0
-	github.com/hyperledger/firefly-common v1.2.1
+	github.com/hyperledger/firefly-common v1.2.3
 	github.com/hyperledger/firefly-signer v1.1.5
 	github.com/jarcoal/httpmock v1.2.0
 	github.com/karlseguin/ccache v2.0.3+incompatible
@@ -28,8 +28,8 @@ require (
 	github.com/spf13/viper v1.14.0
 	github.com/stretchr/testify v1.8.1
 	gitlab.com/hfuss/mux-prometheus v0.0.4
-	golang.org/x/net v0.4.0
-	golang.org/x/text v0.5.0
+	golang.org/x/net v0.7.0
+	golang.org/x/text v0.7.0
 	gopkg.in/yaml.v2 v2.4.0
 )
 
@@ -79,8 +79,8 @@ require (
 	github.com/x-cray/logrus-prefixed-formatter v0.5.2 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	golang.org/x/crypto v0.4.0 // indirect
-	golang.org/x/sys v0.3.0 // indirect
-	golang.org/x/term v0.3.0 // indirect
+	golang.org/x/sys v0.5.0 // indirect
+	golang.org/x/term v0.5.0 // indirect
 	google.golang.org/protobuf v1.28.1 // indirect
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/natefinch/lumberjack.v2 v2.0.0 // indirect
diff --git a/go.sum b/go.sum
index e199f70e2..d0bdc5d18 100644
--- a/go.sum
+++ b/go.sum
@@ -674,8 +674,8 @@ github.com/hashicorp/mdns v1.0.0/go.mod h1:tL+uN++7HEJ6SQLQ2/p+z2pH24WQKWjBPkE0m
 github.com/hashicorp/memberlist v0.1.3/go.mod h1:ajVTdAv/9Im8oMAAj5G31PhhMCZJV2pPBoIllUwCN7I=
 github.com/hashicorp/serf v0.8.2/go.mod h1:6hOLApaqBFA1NXqRQAsxw9QxuDEvNxSQRwA/JwenrHc=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
-github.com/hyperledger/firefly-common v1.2.1 h1:h35T5Ask/ixlW70KZxHpPsO5QfdnKBFEyX+7L10lwoo=
-github.com/hyperledger/firefly-common v1.2.1/go.mod h1:aqaq2ZUzSlkwVC/TF+pwUkHxGDLYSE2DzdozZ1MIMeM=
+github.com/hyperledger/firefly-common v1.2.3 h1:6vlCEVUHZn8QdXdu3qrzfKLnzXsBffAKjWFiT9uWhA4=
+github.com/hyperledger/firefly-common v1.2.3/go.mod h1:02hp5BYzAito5JMTjQZluk92xqhL0LxqEEHNENsVcFk=
 github.com/hyperledger/firefly-signer v1.1.5 h1:kCHe4O3tmYaMYf1hXrG40sIlpiGfOICmNNAcuMH5+jQ=
 github.com/hyperledger/firefly-signer v1.1.5/go.mod h1:XZP8ZXUZJSAPB6WUSM9RqRdpjnjm6Jsn0784WM4IKug=
 github.com/iancoleman/strcase v0.2.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47ZCWhYzw7ho=
@@ -1396,8 +1396,8 @@ golang.org/x/net v0.0.0-20211209124913-491a49abca63/go.mod h1:9nx3DQGgdP8bBQD5qx
 golang.org/x/net v0.0.0-20211216030914-fe4d6282115f/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220111093109-d55c255bac03/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
 golang.org/x/net v0.0.0-20220225172249-27dd8689420f/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
-golang.org/x/net v0.4.0 h1:Q5QPcMlvfxFTAPV0+07Xz/MpK9NTXu2VDUuy0FeMfaU=
-golang.org/x/net v0.4.0/go.mod h1:MBQ8lrhLObU/6UmLb4fmbmk5OcyYmqtbGd/9yIeKjEE=
+golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
+golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/oauth2 v0.0.0-20180227000427-d7d64896b5ff/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
 golang.org/x/oauth2 v0.0.0-20181106182150-f42d05182288/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
@@ -1552,15 +1552,15 @@ golang.org/x/sys v0.0.0-20220317061510-51cd9980dadf/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.3.0 h1:w8ZOecv6NaNa/zC8944JTU3vz4u6Lagfk4RPQxv92NQ=
-golang.org/x/sys v0.3.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU=
+golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210615171337-6886f2dfbf5b/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
-golang.org/x/term v0.3.0 h1:qoo4akIqOcDME5bhc/NgxUdovd6BSS2uMsVjB56q1xI=
-golang.org/x/term v0.3.0/go.mod h1:q750SLmJuPmVoN1blW3UFBPREJfb1KmY3vwxfr+nFDA=
+golang.org/x/term v0.5.0 h1:n2a8QNdAb0sZNpU9R1ALUXBbY+w51fCQDN+7EdxNBsY=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/text v0.0.0-20170915032832-14c0d48ead0c/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
@@ -1570,8 +1570,8 @@ golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
-golang.org/x/text v0.5.0 h1:OLmvp0KP+FVG99Ct/qFiL/Fhk4zp4QQnZ7b2U+5piUM=
-golang.org/x/text v0.5.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
 golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
diff --git a/internal/batch/batch_processor.go b/internal/batch/batch_processor.go
index 011a6559d..ba25c04c4 100644
--- a/internal/batch/batch_processor.go
+++ b/internal/batch/batch_processor.go
@@ -577,7 +577,8 @@ func (bp *batchProcessor) sealBatch(state *DispatchState) (err error) {
 			log.L(ctx).Debugf("Batch %s sealed. Hash=%s", state.Persisted.ID, state.Persisted.Hash)
 
 			// At this point the manifest of the batch is finalized. We write it to the database
-			return bp.database.UpsertBatch(ctx, &state.Persisted)
+			_, err = bp.database.InsertOrGetBatch(ctx, &state.Persisted)
+			return err
 		})
 	})
 	if err != nil {
diff --git a/internal/coremsgs/en_api_translations.go b/internal/coremsgs/en_api_translations.go
index b62c69259..1bd482b42 100644
--- a/internal/coremsgs/en_api_translations.go
+++ b/internal/coremsgs/en_api_translations.go
@@ -169,7 +169,7 @@ var (
 	APIEndpointsPostNewOrganization             = ffm("api.endpoints.postNewOrganization", "Registers a new org in the network")
 	APIEndpointsPostNewSubscription             = ffm("api.endpoints.postNewSubscription", "Creates a new subscription for an application to receive events from FireFly")
 	APIEndpointsPostOpRetry                     = ffm("api.endpoints.postOpRetry", "Retries a failed operation")
-	APIEndpointsPostPinsRewind                  = ffm("api.endpoints.postPinsRewind", "Force a rewind of the event aggregator to a previous position, to re-evaluate all unconfirmed pins since that point")
+	APIEndpointsPostPinsRewind                  = ffm("api.endpoints.postPinsRewind", "Force a rewind of the event aggregator to a previous position, to re-evaluate (and possibly dispatch) that pin and others after it. Only accepts a sequence or batch ID for a currently undispatched pin")
 	APIEndpointsPostTokenApproval               = ffm("api.endpoints.postTokenApproval", "Creates a token approval")
 	APIEndpointsPostTokenBurn                   = ffm("api.endpoints.postTokenBurn", "Burns some tokens")
 	APIEndpointsPostTokenMint                   = ffm("api.endpoints.postTokenMint", "Mints some tokens")
diff --git a/internal/coremsgs/en_error_messages.go b/internal/coremsgs/en_error_messages.go
index a704e98cb..92c7aa368 100644
--- a/internal/coremsgs/en_error_messages.go
+++ b/internal/coremsgs/en_error_messages.go
@@ -241,7 +241,7 @@ var (
 	MsgDefRejectedBadPayload              = ffe("FF10400", "Rejected %s message '%s' - invalid payload")
 	MsgDefRejectedAuthorBlank             = ffe("FF10401", "Rejected %s message '%s' - author is blank")
 	MsgDefRejectedSignatureMismatch       = ffe("FF10402", "Rejected %s message '%s' - signature mismatch")
-	MsgDefRejectedValidateFail            = ffe("FF10403", "Rejected %s '%s' - validate failed: %s")
+	MsgDefRejectedValidateFail            = ffe("FF10403", "Rejected %s '%s' - validate failed")
 	MsgDefRejectedIDMismatch              = ffe("FF10404", "Rejected %s '%s' - ID mismatch with existing record")
 	MsgDefRejectedLocationMismatch        = ffe("FF10405", "Rejected %s '%s' - location mismatch with existing record")
 	MsgDefRejectedSchemaFail              = ffe("FF10406", "Rejected %s '%s' - schema check: %s")
diff --git a/internal/database/sqlcommon/batch_sql.go b/internal/database/sqlcommon/batch_sql.go
index 6e9684f4f..007209e40 100644
--- a/internal/database/sqlcommon/batch_sql.go
+++ b/internal/database/sqlcommon/batch_sql.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -57,87 +57,49 @@ var (
 
 const batchesTable = "batches"
 
-func (s *SQLCommon) UpsertBatch(ctx context.Context, batch *core.BatchPersisted) (err error) {
+func (s *SQLCommon) InsertOrGetBatch(ctx context.Context, batch *core.BatchPersisted) (existing *core.BatchPersisted, err error) {
 	ctx, tx, autoCommit, err := s.BeginOrUseTx(ctx)
 	if err != nil {
-		return err
+		return nil, err
 	}
 	defer s.RollbackTx(ctx, tx, autoCommit)
 
-	// Do a select within the transaction to detemine if the UUID already exists
-	batchRows, _, err := s.QueryTx(ctx, batchesTable, tx,
-		sq.Select("hash").
-			From(batchesTable).
-			Where(sq.Eq{"id": batch.ID, "namespace": batch.Namespace}),
+	// Try the insert first
+	_, insertErr := s.InsertTxExt(ctx, batchesTable, tx,
+		sq.Insert(batchesTable).
+			Columns(batchColumns...).
+			Values(
+				batch.ID,
+				string(batch.Type),
+				batch.Namespace,
+				batch.Author,
+				batch.Key,
+				batch.Group,
+				batch.Created,
+				batch.Hash,
+				batch.Manifest,
+				batch.Confirmed,
+				batch.TX.Type,
+				batch.TX.ID,
+				batch.Node,
+			),
+		func() {
+			s.callbacks.UUIDCollectionNSEvent(database.CollectionBatches, core.ChangeEventTypeCreated, batch.Namespace, batch.ID)
+		},
+		true, /* we want a failure here we can progress past */
 	)
-	if err != nil {
-		return err
+	if insertErr == nil {
+		return nil, s.CommitTx(ctx, tx, autoCommit)
 	}
 
-	existing := batchRows.Next()
-	if existing {
-		var hash *fftypes.Bytes32
-		_ = batchRows.Scan(&hash)
-		if !fftypes.SafeHashCompare(hash, batch.Hash) {
-			batchRows.Close()
-			log.L(ctx).Errorf("Existing=%s New=%s", hash, batch.Hash)
-			return database.HashMismatch
-		}
-	}
-	batchRows.Close()
-
-	if existing {
-
-		// Update the batch
-		if _, err = s.UpdateTx(ctx, batchesTable, tx,
-			sq.Update(batchesTable).
-				Set("btype", string(batch.Type)).
-				Set("author", batch.Author).
-				Set("key", batch.Key).
-				Set("group_hash", batch.Group).
-				Set("created", batch.Created).
-				Set("hash", batch.Hash).
-				Set("manifest", batch.Manifest).
-				Set("confirmed", batch.Confirmed).
-				Set("tx_type", batch.TX.Type).
-				Set("tx_id", batch.TX.ID).
-				Set("node_id", batch.Node).
-				Where(sq.Eq{"id": batch.ID, "namespace": batch.Namespace}),
-			func() {
-				s.callbacks.UUIDCollectionNSEvent(database.CollectionBatches, core.ChangeEventTypeUpdated, batch.Namespace, batch.ID)
-			},
-		); err != nil {
-			return err
-		}
-	} else {
-
-		if _, err = s.InsertTx(ctx, batchesTable, tx,
-			sq.Insert(batchesTable).
-				Columns(batchColumns...).
-				Values(
-					batch.ID,
-					string(batch.Type),
-					batch.Namespace,
-					batch.Author,
-					batch.Key,
-					batch.Group,
-					batch.Created,
-					batch.Hash,
-					batch.Manifest,
-					batch.Confirmed,
-					batch.TX.Type,
-					batch.TX.ID,
-					batch.Node,
-				),
-			func() {
-				s.callbacks.UUIDCollectionNSEvent(database.CollectionBatches, core.ChangeEventTypeCreated, batch.Namespace, batch.ID)
-			},
-		); err != nil {
-			return err
-		}
+	// Do a select within the transaction to determine if the batch already exists
+	existing, err = s.GetBatchByID(ctx, batch.Namespace, batch.ID)
+	if err != nil || existing != nil {
+		return existing, err
 	}
 
-	return s.CommitTx(ctx, tx, autoCommit)
+	// Error was apparently not an ID conflict - must have been something else
+	return nil, insertErr
 }
 
 func (s *SQLCommon) batchResult(ctx context.Context, row *sql.Rows) (*core.BatchPersisted, error) {
diff --git a/internal/database/sqlcommon/blob_sql.go b/internal/database/sqlcommon/blob_sql.go
index 98a8fb8cb..f702eb7c5 100644
--- a/internal/database/sqlcommon/blob_sql.go
+++ b/internal/database/sqlcommon/blob_sql.go
@@ -88,7 +88,7 @@ func (s *SQLCommon) InsertBlobs(ctx context.Context, blobs []*core.Blob) (err er
 	}
 	defer s.RollbackTx(ctx, tx, autoCommit)
 
-	if s.features.MultiRowInsert {
+	if s.Features().MultiRowInsert {
 		query := sq.Insert(blobsTable).Columns(blobColumns...)
 		for _, blob := range blobs {
 			query = s.setBlobInsertValues(query, blob)
diff --git a/internal/database/sqlcommon/data_sql.go b/internal/database/sqlcommon/data_sql.go
index d2faa656d..4657874cc 100644
--- a/internal/database/sqlcommon/data_sql.go
+++ b/internal/database/sqlcommon/data_sql.go
@@ -195,7 +195,7 @@ func (s *SQLCommon) InsertDataArray(ctx context.Context, dataArray core.DataArra
 	}
 	defer s.RollbackTx(ctx, tx, autoCommit)
 
-	if s.features.MultiRowInsert {
+	if s.Features().MultiRowInsert {
 		query := sq.Insert(dataTable).Columns(dataColumnsWithValue...)
 		for _, data := range dataArray {
 			query = s.setDataInsertValues(query, data)
diff --git a/internal/database/sqlcommon/event_sql.go b/internal/database/sqlcommon/event_sql.go
index 2af652559..574add68e 100644
--- a/internal/database/sqlcommon/event_sql.go
+++ b/internal/database/sqlcommon/event_sql.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -118,7 +118,7 @@ func (p *eventsPCA) PreCommit(ctx context.Context, tx *dbsql.TXWrapper) (err err
 		}
 	}
 
-	if p.s.features.MultiRowInsert {
+	if p.s.Features().MultiRowInsert {
 		query := sq.Insert(eventsTable).Columns(eventColumns...)
 		for _, event := range p.events {
 			query = p.s.setEventInsertValues(query, event)
diff --git a/internal/database/sqlcommon/message_sql.go b/internal/database/sqlcommon/message_sql.go
index e56cb3471..c1339a4c4 100644
--- a/internal/database/sqlcommon/message_sql.go
+++ b/internal/database/sqlcommon/message_sql.go
@@ -236,8 +236,7 @@ func (s *SQLCommon) InsertMessages(ctx context.Context, messages []*core.Message
 		return err
 	}
 	defer s.RollbackTx(ctx, tx, autoCommit)
-
-	if s.features.MultiRowInsert {
+	if s.Features().MultiRowInsert {
 		msgQuery := sq.Insert(messagesTable).Columns(msgColumns...)
 		dataRefQuery := sq.Insert(messagesDataJoinTable).Columns(
 			"namespace",
diff --git a/internal/database/sqlcommon/pin_sql.go b/internal/database/sqlcommon/pin_sql.go
index 794073b29..a2961aa40 100644
--- a/internal/database/sqlcommon/pin_sql.go
+++ b/internal/database/sqlcommon/pin_sql.go
@@ -124,7 +124,7 @@ func (s *SQLCommon) InsertPins(ctx context.Context, pins []*core.Pin) error {
 	}
 	defer s.RollbackTx(ctx, tx, autoCommit)
 
-	if s.features.MultiRowInsert {
+	if s.Features().MultiRowInsert {
 		query := sq.Insert(pinsTable).Columns(pinColumns...)
 		for _, pin := range pins {
 			query = s.setPinInsertValues(query, pin)
diff --git a/internal/database/sqlcommon/sqlcommon.go b/internal/database/sqlcommon/sqlcommon.go
index 1105604b8..02dbdd8ee 100644
--- a/internal/database/sqlcommon/sqlcommon.go
+++ b/internal/database/sqlcommon/sqlcommon.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -33,7 +33,6 @@ type SQLCommon struct {
 	dbsql.Database
 	capabilities *database.Capabilities
 	callbacks    callbacks
-	features     dbsql.SQLFeatures
 }
 
 type callbacks struct {
diff --git a/internal/definitions/handler_contracts.go b/internal/definitions/handler_contracts.go
index ca692c2ab..743e02a79 100644
--- a/internal/definitions/handler_contracts.go
+++ b/internal/definitions/handler_contracts.go
@@ -29,7 +29,7 @@ import (
 
 func (dh *definitionHandler) persistFFI(ctx context.Context, ffi *fftypes.FFI) (retry bool, err error) {
 	if err = dh.contracts.ResolveFFI(ctx, ffi); err != nil {
-		return false, i18n.NewError(ctx, coremsgs.MsgDefRejectedValidateFail, "contract interface", ffi.ID, err)
+		return false, i18n.WrapError(ctx, err, coremsgs.MsgDefRejectedValidateFail, "contract interface", ffi.ID)
 	}
 
 	err = dh.database.UpsertFFI(ctx, ffi)
@@ -58,7 +58,7 @@ func (dh *definitionHandler) persistFFI(ctx context.Context, ffi *fftypes.FFI) (
 
 func (dh *definitionHandler) persistContractAPI(ctx context.Context, httpServerURL string, api *core.ContractAPI) (retry bool, err error) {
 	if err := dh.contracts.ResolveContractAPI(ctx, httpServerURL, api); err != nil {
-		return false, i18n.NewError(ctx, coremsgs.MsgDefRejectedValidateFail, "contract API", api.ID, err)
+		return false, i18n.WrapError(ctx, err, coremsgs.MsgDefRejectedValidateFail, "contract API", api.ID)
 	}
 
 	err = dh.database.UpsertContractAPI(ctx, api)
@@ -84,7 +84,7 @@ func (dh *definitionHandler) handleFFIDefinition(ctx context.Context, state *cor
 	l := log.L(ctx)
 	ffi.Namespace = dh.namespace.Name
 	if err := ffi.Validate(ctx, true); err != nil {
-		return HandlerResult{Action: ActionReject}, i18n.NewError(ctx, coremsgs.MsgDefRejectedValidateFail, "contract interface", ffi.ID, err)
+		return HandlerResult{Action: ActionReject}, i18n.WrapError(ctx, err, coremsgs.MsgDefRejectedValidateFail, "contract interface", ffi.ID)
 	}
 
 	if retry, err := dh.persistFFI(ctx, ffi); err != nil {
@@ -115,7 +115,7 @@ func (dh *definitionHandler) handleContractAPIDefinition(ctx context.Context, st
 	l := log.L(ctx)
 	api.Namespace = dh.namespace.Name
 	if err := api.Validate(ctx, true); err != nil {
-		return HandlerResult{Action: ActionReject}, i18n.NewError(ctx, coremsgs.MsgDefRejectedValidateFail, "contract API", api.ID, err)
+		return HandlerResult{Action: ActionReject}, i18n.WrapError(ctx, err, coremsgs.MsgDefRejectedValidateFail, "contract API", api.ID)
 	}
 
 	if retry, err := dh.persistContractAPI(ctx, httpServerURL, api); err != nil {
diff --git a/internal/definitions/handler_identity_update.go b/internal/definitions/handler_identity_update.go
index 511cae7ed..bb3947f04 100644
--- a/internal/definitions/handler_identity_update.go
+++ b/internal/definitions/handler_identity_update.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -45,7 +45,7 @@ func (dh *definitionHandler) handleIdentityUpdateBroadcast(ctx context.Context,
 
 func (dh *definitionHandler) handleIdentityUpdate(ctx context.Context, state *core.BatchState, msg *identityUpdateMsgInfo, update *core.IdentityUpdate) (HandlerResult, error) {
 	if err := update.Identity.Validate(ctx); err != nil {
-		return HandlerResult{Action: ActionReject}, i18n.NewError(ctx, coremsgs.MsgDefRejectedValidateFail, "identity update", update.Identity.ID, err)
+		return HandlerResult{Action: ActionReject}, i18n.WrapError(ctx, err, coremsgs.MsgDefRejectedValidateFail, "identity update", update.Identity.ID)
 	}
 
 	// Get the existing identity (must be a confirmed identity at the point an update is issued)
diff --git a/internal/definitions/handler_identity_verification.go b/internal/definitions/handler_identity_verification.go
index 834108fe6..6ada77d45 100644
--- a/internal/definitions/handler_identity_verification.go
+++ b/internal/definitions/handler_identity_verification.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -33,7 +33,7 @@ func (dh *definitionHandler) handleIdentityVerificationBroadcast(ctx context.Con
 	verification.Identity.Namespace = dh.namespace.Name
 	err := verification.Identity.Validate(ctx)
 	if err != nil || verification.Identity.Parent == nil || verification.Claim.ID == nil || verification.Claim.Hash == nil {
-		return HandlerResult{Action: ActionReject}, i18n.NewError(ctx, coremsgs.MsgDefRejectedValidateFail, "identity verification", verifyMsg.Header.ID, err)
+		return HandlerResult{Action: ActionReject}, i18n.WrapError(ctx, err, coremsgs.MsgDefRejectedValidateFail, "identity verification", verifyMsg.Header.ID)
 	}
 
 	// Check the verification is signed by the correct org
diff --git a/internal/definitions/handler_tokenpool.go b/internal/definitions/handler_tokenpool.go
index 0fa5f3e1e..53d016de5 100644
--- a/internal/definitions/handler_tokenpool.go
+++ b/internal/definitions/handler_tokenpool.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -52,7 +52,7 @@ func (dh *definitionHandler) handleTokenPoolDefinition(ctx context.Context, stat
 
 	pool.Namespace = dh.namespace.Name
 	if err := pool.Validate(ctx); err != nil {
-		return HandlerResult{Action: ActionReject, CustomCorrelator: correlator}, i18n.NewError(ctx, coremsgs.MsgDefRejectedValidateFail, "token pool", pool.ID, err)
+		return HandlerResult{Action: ActionReject, CustomCorrelator: correlator}, i18n.WrapError(ctx, err, coremsgs.MsgDefRejectedValidateFail, "token pool", pool.ID)
 	}
 
 	// Check if pool has already been confirmed on chain (and confirm the message if so)
diff --git a/internal/definitions/sender_tokenpool.go b/internal/definitions/sender_tokenpool.go
index a5172419d..b3df2b5a5 100644
--- a/internal/definitions/sender_tokenpool.go
+++ b/internal/definitions/sender_tokenpool.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -18,6 +18,7 @@ package definitions
 
 import (
 	"context"
+	"errors"
 
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly-common/pkg/log"
@@ -26,15 +27,16 @@ import (
 )
 
 func (bm *definitionSender) DefineTokenPool(ctx context.Context, pool *core.TokenPoolAnnouncement, waitConfirm bool) error {
-	// Map token connector name -> broadcast name
-	if broadcastName, exists := bm.tokenBroadcastNames[pool.Pool.Connector]; exists {
-		pool.Pool.Connector = broadcastName
-	} else {
-		log.L(ctx).Infof("Could not find broadcast name for token connector: %s", pool.Pool.Connector)
-		return i18n.NewError(ctx, coremsgs.MsgInvalidConnectorName, broadcastName, "token")
-	}
 
 	if bm.multiparty {
+		// Map token connector name -> broadcast name
+		if broadcastName, exists := bm.tokenBroadcastNames[pool.Pool.Connector]; exists {
+			pool.Pool.Connector = broadcastName
+		} else {
+			log.L(ctx).Infof("Could not find broadcast name for token connector: %s", pool.Pool.Connector)
+			return i18n.NewError(ctx, coremsgs.MsgInvalidConnectorName, broadcastName, "token")
+		}
+
 		if err := pool.Pool.Validate(ctx); err != nil {
 			return err
 		}
@@ -49,6 +51,12 @@ func (bm *definitionSender) DefineTokenPool(ctx context.Context, pool *core.Toke
 	}
 
 	return fakeBatch(ctx, func(ctx context.Context, state *core.BatchState) (HandlerResult, error) {
-		return bm.handler.handleTokenPoolDefinition(ctx, state, pool.Pool)
+		hr, err := bm.handler.handleTokenPoolDefinition(ctx, state, pool.Pool)
+		if err != nil {
+			if innerErr := errors.Unwrap(err); innerErr != nil {
+				return hr, innerErr
+			}
+		}
+		return hr, err
 	})
 }
diff --git a/internal/events/event_poller.go b/internal/events/event_poller.go
index 6ffcc4a9b..9d876b6c8 100644
--- a/internal/events/event_poller.go
+++ b/internal/events/event_poller.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -283,9 +283,15 @@ func (ep *eventPoller) shoulderTap() {
 func (ep *eventPoller) waitForShoulderTapOrPollTimeout(lastEventCount int) bool {
 	l := log.L(ep.ctx)
 	longTimeoutDuration := ep.conf.eventPollTimeout
+
+	if lastEventCount >= ep.conf.eventBatchSize {
+		l.Tracef("Polling immediately due to full previous event batch")
+		return true
+	}
+
 	// For throughput optimized environments, we can set an eventBatchingTimeout to allow messages to arrive
 	// between polling cycles (at the cost of some dispatch latency)
-	if ep.conf.eventBatchTimeout > 0 && lastEventCount > 0 && lastEventCount < ep.conf.eventBatchSize {
+	if ep.conf.eventBatchTimeout > 0 && lastEventCount > 0 {
 		shortTimeout := time.NewTimer(ep.conf.eventBatchTimeout)
 		select {
 		case <-shortTimeout.C:
diff --git a/internal/events/persist_batch.go b/internal/events/persist_batch.go
index 7a9b5131f..5a545d89d 100644
--- a/internal/events/persist_batch.go
+++ b/internal/events/persist_batch.go
@@ -69,13 +69,9 @@ func (em *eventManager) persistBatch(ctx context.Context, batch *core.Batch) (pe
 		}
 	}
 
-	// Upsert the batch
-	err = em.database.UpsertBatch(ctx, persistedBatch)
+	// Insert the batch
+	existing, err := em.database.InsertOrGetBatch(ctx, persistedBatch)
 	if err != nil {
-		if err == database.HashMismatch {
-			l.Errorf("Invalid batch '%s'. Batch hash mismatch with existing record", batch.ID)
-			return nil, false, nil // This is not retryable. skip this batch
-		}
 		l.Errorf("Failed to insert batch '%s': %s", batch.ID, err)
 		return nil, false, err // a persistence failure here is considered retryable (so returned)
 	}
@@ -84,6 +80,11 @@ func (em *eventManager) persistBatch(ctx context.Context, batch *core.Batch) (pe
 	if err != nil || !valid {
 		return nil, valid, err
 	}
+
+	if existing != nil {
+		l.Infof("Skipped insert of batch '%s' (already exists)", batch.ID)
+		return existing, true, nil
+	}
 	em.aggregator.cacheBatch(em.aggregator.getBatchCacheKey(persistedBatch.ID, persistedBatch.Hash), persistedBatch, manifest)
 	return persistedBatch, true, err
 }
diff --git a/internal/orchestrator/orchestrator.go b/internal/orchestrator/orchestrator.go
index c744e5afc..60e3ab26d 100644
--- a/internal/orchestrator/orchestrator.go
+++ b/internal/orchestrator/orchestrator.go
@@ -552,7 +552,7 @@ func (or *orchestrator) Authorize(ctx context.Context, authReq *fftypes.AuthReq)
 func (or *orchestrator) RewindPins(ctx context.Context, rewind *core.PinRewind) (*core.PinRewind, error) {
 	if rewind.Sequence > 0 {
 		fb := database.PinQueryFactory.NewFilter(ctx)
-		if pins, _, err := or.GetPins(ctx, fb.And(fb.Eq("seq", rewind.Sequence))); err != nil {
+		if pins, _, err := or.GetPins(ctx, fb.And(fb.Eq("sequence", rewind.Sequence))); err != nil {
 			return nil, err
 		} else if len(pins) > 0 {
 			rewind.Batch = pins[0].Batch
diff --git a/mocks/databasemocks/plugin.go b/mocks/databasemocks/plugin.go
index 581e14516..3ca95b7d7 100644
--- a/mocks/databasemocks/plugin.go
+++ b/mocks/databasemocks/plugin.go
@@ -2450,6 +2450,32 @@ func (_m *Plugin) InsertOperation(ctx context.Context, operation *core.Operation
 	return r0
 }
 
+// InsertOrGetBatch provides a mock function with given fields: ctx, data
+func (_m *Plugin) InsertOrGetBatch(ctx context.Context, data *core.BatchPersisted) (*core.BatchPersisted, error) {
+	ret := _m.Called(ctx, data)
+
+	var r0 *core.BatchPersisted
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *core.BatchPersisted) (*core.BatchPersisted, error)); ok {
+		return rf(ctx, data)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *core.BatchPersisted) *core.BatchPersisted); ok {
+		r0 = rf(ctx, data)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*core.BatchPersisted)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *core.BatchPersisted) error); ok {
+		r1 = rf(ctx, data)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // InsertOrGetBlockchainEvent provides a mock function with given fields: ctx, event
 func (_m *Plugin) InsertOrGetBlockchainEvent(ctx context.Context, event *core.BlockchainEvent) (*core.BlockchainEvent, error) {
 	ret := _m.Called(ctx, event)
@@ -2733,20 +2759,6 @@ func (_m *Plugin) UpdateTransaction(ctx context.Context, namespace string, id *f
 	return r0
 }
 
-// UpsertBatch provides a mock function with given fields: ctx, data
-func (_m *Plugin) UpsertBatch(ctx context.Context, data *core.BatchPersisted) error {
-	ret := _m.Called(ctx, data)
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, *core.BatchPersisted) error); ok {
-		r0 = rf(ctx, data)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
 // UpsertContractAPI provides a mock function with given fields: ctx, cd
 func (_m *Plugin) UpsertContractAPI(ctx context.Context, cd *core.ContractAPI) error {
 	ret := _m.Called(ctx, cd)
diff --git a/pkg/database/plugin.go b/pkg/database/plugin.go
index 1f5bc3456..51c41f66f 100644
--- a/pkg/database/plugin.go
+++ b/pkg/database/plugin.go
@@ -139,10 +139,10 @@ type iDataCollection interface {
 }
 
 type iBatchCollection interface {
-	// UpsertBatch - Upsert a batch - the hash cannot change
-	UpsertBatch(ctx context.Context, data *core.BatchPersisted) (err error)
+	// InsertBatch - Insert a new batch, or retrieve the existing one if it has already been recorded
+	InsertOrGetBatch(ctx context.Context, data *core.BatchPersisted) (existing *core.BatchPersisted, err error)
 
-	// UpdateBatch - Update data
+	// UpdateBatch - Update a batch
 	UpdateBatch(ctx context.Context, namespace string, id *fftypes.UUID, update ffapi.Update) (err error)
 
 	// GetBatchByID - Get a batch by ID


Test patch:
diff --git a/internal/batch/batch_manager_test.go b/internal/batch/batch_manager_test.go
index f563cd76a..407c00466 100644
--- a/internal/batch/batch_manager_test.go
+++ b/internal/batch/batch_manager_test.go
@@ -6,7 +6,7 @@
 // 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
+//	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,
@@ -132,7 +132,7 @@ func TestE2EDispatchBroadcast(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("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 	mdi.On("UpdateBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	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)
@@ -255,7 +255,7 @@ func TestE2EDispatchPrivateUnpinned(t *testing.T) {
 	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, "ns1", mock.Anything, mock.Anything).Return(nil) // pins
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 	mdi.On("UpdateBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything, mock.Anything).Return(nil)
 	rag.RunFn = func(a mock.Arguments) {
@@ -441,7 +441,7 @@ func TestMessageSequencerUpdateMessagesFail(t *testing.T) {
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 	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("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 	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) {
@@ -538,7 +538,7 @@ func TestMessageSequencerUpdateBatchFail(t *testing.T) {
 	}
 	mdi.On("GetMessageIDs", mock.Anything, "ns1", mock.Anything).Return([]*core.IDAndSequence{{ID: *msg.Header.ID}}, nil)
 	mdm.On("GetMessageWithDataCached", mock.Anything, mock.Anything).Return(msg, core.DataArray{{ID: dataID}}, true, nil)
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(fmt.Errorf("fizzle"))
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, 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 6e9782325..8d692bb38 100644
--- a/internal/batch/batch_processor_test.go
+++ b/internal/batch/batch_processor_test.go
@@ -6,7 +6,7 @@
 // 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
+//	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,
@@ -85,7 +85,7 @@ func TestUnfilledBatch(t *testing.T) {
 
 	mockRunAsGroupPassthrough(mdi)
 	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("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 
 	mth := bp.txHelper.(*txcommonmocks.Helper)
 	mth.On("SubmitNewTransaction", mock.Anything, core.TransactionTypeBatchPin, core.IdempotencyKey("")).Return(fftypes.NewUUID(), nil)
@@ -134,7 +134,7 @@ func TestBatchSizeOverflow(t *testing.T) {
 	bp.conf.BatchMaxBytes = batchSizeEstimateBase + (&core.Message{}).EstimateSize(false) + 100
 	mockRunAsGroupPassthrough(mdi)
 	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("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 
 	mth := bp.txHelper.(*txcommonmocks.Helper)
 	mth.On("SubmitNewTransaction", mock.Anything, core.TransactionTypeBatchPin, core.IdempotencyKey("")).Return(fftypes.NewUUID(), nil)
@@ -419,7 +419,7 @@ func TestMarkMessageDispatchedUnpinnedOK(t *testing.T) {
 
 	mockRunAsGroupPassthrough(mdi)
 	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("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(fmt.Errorf("pop")).Once()
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil)
 
@@ -476,7 +476,7 @@ func TestMaskContextsRetryAfterPinsAssigned(t *testing.T) {
 		return dbNonce.Nonce == 12347 // twice incremented
 	})).Return(nil).Once()
 	mdi.On("UpdateMessage", mock.Anything, "ns1", mock.Anything, mock.Anything).Return(nil).Twice()
-	mdi.On("UpsertBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything, mock.Anything).Return(nil, nil)
 
 	mdm := bp.data.(*datamocks.Manager)
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
diff --git a/internal/database/sqlcommon/batch_sql_test.go b/internal/database/sqlcommon/batch_sql_test.go
index 15f462d20..76d4a946d 100644
--- a/internal/database/sqlcommon/batch_sql_test.go
+++ b/internal/database/sqlcommon/batch_sql_test.go
@@ -63,10 +63,10 @@ func TestBatch2EWithDB(t *testing.T) {
 	}
 
 	s.callbacks.On("UUIDCollectionNSEvent", database.CollectionBatches, core.ChangeEventTypeCreated, "ns1", batchID, mock.Anything).Return()
-	s.callbacks.On("UUIDCollectionNSEvent", database.CollectionBatches, core.ChangeEventTypeUpdated, "ns1", batchID, mock.Anything).Return()
 
-	err := s.UpsertBatch(ctx, batch)
+	existing, err := s.InsertOrGetBatch(ctx, batch)
 	assert.NoError(t, err)
+	assert.Nil(t, existing)
 
 	// Check we get the exact same batch back
 	batchRead, err := s.GetBatchByID(ctx, "ns1", batchID)
@@ -76,58 +76,17 @@ func TestBatch2EWithDB(t *testing.T) {
 	batchReadJson, _ := json.Marshal(&batchRead)
 	assert.Equal(t, string(batchJson), string(batchReadJson))
 
-	// Update the batch (this is testing what's possible at the database layer,
-	// and does not account for the verification that happens at the higher level)
-	txid := fftypes.NewUUID()
-	msgID2 := fftypes.NewUUID()
-	batchUpdated := &core.BatchPersisted{
-		BatchHeader: core.BatchHeader{
-			ID:   batchID,
-			Type: core.BatchTypePrivate,
-			SignerRef: core.SignerRef{
-				Key:    "0x12345",
-				Author: "did:firefly:org/abcd",
-			},
-			Namespace: "ns1",
-			Node:      fftypes.NewUUID(),
-			Created:   fftypes.Now(),
-		},
-		Hash: fftypes.NewRandB32(),
-		TX: core.TransactionRef{
-			ID:   txid,
-			Type: core.TransactionTypeBatchPin,
-		},
-		Manifest: fftypes.JSONAnyPtr((&core.BatchManifest{
-			Messages: []*core.MessageManifestEntry{
-				{MessageRef: core.MessageRef{ID: msgID1}},
-				{MessageRef: core.MessageRef{ID: msgID2}},
-			},
-		}).String()),
-		Confirmed: fftypes.Now(),
-	}
-
-	// Rejects hash change
-	err = s.UpsertBatch(context.Background(), batchUpdated)
-	assert.Equal(t, database.HashMismatch, err)
-
-	batchUpdated.Hash = batch.Hash
-	err = s.UpsertBatch(context.Background(), batchUpdated)
+	// Try to insert again - should get back the existing row
+	existing, err = s.InsertOrGetBatch(ctx, batch)
 	assert.NoError(t, err)
-
-	// Check we get the exact same message back - note the removal of one of the batch elements
-	batchRead, err = s.GetBatchByID(ctx, "ns1", batchID)
-	assert.NoError(t, err)
-	batchJson, _ = json.Marshal(&batchUpdated)
-	batchReadJson, _ = json.Marshal(&batchRead)
-	assert.Equal(t, string(batchJson), string(batchReadJson))
+	assert.NotNil(t, existing)
 
 	// Query back the batch
 	fb := database.BatchQueryFactory.NewFilter(ctx)
 	filter := fb.And(
-		fb.Eq("id", batchUpdated.ID.String()),
-		fb.Eq("author", batchUpdated.Author),
+		fb.Eq("id", batch.ID.String()),
+		fb.Eq("author", batch.Author),
 		fb.Gt("created", "0"),
-		fb.Gt("confirmed", "0"),
 	)
 	batches, _, err := s.GetBatches(ctx, "ns1", filter)
 	assert.NoError(t, err)
@@ -137,7 +96,7 @@ func TestBatch2EWithDB(t *testing.T) {
 
 	// Negative test on filter
 	filter = fb.And(
-		fb.Eq("id", batchUpdated.ID.String()),
+		fb.Eq("id", batch.ID.String()),
 		fb.Eq("created", "0"),
 	)
 	batches, _, err = s.GetBatches(ctx, "ns1", filter)
@@ -152,7 +111,7 @@ func TestBatch2EWithDB(t *testing.T) {
 
 	// Test find updated value
 	filter = fb.And(
-		fb.Eq("id", batchUpdated.ID.String()),
+		fb.Eq("id", batch.ID.String()),
 		fb.Eq("author", author2),
 	)
 	batches, res, err := s.GetBatches(ctx, "ns1", filter.Count(true))
@@ -166,55 +125,30 @@ func TestBatch2EWithDB(t *testing.T) {
 func TestUpsertBatchFailBegin(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
-	err := s.UpsertBatch(context.Background(), &core.BatchPersisted{})
+	_, err := s.InsertOrGetBatch(context.Background(), &core.BatchPersisted{})
 	assert.Regexp(t, "FF00175", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
-func TestUpsertBatchFailSelect(t *testing.T) {
-	s, mock := newMockProvider().init()
-	mock.ExpectBegin()
-	mock.ExpectQuery("SELECT .*").WillReturnError(fmt.Errorf("pop"))
-	mock.ExpectRollback()
-	batchID := fftypes.NewUUID()
-	err := s.UpsertBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}})
-	assert.Regexp(t, "FF00176", err)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
 func TestUpsertBatchFailInsert(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin()
-	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{}))
 	mock.ExpectExec("INSERT .*").WillReturnError(fmt.Errorf("pop"))
+	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{}))
 	mock.ExpectRollback()
 	batchID := fftypes.NewUUID()
-	err := s.UpsertBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}})
+	_, err := s.InsertOrGetBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}})
 	assert.Regexp(t, "FF00177", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
 
-func TestUpsertBatchFailUpdate(t *testing.T) {
-	s, mock := newMockProvider().init()
-	batchID := fftypes.NewUUID()
-	hash := fftypes.NewRandB32()
-	mock.ExpectBegin()
-	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{"hash"}).AddRow(hash))
-	mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
-	mock.ExpectRollback()
-	err := s.UpsertBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}, Hash: hash})
-	assert.Regexp(t, "FF00178", err)
-	assert.NoError(t, mock.ExpectationsWereMet())
-}
-
 func TestUpsertBatchFailCommit(t *testing.T) {
 	s, mock := newMockProvider().init()
 	batchID := fftypes.NewUUID()
 	mock.ExpectBegin()
-	mock.ExpectQuery("SELECT .*").WillReturnRows(sqlmock.NewRows([]string{"id"}))
 	mock.ExpectExec("INSERT .*").WillReturnResult(sqlmock.NewResult(1, 1))
 	mock.ExpectCommit().WillReturnError(fmt.Errorf("pop"))
-	err := s.UpsertBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}})
+	_, err := s.InsertOrGetBatch(context.Background(), &core.BatchPersisted{BatchHeader: core.BatchHeader{ID: batchID}})
 	assert.Regexp(t, "FF00180", err)
 	assert.NoError(t, mock.ExpectationsWereMet())
 }
diff --git a/internal/database/sqlcommon/blob_sql_test.go b/internal/database/sqlcommon/blob_sql_test.go
index 3c8df1fb2..5324fda69 100644
--- a/internal/database/sqlcommon/blob_sql_test.go
+++ b/internal/database/sqlcommon/blob_sql_test.go
@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -124,9 +124,10 @@ func TestInsertBlobsBeginFail(t *testing.T) {
 }
 
 func TestInsertBlobsMultiRowOK(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 
 	blob1 := &core.Blob{Hash: fftypes.NewRandB32(), PayloadRef: "pay1"}
 	blob2 := &core.Blob{Hash: fftypes.NewRandB32(), PayloadRef: "pay2"}
@@ -144,9 +145,10 @@ func TestInsertBlobsMultiRowOK(t *testing.T) {
 }
 
 func TestInsertBlobsMultiRowFail(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 	blob1 := &core.Blob{Hash: fftypes.NewRandB32(), PayloadRef: "pay1"}
 	mock.ExpectBegin()
 	mock.ExpectQuery("INSERT.*").WillReturnError(fmt.Errorf("pop"))
diff --git a/internal/database/sqlcommon/data_sql_test.go b/internal/database/sqlcommon/data_sql_test.go
index 6162a09ef..a628ba539 100644
--- a/internal/database/sqlcommon/data_sql_test.go
+++ b/internal/database/sqlcommon/data_sql_test.go
@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -240,9 +240,10 @@ func TestInsertDataArrayBeginFail(t *testing.T) {
 }
 
 func TestInsertDataArrayMultiRowOK(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 
 	data1 := &core.Data{ID: fftypes.NewUUID(), Namespace: "ns1"}
 	data2 := &core.Data{ID: fftypes.NewUUID(), Namespace: "ns1"}
@@ -262,9 +263,10 @@ func TestInsertDataArrayMultiRowOK(t *testing.T) {
 }
 
 func TestInsertDataArrayMultiRowFail(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 	data1 := &core.Data{ID: fftypes.NewUUID(), Namespace: "ns1"}
 	mock.ExpectBegin()
 	mock.ExpectQuery("INSERT.*").WillReturnError(fmt.Errorf("pop"))
diff --git a/internal/database/sqlcommon/event_sql_test.go b/internal/database/sqlcommon/event_sql_test.go
index 59d4264a2..134460deb 100644
--- a/internal/database/sqlcommon/event_sql_test.go
+++ b/internal/database/sqlcommon/event_sql_test.go
@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -133,9 +133,10 @@ func TestInsertEventFailCommit(t *testing.T) {
 }
 
 func TestInsertEventsPreCommitMultiRowOK(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 
 	ev1 := &core.Event{ID: fftypes.NewUUID(), Namespace: "ns1"}
 	ev2 := &core.Event{ID: fftypes.NewUUID(), Namespace: "ns1"}
@@ -162,9 +163,10 @@ func TestInsertEventsPreCommitMultiRowOK(t *testing.T) {
 }
 
 func TestInsertEventsPreCommitMultiRowFail(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 	ev1 := &core.Event{ID: fftypes.NewUUID(), Namespace: "ns1"}
 	mock.ExpectBegin()
 	mock.ExpectExec("<acquire lock ns1>").WillReturnResult(driver.ResultNoRows)
diff --git a/internal/database/sqlcommon/message_sql_test.go b/internal/database/sqlcommon/message_sql_test.go
index d906b620f..a0a2604e9 100644
--- a/internal/database/sqlcommon/message_sql_test.go
+++ b/internal/database/sqlcommon/message_sql_test.go
@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -318,9 +318,10 @@ func TestInsertMessagesBeginFail(t *testing.T) {
 }
 
 func TestInsertMessagesMultiRowOK(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 
 	msg1 := &core.Message{Header: core.MessageHeader{ID: fftypes.NewUUID(), Namespace: "ns1"}, LocalNamespace: "ns1", Data: core.DataRefs{{ID: fftypes.NewUUID()}}}
 	msg2 := &core.Message{Header: core.MessageHeader{ID: fftypes.NewUUID(), Namespace: "ns1"}, LocalNamespace: "ns1", Data: core.DataRefs{{ID: fftypes.NewUUID()}}}
@@ -348,9 +349,10 @@ func TestInsertMessagesMultiRowOK(t *testing.T) {
 }
 
 func TestInsertMessagesMultiRowDataRefsFail(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 
 	msg1 := &core.Message{Header: core.MessageHeader{ID: fftypes.NewUUID(), Namespace: "ns1"}, Data: core.DataRefs{{ID: fftypes.NewUUID()}}}
 
@@ -364,9 +366,10 @@ func TestInsertMessagesMultiRowDataRefsFail(t *testing.T) {
 }
 
 func TestInsertMessagesMultiRowFail(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 	msg1 := &core.Message{Header: core.MessageHeader{ID: fftypes.NewUUID(), Namespace: "ns1"}}
 	mock.ExpectBegin()
 	mock.ExpectQuery("INSERT.*").WillReturnError(fmt.Errorf("pop"))
diff --git a/internal/database/sqlcommon/pin_sql_test.go b/internal/database/sqlcommon/pin_sql_test.go
index 321042b5b..17749cf90 100644
--- a/internal/database/sqlcommon/pin_sql_test.go
+++ b/internal/database/sqlcommon/pin_sql_test.go
@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -146,9 +146,10 @@ func TestInsertPinsBeginFail(t *testing.T) {
 }
 
 func TestInsertPinsMultiRowOK(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 
 	pin1 := &core.Pin{Namespace: "ns1", Hash: fftypes.NewRandB32()}
 	pin2 := &core.Pin{Namespace: "ns1", Hash: fftypes.NewRandB32()}
@@ -168,9 +169,10 @@ func TestInsertPinsMultiRowOK(t *testing.T) {
 }
 
 func TestInsertPinsMultiRowFail(t *testing.T) {
-	s, mock := newMockProvider().init()
-	s.features.MultiRowInsert = true
+	s := newMockProvider()
+	s.multiRowInsert = true
 	s.fakePSQLInsert = true
+	s, mock := s.init()
 	pin1 := &core.Pin{Hash: fftypes.NewRandB32()}
 	mock.ExpectBegin()
 	mock.ExpectQuery("INSERT.*").WillReturnError(fmt.Errorf("pop"))
diff --git a/internal/database/sqlcommon/provider_mock_test.go b/internal/database/sqlcommon/provider_mock_test.go
index b3cdf5176..38faa5eb9 100644
--- a/internal/database/sqlcommon/provider_mock_test.go
+++ b/internal/database/sqlcommon/provider_mock_test.go
@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -45,6 +45,7 @@ type mockProvider struct {
 	openError               error
 	getMigrationDriverError error
 	individualSort          bool
+	multiRowInsert          bool
 }
 
 func newMockProvider() *mockProvider {
@@ -59,6 +60,7 @@ func newMockProvider() *mockProvider {
 	mp.SQLCommon.InitConfig(mp, mp.config)
 	mp.config.Set(SQLConfMaxConnections, 10)
 	mp.mockDB, mp.mdb, _ = sqlmock.New()
+	mp.multiRowInsert = false
 	return mp
 }
 
@@ -84,6 +86,7 @@ func (mp *mockProvider) MigrationsDir() string {
 func (psql *mockProvider) Features() dbsql.SQLFeatures {
 	features := dbsql.DefaultSQLProviderFeatures()
 	features.UseILIKE = true
+	features.MultiRowInsert = psql.multiRowInsert
 	features.AcquireLock = func(lockName string) string {
 		return fmt.Sprintf(`<acquire lock %s>`, lockName)
 	}
diff --git a/internal/definitions/sender_tokenpool_test.go b/internal/definitions/sender_tokenpool_test.go
index dc40392de..918916bf7 100644
--- a/internal/definitions/sender_tokenpool_test.go
+++ b/internal/definitions/sender_tokenpool_test.go
@@ -1,4 +1,4 @@
-// Copyright © 2021 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -75,7 +75,32 @@ func TestBroadcastTokenPoolInvalidNonMultiparty(t *testing.T) {
 	}
 
 	err := ds.DefineTokenPool(context.Background(), pool, false)
-	assert.Regexp(t, "FF10420", err)
+	assert.Regexp(t, "FF00140", err)
+
+	mdm.AssertExpectations(t)
+}
+
+func TestBroadcastTokenPoolInvalidNameMultiparty(t *testing.T) {
+	ds, cancel := newTestDefinitionSender(t)
+	defer cancel()
+	ds.multiparty = true
+
+	mdm := ds.data.(*datamocks.Manager)
+
+	pool := &core.TokenPoolAnnouncement{
+		Pool: &core.TokenPool{
+			ID:        fftypes.NewUUID(),
+			Namespace: "",
+			Name:      "",
+			Type:      core.TokenTypeNonFungible,
+			Locator:   "N1",
+			Symbol:    "COIN",
+			Connector: "connector1",
+		},
+	}
+
+	err := ds.DefineTokenPool(context.Background(), pool, false)
+	assert.Regexp(t, "FF00140", err)
 
 	mdm.AssertExpectations(t)
 }
@@ -115,6 +140,37 @@ func TestDefineTokenPoolOk(t *testing.T) {
 	mms.AssertExpectations(t)
 }
 
+func TestDefineTokenPoolkONonMultiparty(t *testing.T) {
+	ds, cancel := newTestDefinitionSender(t)
+	defer cancel()
+	ds.multiparty = false
+
+	mdm := ds.data.(*datamocks.Manager)
+	mdb := ds.database.(*databasemocks.Plugin)
+
+	pool := &core.TokenPool{
+		ID:        fftypes.NewUUID(),
+		Namespace: "ns1",
+		Name:      "mypool",
+		Type:      core.TokenTypeNonFungible,
+		Locator:   "N1",
+		Symbol:    "COIN",
+		Connector: "connector1",
+		State:     core.TokenPoolStateConfirmed,
+	}
+	poolAnnouncement := &core.TokenPoolAnnouncement{
+		Pool: pool,
+	}
+
+	mdb.On("GetTokenPoolByID", mock.Anything, mock.Anything, mock.Anything).Return(pool, nil)
+
+	err := ds.DefineTokenPool(context.Background(), poolAnnouncement, false)
+	assert.NoError(t, err)
+
+	mdm.AssertExpectations(t)
+	mdb.AssertExpectations(t)
+}
+
 func TestDefineTokenPoolNonMultipartyTokenPoolFail(t *testing.T) {
 	ds, cancel := newTestDefinitionSender(t)
 	defer cancel()
diff --git a/internal/events/batch_pin_complete_test.go b/internal/events/batch_pin_complete_test.go
index 755792ab6..5fb4a6b22 100644
--- a/internal/events/batch_pin_complete_test.go
+++ b/internal/events/batch_pin_complete_test.go
@@ -469,20 +469,6 @@ func TestPersistBatchMismatchChainHash(t *testing.T) {
 	assert.False(t, valid)
 }
 
-func TestPersistBatchUpsertBatchMismatchHash(t *testing.T) {
-	em := newTestEventManager(t)
-	defer em.cleanup(t)
-	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
-	batch := sampleBatch(t, core.BatchTypeBroadcast, core.TransactionTypeBatchPin, core.DataArray{data})
-
-	em.mdi.On("UpsertBatch", mock.Anything, mock.Anything).Return(database.HashMismatch)
-
-	bp, valid, err := em.persistBatch(context.Background(), batch)
-	assert.False(t, valid)
-	assert.Nil(t, bp)
-	assert.NoError(t, err)
-}
-
 func TestPersistBatchBadHash(t *testing.T) {
 	em := newTestEventManager(t)
 	defer em.cleanup(t)
@@ -528,7 +514,7 @@ func TestPersistBatchUpsertBatchFail(t *testing.T) {
 	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, core.BatchTypeBroadcast, core.TransactionTypeBatchPin, core.DataArray{data})
 
-	em.mdi.On("UpsertBatch", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	em.mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything).Return(nil, fmt.Errorf("pop"))
 
 	bp, valid, err := em.persistBatch(context.Background(), batch)
 	assert.Nil(t, bp)
@@ -559,7 +545,7 @@ func TestPersistBatchSwallowBadData(t *testing.T) {
 	}
 	batch.Hash = batch.Payload.Hash()
 
-	em.mdi.On("UpsertBatch", mock.Anything, mock.Anything).Return(nil)
+	em.mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything).Return(nil, nil)
 
 	bp, valid, err := em.persistBatch(context.Background(), batch)
 	assert.False(t, valid)
@@ -573,7 +559,7 @@ func TestPersistBatchGoodDataUpsertOptimizeFail(t *testing.T) {
 	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, core.BatchTypeBroadcast, core.TransactionTypeBatchPin, core.DataArray{data})
 
-	em.mdi.On("UpsertBatch", mock.Anything, mock.Anything).Return(nil)
+	em.mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertDataArray", mock.Anything, mock.Anything).Return(fmt.Errorf("optimzation miss"))
 	em.mdi.On("UpsertData", mock.Anything, mock.Anything, database.UpsertOptimizationExisting).Return(fmt.Errorf("pop"))
 
@@ -591,7 +577,7 @@ func TestPersistBatchGoodDataMessageFail(t *testing.T) {
 	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, core.BatchTypeBroadcast, core.TransactionTypeBatchPin, core.DataArray{data})
 
-	em.mdi.On("UpsertBatch", mock.Anything, mock.Anything).Return(nil)
+	em.mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertDataArray", mock.Anything, mock.Anything).Return(nil)
 	em.mdi.On("InsertMessages", mock.Anything, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("optimzation miss"))
 	em.mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationExisting, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("pop"))
@@ -614,7 +600,7 @@ func TestPersistBatchGoodMessageAuthorMismatch(t *testing.T) {
 	batch.Payload.Messages[0].Hash = batch.Payload.Messages[0].Header.Hash()
 	batch.Hash = batch.Payload.Hash()
 
-	em.mdi.On("UpsertBatch", mock.Anything, mock.Anything).Return(nil)
+	em.mdi.On("InsertOrGetBatch", mock.Anything, mock.Anything).Return(nil, nil)
 
 	bp, valid, err := em.persistBatch(context.Background(), batch)
 	assert.Nil(t, bp)
diff --git a/internal/events/dx_callbacks_test.go b/internal/events/dx_callbacks_test.go
index c34529a2f..eacbb763f 100644
--- a/internal/events/dx_callbacks_test.go
+++ b/internal/events/dx_callbacks_test.go
@@ -153,7 +153,7 @@ func TestPinnedReceiveOK(t *testing.T) {
 	em.mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
 	em.mim.On("ValidateNodeOwner", em.ctx, mock.Anything, mock.Anything).Return(true, nil)
 
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
 		args[2].(database.PostCompletionHook)()
@@ -224,7 +224,7 @@ func TestMessageReceivePersistBatchError(t *testing.T) {
 	}).Return(node1, nil)
 	em.mim.On("CachedIdentityLookupMustExist", em.ctx, "signingOrg").Return(org1, false, nil)
 	em.mim.On("ValidateNodeOwner", em.ctx, mock.Anything, mock.Anything).Return(true, nil)
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(fmt.Errorf("pop"))
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, fmt.Errorf("pop"))
 
 	// no ack as we are simulating termination mid retry
 	mde := newMessageReceivedNoAck("peer1", b)
@@ -607,7 +607,7 @@ func TestMessageReceiveMessagePersistMessageFail(t *testing.T) {
 	em.mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
 	em.mim.On("ValidateNodeOwner", em.ctx, mock.Anything, mock.Anything).Return(true, nil)
 
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil)
 	em.mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("optimization fail"))
 	em.mdi.On("UpsertMessage", em.ctx, mock.Anything, database.UpsertOptimizationExisting, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("pop"))
@@ -648,7 +648,7 @@ func TestMessageReceiveMessagePersistDataFail(t *testing.T) {
 	em.mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
 	em.mim.On("ValidateNodeOwner", em.ctx, mock.Anything, mock.Anything).Return(true, nil)
 
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(fmt.Errorf("optimization miss"))
 	em.mdi.On("UpsertData", em.ctx, mock.Anything, database.UpsertOptimizationExisting).Return(fmt.Errorf("pop"))
 
@@ -687,7 +687,7 @@ func TestMessageReceiveUnpinnedBatchOk(t *testing.T) {
 	em.mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
 	em.mim.On("ValidateNodeOwner", em.ctx, mock.Anything, mock.Anything).Return(true, nil)
 
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil)
 	em.mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
 		args[2].(database.PostCompletionHook)()
@@ -730,7 +730,7 @@ func TestMessageReceiveUnpinnedBatchConfirmMessagesFail(t *testing.T) {
 	em.mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
 	em.mim.On("ValidateNodeOwner", em.ctx, mock.Anything, mock.Anything).Return(true, nil)
 
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil)
 	em.mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
 		args[2].(database.PostCompletionHook)()
@@ -773,7 +773,7 @@ func TestMessageReceiveUnpinnedBatchPersistEventFail(t *testing.T) {
 	em.mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
 	em.mim.On("ValidateNodeOwner", em.ctx, mock.Anything, mock.Anything).Return(true, nil)
 
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil)
 	em.mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
 		args[2].(database.PostCompletionHook)()
diff --git a/internal/events/event_poller_test.go b/internal/events/event_poller_test.go
index 68f39047f..9d9366525 100644
--- a/internal/events/event_poller_test.go
+++ b/internal/events/event_poller_test.go
@@ -280,12 +280,21 @@ func TestWaitForShoulderTapOrExitCloseBatch(t *testing.T) {
 	assert.False(t, ep.waitForShoulderTapOrPollTimeout(1))
 }
 
-func TestWaitForShoulderTapOrExitClosePoll(t *testing.T) {
+func TestWaitForShoulderTapImmediateRepollOnFullBatch(t *testing.T) {
 	mdi := &databasemocks.Plugin{}
 	ep, cancel := newTestEventPoller(t, mdi, nil, nil)
 	cancel()
 	ep.conf.eventBatchTimeout = 1 * time.Minute
 	ep.conf.eventBatchSize = 1
+	assert.True(t, ep.waitForShoulderTapOrPollTimeout(1))
+}
+
+func TestWaitForShoulderTapOrExitClosePoll(t *testing.T) {
+	mdi := &databasemocks.Plugin{}
+	ep, cancel := newTestEventPoller(t, mdi, nil, nil)
+	cancel()
+	ep.conf.eventBatchTimeout = 1 * time.Minute
+	ep.conf.eventBatchSize = 2
 	assert.False(t, ep.waitForShoulderTapOrPollTimeout(1))
 }
 
diff --git a/internal/events/persist_batch_test.go b/internal/events/persist_batch_test.go
index 49c142da6..8ce8d2c49 100644
--- a/internal/events/persist_batch_test.go
+++ b/internal/events/persist_batch_test.go
@@ -33,7 +33,7 @@ func TestPersistBatch(t *testing.T) {
 	em := newTestEventManager(t)
 	defer em.cleanup(t)
 
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(fmt.Errorf(("pop")))
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, fmt.Errorf(("pop")))
 
 	org := newTestOrg("org1")
 	orgBytes, err := json.Marshal(&org)
@@ -88,12 +88,33 @@ func TestPersistBatch(t *testing.T) {
 
 }
 
+func TestPersistBatchAlreadyExisting(t *testing.T) {
+
+	em := newTestEventManager(t)
+	defer em.cleanup(t)
+
+	existing := &core.BatchPersisted{}
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(existing, nil)
+	em.mim.On("GetLocalNode", mock.Anything).Return(testNode, nil)
+	em.mdi.On("InsertDataArray", mock.Anything, mock.Anything).Return(nil)
+	em.mdi.On("InsertMessages", em.ctx, mock.Anything, mock.Anything).Return(nil, nil)
+
+	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
+	batch := sampleBatch(t, core.BatchTypeBroadcast, core.TransactionTypeBatchPin, core.DataArray{data})
+
+	result, valid, err := em.persistBatch(em.ctx, batch)
+	assert.True(t, valid)
+	assert.NoError(t, err)
+	assert.Equal(t, existing, result)
+
+}
+
 func TestPersistBatchNoCacheDataNotInBatch(t *testing.T) {
 
 	em := newTestEventManager(t)
 	defer em.cleanup(t)
 
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil)
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, nil)
 
 	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, core.BatchTypeBroadcast, core.TransactionTypeBatchPin, core.DataArray{data})
@@ -113,7 +134,7 @@ func TestPersistBatchExtraDataInBatch(t *testing.T) {
 	em := newTestEventManager(t)
 	defer em.cleanup(t)
 
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil)
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, nil)
 
 	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, core.BatchTypeBroadcast, core.TransactionTypeBatchPin, core.DataArray{data})
diff --git a/internal/events/ss_callbacks_test.go b/internal/events/ss_callbacks_test.go
index 0a56dba9c..de5ab0186 100644
--- a/internal/events/ss_callbacks_test.go
+++ b/internal/events/ss_callbacks_test.go
@@ -39,7 +39,7 @@ func TestSharedStorageBatchDownloadedOk(t *testing.T) {
 	b, _ := json.Marshal(&batch)
 
 	mss := &sharedstoragemocks.Plugin{}
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil, nil)
 	em.mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(nil, nil).Run(func(args mock.Arguments) {
 		args[2].(database.PostCompletionHook)()
@@ -71,7 +71,7 @@ func TestSharedStorageBatchDownloadedPersistFail(t *testing.T) {
 	b, _ := json.Marshal(&batch)
 
 	mss := &sharedstoragemocks.Plugin{}
-	em.mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(fmt.Errorf("pop"))
+	em.mdi.On("InsertOrGetBatch", em.ctx, mock.Anything).Return(nil, fmt.Errorf("pop"))
 	mss.On("Name").Return("utdx").Maybe()
 
 	_, err := em.SharedStorageBatchDownloaded(mss, "payload1", b)
diff --git a/test/e2e/gateway/tokens_only.go b/test/e2e/gateway/tokens_only.go
index 022140312..7fee4f159 100644
--- a/test/e2e/gateway/tokens_only.go
+++ b/test/e2e/gateway/tokens_only.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -71,6 +71,7 @@ func (suite *TokensOnlyTestSuite) TestTokensOnlyNamespaces() {
 
 	// Add the new namespace
 	data1 := e2e.ReadConfig(suite.T(), suite.testState.configFile1)
+	e2e.AddPluginBroadcastName(data1, "tokens", "testremote")
 	e2e.AddNamespace(data1, namespaceInfo)
 	e2e.WriteConfig(suite.T(), suite.testState.configFile1, data1)