Repository: hyperledger/firefly
Pull request #1202 (2023-03-03 21:00:52)

Cleanup around batches and transactions
Some general cleanup identified while working toward [FIR-17](https://github.com/hyperledger/firefly-fir/pull/17)

* Add helper FindOperationInTransaction
* Split internal batch dispatcher state from dispatched payload
* Add IsPinned() transaction helper
* Use message TransactionID in lookup
* Move definition of MultipartyContract fields into blockchain plugin

codecov-commenter: # [Codecov](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) Report
> Merging [#1202](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) (13d7a24) into [main](https://codecov.io/gh/hyperledger/firefly/commit/5b6d237e128185568b77380cecf0eebc835d484e?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) (5b6d237) will **not change** coverage.
> The diff coverage is `100.00%`.

> :exclamation: Current head 13d7a24 differs from pull request most recent head c22884c. Consider uploading reports for the commit c22884c to get more accurate results

```diff
@@            Coverage Diff            @@
##              main     #1202   +/-   ##
=========================================
  Coverage   100.00%   100.00%           
=========================================
  Files          305       305           
  Lines        20074     20083    +9     
=========================================
+ Hits         20074     20083    +9     
```


| [Impacted Files](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger) | Coverage Δ | |
|---|---|---|
| [internal/batch/batch\_manager.go](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvYmF0Y2gvYmF0Y2hfbWFuYWdlci5nbw==) | `100.00% <ø> (ø)` | |
| [internal/multiparty/manager.go](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvbXVsdGlwYXJ0eS9tYW5hZ2VyLmdv) | `100.00% <ø> (ø)` | |
| [internal/assets/token\_pool.go](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvYXNzZXRzL3Rva2VuX3Bvb2wuZ28=) | `100.00% <100.00%> (ø)` | |
| [internal/batch/batch\_processor.go](https://codecov.io/gh/hyperledger/firefly/pull/1202?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/manager.go](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvYnJvYWRjYXN0L21hbmFnZXIuZ28=) | `100.00% <100.00%> (ø)` | |
| [internal/events/dx\_callbacks.go](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZXZlbnRzL2R4X2NhbGxiYWNrcy5nbw==) | `100.00% <100.00%> (ø)` | |
| [internal/events/token\_pool\_created.go](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZXZlbnRzL3Rva2VuX3Bvb2xfY3JlYXRlZC5nbw==) | `100.00% <100.00%> (ø)` | |
| [internal/events/tokens\_approved.go](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZXZlbnRzL3Rva2Vuc19hcHByb3ZlZC5nbw==) | `100.00% <100.00%> (ø)` | |
| [internal/events/tokens\_transferred.go](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvZXZlbnRzL3Rva2Vuc190cmFuc2ZlcnJlZC5nbw==) | `100.00% <100.00%> (ø)` | |
| [internal/orchestrator/data\_query.go](https://codecov.io/gh/hyperledger/firefly/pull/1202?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=hyperledger#diff-aW50ZXJuYWwvb3JjaGVzdHJhdG9yL2RhdGFfcXVlcnkuZ28=) | `100.00% <100.00%> (ø)` | |
| ... and [10 more](https://codecov.io/gh/hyperledger/firefly/pull/1202?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/internal/assets/token_pool.go b/internal/assets/token_pool.go
index 87fd56e99..d7a6101b7 100644
--- a/internal/assets/token_pool.go
+++ b/internal/assets/token_pool.go
@@ -26,7 +26,6 @@ import (
 	"github.com/hyperledger/firefly/internal/coremsgs"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/core"
-	"github.com/hyperledger/firefly/pkg/database"
 )
 
 func (am *assetManager) CreateTokenPool(ctx context.Context, pool *core.TokenPoolInput, waitConfirm bool) (*core.TokenPool, error) {
@@ -112,14 +111,9 @@ func (am *assetManager) ActivateTokenPool(ctx context.Context, pool *core.TokenP
 
 	var op *core.Operation
 	err = am.database.RunAsGroup(ctx, func(ctx context.Context) (err error) {
-		fb := database.OperationQueryFactory.NewFilter(ctx)
-		filter := fb.And(
-			fb.Eq("tx", pool.TX.ID),
-			fb.Eq("type", core.OpTypeTokenActivatePool),
-		)
-		if existing, _, err := am.database.GetOperations(ctx, am.namespace, filter); err != nil {
+		if existing, err := am.txHelper.FindOperationInTransaction(ctx, pool.TX.ID, core.OpTypeTokenActivatePool); err != nil {
 			return err
-		} else if len(existing) > 0 {
+		} else if existing != nil {
 			log.L(ctx).Debugf("Dropping duplicate token pool activation request for pool %s", pool.ID)
 			return nil
 		}
diff --git a/internal/batch/batch_manager.go b/internal/batch/batch_manager.go
index de561ff15..1a05eb0ca 100644
--- a/internal/batch/batch_manager.go
+++ b/internal/batch/batch_manager.go
@@ -117,7 +117,7 @@ type batchManager struct {
 	startupOffsetRetryAttempts int
 }
 
-type DispatchHandler func(context.Context, *DispatchState) error
+type DispatchHandler func(context.Context, *DispatchPayload) error
 
 type DispatcherOptions struct {
 	BatchType      core.BatchType
diff --git a/internal/batch/batch_processor.go b/internal/batch/batch_processor.go
index ba25c04c4..c28de708b 100644
--- a/internal/batch/batch_processor.go
+++ b/internal/batch/batch_processor.go
@@ -97,13 +97,16 @@ type nonceState struct {
 	new    bool
 }
 
-type DispatchState struct {
-	Persisted      core.BatchPersisted
-	Messages       []*core.Message
-	Data           core.DataArray
-	Pins           []*fftypes.Bytes32
-	noncesAssigned map[fftypes.Bytes32]*nonceState
+type dispatchState struct {
 	msgPins        map[fftypes.UUID]fftypes.FFStringArray
+	noncesAssigned map[fftypes.Bytes32]*nonceState
+}
+
+type DispatchPayload struct {
+	Batch    core.BatchPersisted
+	Messages []*core.Message
+	Data     core.DataArray
+	Pins     []*fftypes.Bytes32
 }
 
 const batchSizeEstimateBase = int64(512)
@@ -157,9 +160,9 @@ func (bp *batchProcessor) status() *ProcessorStatus {
 	}
 }
 
-func (bp *batchProcessor) newAssembly(initalWork ...*batchWork) {
+func (bp *batchProcessor) newAssembly(initialWork ...*batchWork) {
 	bp.assemblyID = fftypes.NewUUID()
-	bp.assemblyQueue = append([]*batchWork{}, initalWork...)
+	bp.assemblyQueue = append([]*batchWork{}, initialWork...)
 	bp.assemblyQueueBytes = batchSizeEstimateBase
 }
 
@@ -221,7 +224,7 @@ func (bp *batchProcessor) notifyFlushComplete(flushWork []*batchWork) {
 	bp.bm.notifyFlushed(sequences)
 }
 
-func (bp *batchProcessor) updateFlushStats(state *DispatchState, byteSize int64) {
+func (bp *batchProcessor) updateFlushStats(payload *DispatchPayload, byteSize int64) {
 	bp.statusMux.Lock()
 	defer bp.statusMux.Unlock()
 	fs := &bp.flushStatus
@@ -237,10 +240,10 @@ func (bp *batchProcessor) updateFlushStats(state *DispatchState, byteSize int64)
 	fs.totalBytesFlushed += byteSize
 	fs.AverageBatchBytes = (fs.totalBytesFlushed / fs.TotalBatches)
 
-	fs.totalMessagesFlushed += int64(len(state.Messages))
+	fs.totalMessagesFlushed += int64(len(payload.Messages))
 	fs.AverageBatchMessages = math.Round((float64(fs.totalMessagesFlushed)/float64(fs.TotalBatches))*100) / 100
 
-	fs.totalDataFlushed += int64(len(state.Data))
+	fs.totalDataFlushed += int64(len(payload.Data))
 	fs.AverageBatchData = math.Round((float64(fs.totalDataFlushed)/float64(fs.TotalBatches))*100) / 100
 }
 
@@ -275,8 +278,8 @@ func (bp *batchProcessor) assemblyLoop() {
 
 	var batchTimeout = time.NewTimer(bp.conf.DisposeTimeout)
 	idle := true
-	quescing := false
-	for !quescing {
+	quiescing := false
+	for !quiescing {
 
 		var timedout, full, overflow bool
 		select {
@@ -294,7 +297,7 @@ func (bp *batchProcessor) assemblyLoop() {
 			}
 		case work, ok := <-bp.newWork:
 			if !ok {
-				quescing = true
+				quiescing = true
 			} else {
 				full, overflow = bp.addWork(work)
 				if idle {
@@ -305,7 +308,7 @@ func (bp *batchProcessor) assemblyLoop() {
 				}
 			}
 		}
-		if (full || timedout || quescing) && len(bp.assemblyQueue) > 0 {
+		if (full || timedout || quiescing) && len(bp.assemblyQueue) > 0 {
 			// Let Go GC the old timer
 			_ = batchTimeout.Stop()
 
@@ -324,7 +327,7 @@ func (bp *batchProcessor) assemblyLoop() {
 
 			// If we didn't overflow, then just go back to idle - we don't know if we have more work to come, so
 			// either we'll pop straight away (and move to the batch timeout) or wait for the dispose timeout
-			if !overflow && !quescing {
+			if !overflow && !quiescing {
 				batchTimeout = time.NewTimer(bp.conf.DisposeTimeout)
 				idle = true
 			}
@@ -336,7 +339,7 @@ func (bp *batchProcessor) flush(overflow bool) error {
 	id, flushWork, byteSize := bp.startFlush(overflow)
 
 	log.L(bp.ctx).Debugf("Flushing batch %s", id)
-	state := bp.initFlushState(id, flushWork)
+	state := bp.initPayload(id, flushWork)
 
 	// Sealing phase: assigns persisted pins to messages, and finalizes the manifest
 	err := bp.sealBatch(state)
@@ -370,9 +373,9 @@ func (bp *batchProcessor) flush(overflow bool) error {
 	return nil
 }
 
-func (bp *batchProcessor) initFlushState(id *fftypes.UUID, flushWork []*batchWork) *DispatchState {
-	state := &DispatchState{
-		Persisted: core.BatchPersisted{
+func (bp *batchProcessor) initPayload(id *fftypes.UUID, flushWork []*batchWork) *DispatchPayload {
+	payload := &DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				ID:        id,
 				Type:      bp.conf.DispatcherOptions.BatchType,
@@ -385,22 +388,22 @@ func (bp *batchProcessor) initFlushState(id *fftypes.UUID, flushWork []*batchWor
 	}
 	localNode, err := bp.bm.identity.GetLocalNode(bp.ctx)
 	if err == nil && localNode != nil {
-		state.Persisted.BatchHeader.Node = localNode.ID
+		payload.Batch.BatchHeader.Node = localNode.ID
 	}
 	for _, w := range flushWork {
 		if w.msg != nil {
 			w.msg.BatchID = id
-			state.Messages = append(state.Messages, w.msg.BatchMessage())
+			payload.Messages = append(payload.Messages, w.msg.BatchMessage())
 		}
 		for _, d := range w.data {
 			log.L(bp.ctx).Debugf("Adding data '%s' to batch '%s' for message '%s'", d.ID, id, w.msg.Header.ID)
-			state.Data = append(state.Data, d.BatchData(state.Persisted.Type))
+			payload.Data = append(payload.Data, d.BatchData(payload.Batch.Type))
 		}
 	}
-	return state
+	return payload
 }
 
-func (bp *batchProcessor) getNextNonce(ctx context.Context, state *DispatchState, nonceKeyHash *fftypes.Bytes32, contextHash *fftypes.Bytes32) (int64, error) {
+func (bp *batchProcessor) getNextNonce(ctx context.Context, state *dispatchState, nonceKeyHash *fftypes.Bytes32, contextHash *fftypes.Bytes32) (int64, error) {
 
 	// See if the nonceKeyHash is in our cached state already
 	if cached, ok := state.noncesAssigned[*nonceKeyHash]; ok {
@@ -433,7 +436,7 @@ func (bp *batchProcessor) getNextNonce(ctx context.Context, state *DispatchState
 	return nonceState.latest, nil
 }
 
-func (bp *batchProcessor) maskContext(ctx context.Context, state *DispatchState, msg *core.Message, topic string) (msgPinString string, contextOrPin *fftypes.Bytes32, err error) {
+func (bp *batchProcessor) maskContext(ctx context.Context, state *dispatchState, msg *core.Message, topic string) (msgPinString string, contextOrPin *fftypes.Bytes32, err error) {
 
 	hashBuilder := sha256.New()
 	hashBuilder.Write([]byte(topic))
@@ -476,9 +479,11 @@ func (bp *batchProcessor) maskContext(ctx context.Context, state *DispatchState,
 	return pinStr, pin, err
 }
 
-func (bp *batchProcessor) maskContexts(ctx context.Context, state *DispatchState) (contextsOrPins []*fftypes.Bytes32, err error) {
-	// Calculate the sequence hashes
-	for _, msg := range state.Messages {
+// Calculate the contexts/pins for this batch payload
+func (bp *batchProcessor) calculateContexts(ctx context.Context, payload *DispatchPayload, state *dispatchState) error {
+	payload.Pins = make([]*fftypes.Bytes32, 0)
+
+	for _, msg := range payload.Messages {
 		isPrivate := msg.Header.Group != nil
 		if isPrivate && len(msg.Pins) > 0 {
 			// We have already allocated pins to this message, we cannot re-allocate.
@@ -493,18 +498,19 @@ func (bp *batchProcessor) maskContexts(ctx context.Context, state *DispatchState
 		for i, topic := range msg.Header.Topics {
 			pinString, contextOrPin, err := bp.maskContext(ctx, state, msg, topic)
 			if err != nil {
-				return nil, err
+				return err
 			}
-			contextsOrPins = append(contextsOrPins, contextOrPin)
+			payload.Pins = append(payload.Pins, contextOrPin)
 			if isPrivate {
 				pins[i] = pinString
 			}
 		}
 	}
-	return contextsOrPins, nil
+
+	return nil
 }
 
-func (bp *batchProcessor) flushNonceState(ctx context.Context, state *DispatchState) error {
+func (bp *batchProcessor) flushNonceState(ctx context.Context, state *dispatchState) error {
 
 	// Flush all the assigned nonces to the DB
 	for hash, nonceState := range state.noncesAssigned {
@@ -523,6 +529,10 @@ func (bp *batchProcessor) flushNonceState(ctx context.Context, state *DispatchSt
 		}
 	}
 
+	return nil
+}
+
+func (bp *batchProcessor) updateMessagePins(ctx context.Context, batchID *fftypes.UUID, messages []*core.Message, state *dispatchState) error {
 	// It's important we update the message pins at this phase, as we have "spent" a nonce
 	// on this topic from the database. So this message has grabbed a slot in our queue.
 	// If we fail the dispatch, and redo the batch sealing process, we must not allocate
@@ -530,7 +540,7 @@ func (bp *batchProcessor) flushNonceState(ctx context.Context, state *DispatchSt
 	//
 	// Note to consider database transactions, we do not update the msg.Pins array, or the
 	// message cache, until after the Retry/RunAsGroup has ended in sealBatch.
-	for _, msg := range state.Messages {
+	for _, msg := range messages {
 		if pins, ok := state.msgPins[*msg.Header.ID]; ok {
 			if err := bp.database.UpdateMessage(ctx, bp.bm.namespace, msg.Header.ID,
 				database.MessageQueryFactory.NewUpdate(ctx).Set("pins", pins),
@@ -539,45 +549,50 @@ func (bp *batchProcessor) flushNonceState(ctx context.Context, state *DispatchSt
 			}
 		}
 	}
-
 	return nil
 }
 
-func (bp *batchProcessor) sealBatch(state *DispatchState) (err error) {
+func (bp *batchProcessor) sealBatch(payload *DispatchPayload) (err error) {
+	var state *dispatchState
+
 	err = bp.retry.Do(bp.ctx, "batch persist", func(attempt int) (retry bool, err error) {
 		return true, bp.database.RunAsGroup(bp.ctx, func(ctx context.Context) (err error) {
 
 			// Clear state from any previous retry. We need to do fresh queries against the DB for nonces.
-			state.noncesAssigned = make(map[fftypes.Bytes32]*nonceState)
-			state.msgPins = make(map[fftypes.UUID]fftypes.FFStringArray)
+			state = &dispatchState{
+				noncesAssigned: make(map[fftypes.Bytes32]*nonceState),
+				msgPins:        make(map[fftypes.UUID]fftypes.FFStringArray),
+			}
 
-			if bp.conf.txType == core.TransactionTypeBatchPin {
-				// Generate a new Transaction, which will be used to record status of the associated transaction as it happens
-				if state.Pins, err = bp.maskContexts(ctx, state); err != nil {
+			// Assign nonces and update nonces/messages in the database
+			if core.IsPinned(bp.conf.txType) {
+				if err = bp.calculateContexts(ctx, payload, state); err != nil {
 					return err
 				}
-				// Flush the state for all allocated nonces to the database
 				if err = bp.flushNonceState(ctx, state); err != nil {
 					return err
 				}
+				if err = bp.updateMessagePins(ctx, payload.Batch.ID, payload.Messages, state); err != nil {
+					return err
+				}
 			}
 
-			state.Persisted.TX.Type = bp.conf.txType
-			if state.Persisted.TX.ID, err = bp.txHelper.SubmitNewTransaction(ctx, bp.conf.txType, "" /* no idempotency key for batch TX */); err != nil {
+			payload.Batch.TX.Type = bp.conf.txType
+			if payload.Batch.TX.ID, err = bp.txHelper.SubmitNewTransaction(ctx, bp.conf.txType, "" /* no idempotency key for batch TX */); err != nil {
 				return err
 			}
-			manifest := state.Persisted.GenManifest(state.Messages, state.Data)
+			manifest := payload.Batch.GenManifest(payload.Messages, payload.Data)
 
 			// The hash of the batch, is the hash of the manifest to minimize the compute cost.
 			// Note in v0.13 and before, it was the hash of the payload - so the inbound route has a fallback to accepting the full payload hash
 			manifestString := manifest.String()
-			state.Persisted.Manifest = fftypes.JSONAnyPtr(manifestString)
-			state.Persisted.Hash = fftypes.HashString(manifestString)
+			payload.Batch.Manifest = fftypes.JSONAnyPtr(manifestString)
+			payload.Batch.Hash = fftypes.HashString(manifestString)
 
-			log.L(ctx).Debugf("Batch %s sealed. Hash=%s", state.Persisted.ID, state.Persisted.Hash)
+			log.L(ctx).Debugf("Batch %s sealed. Hash=%s", payload.Batch.ID, payload.Batch.Hash)
 
 			// At this point the manifest of the batch is finalized. We write it to the database
-			_, err = bp.database.InsertOrGetBatch(ctx, &state.Persisted)
+			_, err = bp.database.InsertOrGetBatch(ctx, &payload.Batch)
 			return err
 		})
 	})
@@ -588,35 +603,37 @@ func (bp *batchProcessor) sealBatch(state *DispatchState) (err error) {
 	// Once the DB transaction is done, we need to update the messages with the pins.
 	// We do this at this point, so the logic is re-entrant in a way that avoids re-allocating Pins to messages, but
 	// means the retry loops above using the batch state function correctly.
-	for _, msg := range state.Messages {
-		if pins, ok := state.msgPins[*msg.Header.ID]; ok {
-			msg.Pins = pins
-			bp.data.UpdateMessageIfCached(bp.ctx, msg)
+	if state != nil {
+		for _, msg := range payload.Messages {
+			if pins, ok := state.msgPins[*msg.Header.ID]; ok {
+				msg.Pins = pins
+				bp.data.UpdateMessageIfCached(bp.ctx, msg)
+			}
 		}
 	}
 	return nil
 }
 
-func (bp *batchProcessor) dispatchBatch(state *DispatchState) error {
+func (bp *batchProcessor) dispatchBatch(payload *DispatchPayload) error {
 	// Call the dispatcher to do the heavy lifting - will only exit if we're closed
 	return operations.RunWithOperationContext(bp.ctx, func(ctx context.Context) error {
 		return bp.retry.Do(ctx, "batch dispatch", func(attempt int) (retry bool, err error) {
-			return true, bp.conf.dispatch(ctx, state)
+			return true, bp.conf.dispatch(ctx, payload)
 		})
 	})
 }
 
-func (bp *batchProcessor) markPayloadDispatched(state *DispatchState) error {
+func (bp *batchProcessor) markPayloadDispatched(payload *DispatchPayload) error {
 	return bp.retry.Do(bp.ctx, "mark dispatched messages", func(attempt int) (retry bool, err error) {
 		return true, bp.database.RunAsGroup(bp.ctx, func(ctx context.Context) (err error) {
-			// Update all the messages in the batch with the batch ID
-			msgIDs := make([]driver.Value, len(state.Messages))
+			// Update the message state in the cache
+			msgIDs := make([]driver.Value, len(payload.Messages))
 			confirmTime := fftypes.Now()
-			for i, msg := range state.Messages {
+			for i, msg := range payload.Messages {
 				msgIDs[i] = msg.Header.ID
-				msg.BatchID = state.Persisted.ID
-				msg.TransactionID = state.Persisted.TX.ID
-				if bp.conf.txType == core.TransactionTypeBatchPin {
+				msg.BatchID = payload.Batch.ID
+				msg.TransactionID = payload.Batch.TX.ID
+				if core.IsPinned(bp.conf.txType) {
 					msg.State = core.MessageStateSent
 				} else {
 					msg.State = core.MessageStateConfirmed
@@ -633,31 +650,31 @@ func (bp *batchProcessor) markPayloadDispatched(state *DispatchState) error {
 			)
 
 			var allMsgsUpdate ffapi.Update
-			if bp.conf.txType == core.TransactionTypeBatchPin {
+			if core.IsPinned(bp.conf.txType) {
 				// Sent state waiting for confirm
 				allMsgsUpdate = database.MessageQueryFactory.NewUpdate(ctx).
-					Set("batch", state.Persisted.ID).    // Mark the batch they are in
+					Set("batch", payload.Batch.ID).      // Mark the batch they are in
 					Set("state", core.MessageStateSent). // Set them sent, so they won't be picked up and re-sent after restart/rewind
-					Set("txid", state.Persisted.TX.ID)
+					Set("txid", payload.Batch.TX.ID)
 			} else {
 				// Immediate confirmation if no batch pinning
 				allMsgsUpdate = database.MessageQueryFactory.NewUpdate(ctx).
-					Set("batch", state.Persisted.ID).
+					Set("batch", payload.Batch.ID).
 					Set("state", core.MessageStateConfirmed).
 					Set("confirmed", confirmTime).
-					Set("txid", state.Persisted.TX.ID)
+					Set("txid", payload.Batch.TX.ID)
 			}
 
 			if err = bp.database.UpdateMessages(ctx, bp.bm.namespace, filter, allMsgsUpdate); err != nil {
 				return err
 			}
 
-			if bp.conf.txType == core.TransactionTypeUnpinned {
-				for _, msg := range state.Messages {
+			if !core.IsPinned(bp.conf.txType) {
+				for _, msg := range payload.Messages {
 					// Emit a confirmation event locally immediately
 					for _, topic := range msg.Header.Topics {
 						// One event per topic
-						event := core.NewEvent(core.EventTypeMessageConfirmed, state.Persisted.Namespace, msg.Header.ID, state.Persisted.TX.ID, topic)
+						event := core.NewEvent(core.EventTypeMessageConfirmed, payload.Batch.Namespace, msg.Header.ID, payload.Batch.TX.ID, topic)
 						event.Correlator = msg.Header.CID
 						if err := bp.database.InsertEvent(ctx, event); err != nil {
 							return err
diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
index e0fcde5d8..fa8f5585c 100644
--- a/internal/blockchain/ethereum/ethereum.go
+++ b/internal/blockchain/ethereum/ethereum.go
@@ -223,18 +223,18 @@ func (e *Ethereum) Capabilities() *blockchain.Capabilities {
 	return e.capabilities
 }
 
-func (e *Ethereum) AddFireflySubscription(ctx context.Context, namespace *core.Namespace, location *fftypes.JSONAny, firstEvent string) (string, error) {
-	ethLocation, err := e.parseContractLocation(ctx, location)
+func (e *Ethereum) AddFireflySubscription(ctx context.Context, namespace *core.Namespace, contract *blockchain.MultipartyContract) (string, error) {
+	ethLocation, err := e.parseContractLocation(ctx, contract.Location)
 	if err != nil {
 		return "", err
 	}
 
-	version, err := e.GetNetworkVersion(ctx, location)
+	version, err := e.GetNetworkVersion(ctx, contract.Location)
 	if err != nil {
 		return "", err
 	}
 
-	sub, err := e.streams.ensureFireFlySubscription(ctx, namespace.Name, version, ethLocation.Address, firstEvent, e.streamID, batchPinEventABI)
+	sub, err := e.streams.ensureFireFlySubscription(ctx, namespace.Name, version, ethLocation.Address, contract.FirstEvent, e.streamID, batchPinEventABI)
 	if err != nil {
 		return "", err
 	}
diff --git a/internal/blockchain/fabric/fabric.go b/internal/blockchain/fabric/fabric.go
index 9ec154946..70f8badb9 100644
--- a/internal/blockchain/fabric/fabric.go
+++ b/internal/blockchain/fabric/fabric.go
@@ -369,18 +369,18 @@ func (f *Fabric) handleContractEvent(ctx context.Context, msgJSON fftypes.JSONOb
 	})
 }
 
-func (f *Fabric) AddFireflySubscription(ctx context.Context, namespace *core.Namespace, location *fftypes.JSONAny, firstEvent string) (string, error) {
-	fabricOnChainLocation, err := parseContractLocation(ctx, location)
+func (f *Fabric) AddFireflySubscription(ctx context.Context, namespace *core.Namespace, contract *blockchain.MultipartyContract) (string, error) {
+	fabricOnChainLocation, err := parseContractLocation(ctx, contract.Location)
 	if err != nil {
 		return "", err
 	}
 
-	version, err := f.GetNetworkVersion(ctx, location)
+	version, err := f.GetNetworkVersion(ctx, contract.Location)
 	if err != nil {
 		return "", err
 	}
 
-	sub, err := f.streams.ensureFireFlySubscription(ctx, namespace.Name, version, fabricOnChainLocation, firstEvent, f.streamID, batchPinEvent)
+	sub, err := f.streams.ensureFireFlySubscription(ctx, namespace.Name, version, fabricOnChainLocation, contract.FirstEvent, f.streamID, batchPinEvent)
 	if err != nil {
 		return "", err
 	}
diff --git a/internal/broadcast/manager.go b/internal/broadcast/manager.go
index d6db99676..a1822f92f 100644
--- a/internal/broadcast/manager.go
+++ b/internal/broadcast/manager.go
@@ -127,10 +127,10 @@ func (bm *broadcastManager) Name() string {
 	return "BroadcastManager"
 }
 
-func (bm *broadcastManager) dispatchBatch(ctx context.Context, state *batch.DispatchState) error {
+func (bm *broadcastManager) dispatchBatch(ctx context.Context, payload *batch.DispatchPayload) error {
 
 	// Ensure all the blobs are published
-	if err := bm.uploadBlobs(ctx, state.Persisted.TX.ID, state.Data); err != nil {
+	if err := bm.uploadBlobs(ctx, payload.Batch.TX.ID, payload.Data); err != nil {
 		return err
 	}
 
@@ -138,13 +138,13 @@ func (bm *broadcastManager) dispatchBatch(ctx context.Context, state *batch.Disp
 	op := core.NewOperation(
 		bm.sharedstorage,
 		bm.namespace.Name,
-		state.Persisted.TX.ID,
+		payload.Batch.TX.ID,
 		core.OpTypeSharedStorageUploadBatch)
-	addUploadBatchInputs(op, state.Persisted.ID)
+	addUploadBatchInputs(op, payload.Batch.ID)
 	if err := bm.operations.AddOrReuseOperation(ctx, op); err != nil {
 		return err
 	}
-	batch := state.Persisted.GenInflight(state.Messages, state.Data)
+	batch := payload.Batch.GenInflight(payload.Messages, payload.Data)
 
 	// We are in an (indefinite) retry cycle from the batch processor to dispatch this batch, that is only
 	// terminated with shutdown. So we leave the operation pending on failure, as it is still being retried.
@@ -155,7 +155,7 @@ func (bm *broadcastManager) dispatchBatch(ctx context.Context, state *batch.Disp
 	}
 	payloadRef := outputs.GetString("payloadRef")
 	log.L(ctx).Infof("Pinning broadcast batch %s with author=%s key=%s payloadRef=%s", batch.ID, batch.Author, batch.Key, payloadRef)
-	return bm.multiparty.SubmitBatchPin(ctx, &state.Persisted, state.Pins, payloadRef)
+	return bm.multiparty.SubmitBatchPin(ctx, &payload.Batch, payload.Pins, payloadRef)
 }
 
 func (bm *broadcastManager) uploadBlobs(ctx context.Context, tx *fftypes.UUID, data core.DataArray) error {
diff --git a/internal/coremsgs/en_error_messages.go b/internal/coremsgs/en_error_messages.go
index 92c7aa368..e223bfba9 100644
--- a/internal/coremsgs/en_error_messages.go
+++ b/internal/coremsgs/en_error_messages.go
@@ -102,10 +102,9 @@ var (
 	MsgDataReferenceUnresolvable          = ffe("FF10204", "Data reference %d cannot be resolved", 400)
 	MsgDataMissing                        = ffe("FF10205", "Data entry %d has neither 'id' to refer to existing data, or 'value' to include in-line JSON data", 400)
 	MsgAuthorInvalid                      = ffe("FF10206", "Invalid author specified", 400)
-	MsgNoTransaction                      = ffe("FF10207", "Message does not have a transaction", 404)
-	MsgBatchNotSet                        = ffe("FF10208", "Message does not have an assigned batch", 404)
-	MsgBatchNotFound                      = ffe("FF10209", "Batch '%s' not found for message", 500)
-	MsgBatchTXNotSet                      = ffe("FF10210", "Batch '%s' does not have an assigned transaction", 404)
+	MsgMessageNotFound                    = ffe("FF10207", "Message '%s' not found", 404)
+	MsgBatchNotFound                      = ffe("FF10209", "Batch '%s' not found for message", 404)
+	MsgMessageTXNotSet                    = ffe("FF10210", "Message '%s' does not have an assigned transaction", 404)
 	MsgOwnerMissing                       = ffe("FF10211", "Owner missing", 400)
 	MsgUnknownIdentityPlugin              = ffe("FF10212", "Unknown Identity plugin '%s'")
 	MsgUnknownDataExchangePlugin          = ffe("FF10213", "Unknown Data Exchange plugin '%s'")
diff --git a/internal/events/dx_callbacks.go b/internal/events/dx_callbacks.go
index d199c31d6..1be3e797f 100644
--- a/internal/events/dx_callbacks.go
+++ b/internal/events/dx_callbacks.go
@@ -85,7 +85,7 @@ func (em *eventManager) privateBatchReceived(peerID string, batch *core.Batch, w
 		return true, em.database.RunAsGroup(em.ctx, func(ctx context.Context) error {
 			l := log.L(ctx)
 
-			if batch.Payload.TX.Type == core.TransactionTypeUnpinned {
+			if !core.IsPinned(batch.Payload.TX.Type) {
 				if wrapperGroup != nil {
 					if valid, err := em.messaging.EnsureLocalGroup(ctx, wrapperGroup, sender); err != nil {
 						return err
@@ -109,7 +109,7 @@ func (em *eventManager) privateBatchReceived(peerID string, batch *core.Batch, w
 				return err // retry - persistBatch only returns retryable errors
 			}
 
-			if batch.Payload.TX.Type == core.TransactionTypeUnpinned {
+			if !core.IsPinned(batch.Payload.TX.Type) {
 				// We need to confirm all these messages immediately.
 				if err := em.markUnpinnedMessagesConfirmed(ctx, batch); err != nil {
 					return err
@@ -124,7 +124,7 @@ func (em *eventManager) privateBatchReceived(peerID string, batch *core.Batch, w
 		return "", err
 	}
 	// Poke the aggregator to do its stuff - after we have committed the transaction so the pins are visible
-	if batch.Payload.TX.Type == core.TransactionTypeBatchPin {
+	if core.IsPinned(batch.Payload.TX.Type) {
 		em.aggregator.queueBatchRewind(batch.ID)
 	}
 	return manifest, err
diff --git a/internal/events/token_pool_created.go b/internal/events/token_pool_created.go
index 7d06b835c..d99b4ddec 100644
--- a/internal/events/token_pool_created.go
+++ b/internal/events/token_pool_created.go
@@ -25,7 +25,6 @@ import (
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/blockchain"
 	"github.com/hyperledger/firefly/pkg/core"
-	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/tokens"
 )
 
@@ -78,22 +77,6 @@ func (em *eventManager) confirmPool(ctx context.Context, pool *core.TokenPool, e
 	return em.database.InsertEvent(ctx, event)
 }
 
-func (em *eventManager) findTXOperation(ctx context.Context, tx *fftypes.UUID, opType core.OpType) (*core.Operation, error) {
-	// Find a matching operation within this transaction
-	fb := database.OperationQueryFactory.NewFilter(ctx)
-	filter := fb.And(
-		fb.Eq("tx", tx),
-		fb.Eq("type", opType),
-	)
-	if operations, _, err := em.database.GetOperations(ctx, em.namespace.Name, filter); err != nil {
-		return nil, err
-	} else if len(operations) > 0 {
-		return operations[0], nil
-	}
-	log.L(ctx).Debugf("No pool found for tx=%s status=%s", tx, core.OpTypeTokenCreatePool)
-	return nil, nil
-}
-
 func (em *eventManager) loadExisting(ctx context.Context, pool *tokens.TokenPool) (existingPool *core.TokenPool, err error) {
 	if existingPool, err = em.database.GetTokenPoolByLocator(ctx, em.namespace.Name, pool.Connector, pool.PoolLocator); err != nil || existingPool == nil {
 		log.L(ctx).Debugf("Pool not found with ns=%s connector=%s locator=%s (err=%v)", em.namespace.Name, pool.Connector, pool.PoolLocator, err)
@@ -108,10 +91,11 @@ func (em *eventManager) loadExisting(ctx context.Context, pool *tokens.TokenPool
 }
 
 func (em *eventManager) loadFromOperation(ctx context.Context, pool *tokens.TokenPool) (announcePool *core.TokenPool, err error) {
-	op, err := em.findTXOperation(ctx, pool.TX.ID, core.OpTypeTokenCreatePool)
+	op, err := em.txHelper.FindOperationInTransaction(ctx, pool.TX.ID, core.OpTypeTokenCreatePool)
 	if err != nil {
 		return nil, err
 	} else if op == nil {
+		log.L(ctx).Debugf("No pool found for tx=%s status=%s", pool.TX.ID, core.OpTypeTokenCreatePool)
 		return nil, nil
 	}
 
diff --git a/internal/events/tokens_approved.go b/internal/events/tokens_approved.go
index b9ca4621c..e5000ac52 100644
--- a/internal/events/tokens_approved.go
+++ b/internal/events/tokens_approved.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -37,21 +37,14 @@ import (
 //   - The LocalID must not have been used yet. Connectors are allowed to emit multiple events in response to a single operation,
 //     but only the first of them can use the original LocalID.
 func (em *eventManager) loadApprovalID(ctx context.Context, tx *fftypes.UUID, approval *core.TokenApproval) (*fftypes.UUID, error) {
-	// Find a matching operation within the transaction
-	fb := database.OperationQueryFactory.NewFilter(ctx)
-	filter := fb.And(
-		fb.Eq("tx", tx),
-		fb.Eq("type", core.OpTypeTokenApproval),
-	)
-	operations, _, err := em.database.GetOperations(ctx, em.namespace.Name, filter)
+	op, err := em.txHelper.FindOperationInTransaction(ctx, tx, core.OpTypeTokenApproval)
 	if err != nil {
 		return nil, err
 	}
-
-	if len(operations) > 0 {
+	if op != nil {
 		// This approval matches an approval transaction+operation submitted by this node.
 		// Check the operation inputs to see if they match the connector and pool on this event.
-		if input, err := txcommon.RetrieveTokenApprovalInputs(ctx, operations[0]); err != nil {
+		if input, err := txcommon.RetrieveTokenApprovalInputs(ctx, op); err != nil {
 			log.L(ctx).Warnf("Failed to read operation inputs for token approval '%s': %s", approval.Subject, err)
 		} else if input != nil && input.Connector == approval.Connector && input.Pool.Equals(approval.Pool) {
 			// Check if the LocalID has already been used
diff --git a/internal/events/tokens_transferred.go b/internal/events/tokens_transferred.go
index 48bf0d9c0..436d7911c 100644
--- a/internal/events/tokens_transferred.go
+++ b/internal/events/tokens_transferred.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -23,7 +23,6 @@ import (
 	"github.com/hyperledger/firefly-common/pkg/log"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/core"
-	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/tokens"
 )
 
@@ -37,21 +36,14 @@ import (
 //   - The LocalID must not have been used yet. Connectors are allowed to emit multiple events in response to a single operation,
 //     but only the first of them can use the original LocalID.
 func (em *eventManager) loadTransferID(ctx context.Context, tx *fftypes.UUID, transfer *core.TokenTransfer) (*fftypes.UUID, error) {
-	// Find a matching operation within the transaction
-	fb := database.OperationQueryFactory.NewFilter(ctx)
-	filter := fb.And(
-		fb.Eq("tx", tx),
-		fb.Eq("type", core.OpTypeTokenTransfer),
-	)
-	operations, _, err := em.database.GetOperations(ctx, em.namespace.Name, filter)
+	op, err := em.txHelper.FindOperationInTransaction(ctx, tx, core.OpTypeTokenTransfer)
 	if err != nil {
 		return nil, err
 	}
-
-	if len(operations) > 0 {
+	if op != nil {
 		// This transfer matches a transfer transaction+operation submitted by this node.
 		// Check the operation inputs to see if they match the connector and pool on this event.
-		if input, err := txcommon.RetrieveTokenTransferInputs(ctx, operations[0]); err != nil {
+		if input, err := txcommon.RetrieveTokenTransferInputs(ctx, op); err != nil {
 			log.L(ctx).Warnf("Failed to read operation inputs for token transfer '%s': %s", transfer.ProtocolID, err)
 		} else if input != nil && input.Connector == transfer.Connector && input.Pool.Equals(transfer.Pool) {
 			// Check if the LocalID has already been used
diff --git a/internal/multiparty/manager.go b/internal/multiparty/manager.go
index 83ed4dc30..4eb2a7f5d 100644
--- a/internal/multiparty/manager.go
+++ b/internal/multiparty/manager.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -70,7 +70,7 @@ type Config struct {
 	Enabled   bool
 	Org       RootOrg
 	Node      LocalNode
-	Contracts []Contract
+	Contracts []blockchain.MultipartyContract
 }
 
 type RootOrg struct {
@@ -84,11 +84,6 @@ type LocalNode struct {
 	Description string
 }
 
-type Contract struct {
-	Location   *fftypes.JSONAny
-	FirstEvent string
-}
-
 type multipartyManager struct {
 	namespace  *core.Namespace
 	database   database.Plugin
@@ -144,26 +139,26 @@ func (mm *multipartyManager) configureContractCommon(ctx context.Context, migrat
 	}
 	active := mm.namespace.Contracts.Active
 	log.L(ctx).Infof("Resolving FireFly contract at index %d", active.Index)
-	location, firstEvent, err := mm.resolveFireFlyContract(ctx, active.Index)
+	current, err := mm.resolveFireFlyContract(ctx, active.Index)
 	if err != nil {
 		return err
 	}
 
-	version, err := mm.blockchain.GetNetworkVersion(ctx, location)
+	version, err := mm.blockchain.GetNetworkVersion(ctx, current.Location)
 	if err != nil {
 		return err
 	}
 
 	if !migration {
-		if !active.Location.IsNil() && active.Location.String() != location.String() {
-			log.L(ctx).Warnf("FireFly contract location changed from %s to %s", active.Location, location)
+		if !active.Location.IsNil() && active.Location.String() != current.Location.String() {
+			log.L(ctx).Warnf("FireFly contract location changed from %s to %s", active.Location, current.Location)
 		}
 	}
 
-	subID, err := mm.blockchain.AddFireflySubscription(ctx, mm.namespace, location, firstEvent)
+	subID, err := mm.blockchain.AddFireflySubscription(ctx, mm.namespace, current)
 	if err == nil {
-		active.Location = location
-		active.FirstEvent = firstEvent
+		active.Location = current.Location
+		active.FirstEvent = current.FirstEvent
 		active.Info.Subscription = subID
 		active.Info.Version = version
 		err = mm.database.UpsertNamespace(ctx, mm.namespace, true)
@@ -171,24 +166,24 @@ func (mm *multipartyManager) configureContractCommon(ctx context.Context, migrat
 	return err
 }
 
-func (mm *multipartyManager) resolveFireFlyContract(ctx context.Context, contractIndex int) (location *fftypes.JSONAny, firstEvent string, err error) {
+func (mm *multipartyManager) resolveFireFlyContract(ctx context.Context, contractIndex int) (contract *blockchain.MultipartyContract, err error) {
 	if len(mm.config.Contracts) > 0 || contractIndex > 0 {
 		if contractIndex >= len(mm.config.Contracts) {
-			return nil, "", i18n.NewError(ctx, coremsgs.MsgInvalidFireFlyContractIndex,
+			return nil, i18n.NewError(ctx, coremsgs.MsgInvalidFireFlyContractIndex,
 				fmt.Sprintf("%s.multiparty.contracts[%d]", mm.namespace.Name, contractIndex))
 		}
-		active := mm.config.Contracts[contractIndex]
-		location = active.Location
-		firstEvent = active.FirstEvent
-	} else {
-		// handle deprecated config here
-		location, firstEvent, err = mm.blockchain.GetAndConvertDeprecatedContractConfig(ctx)
-		if err != nil {
-			return nil, "", err
-		}
+		return &mm.config.Contracts[contractIndex], nil
 	}
 
-	return location, firstEvent, err
+	// handle deprecated config here
+	location, firstEvent, err := mm.blockchain.GetAndConvertDeprecatedContractConfig(ctx)
+	if err != nil {
+		return nil, err
+	}
+	return &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: firstEvent,
+	}, err
 }
 
 func (mm *multipartyManager) TerminateContract(ctx context.Context, location *fftypes.JSONAny, termination *blockchain.Event) (err error) {
diff --git a/internal/namespace/manager.go b/internal/namespace/manager.go
index 673864ca9..6d8778671 100644
--- a/internal/namespace/manager.go
+++ b/internal/namespace/manager.go
@@ -38,7 +38,6 @@ import (
 	"github.com/hyperledger/firefly/internal/events/system"
 	"github.com/hyperledger/firefly/internal/identity/iifactory"
 	"github.com/hyperledger/firefly/internal/metrics"
-	"github.com/hyperledger/firefly/internal/multiparty"
 	"github.com/hyperledger/firefly/internal/orchestrator"
 	"github.com/hyperledger/firefly/internal/sharedstorage/ssfactory"
 	"github.com/hyperledger/firefly/internal/spievents"
@@ -875,12 +874,12 @@ func (nm *namespaceManager) loadNamespace(ctx context.Context, name string, inde
 	if multipartyEnabled.(bool) {
 		contractsConf := multipartyConf.SubArray(coreconfig.NamespaceMultipartyContract)
 		contractConfArraySize := contractsConf.ArraySize()
-		contracts := make([]multiparty.Contract, contractConfArraySize)
+		contracts := make([]blockchain.MultipartyContract, contractConfArraySize)
 
 		for i := 0; i < contractConfArraySize; i++ {
 			conf := contractsConf.ArrayEntry(i)
 			location := fftypes.JSONAnyPtr(conf.GetObject(coreconfig.NamespaceMultipartyContractLocation).String())
-			contract := multiparty.Contract{
+			contract := blockchain.MultipartyContract{
 				Location:   location,
 				FirstEvent: conf.GetString(coreconfig.NamespaceMultipartyContractFirstEvent),
 			}
diff --git a/internal/orchestrator/data_query.go b/internal/orchestrator/data_query.go
index e97e1bee8..afb0c23ed 100644
--- a/internal/orchestrator/data_query.go
+++ b/internal/orchestrator/data_query.go
@@ -224,26 +224,10 @@ func (or *orchestrator) getMessageTransactionID(ctx context.Context, id string)
 	if err != nil || msg == nil {
 		return nil, err
 	}
-	var txID *fftypes.UUID
-	if msg.Header.TxType == core.TransactionTypeBatchPin {
-		if msg.BatchID == nil {
-			return nil, i18n.NewError(ctx, coremsgs.MsgBatchNotSet)
-		}
-		batch, err := or.database().GetBatchByID(ctx, or.namespace.Name, msg.BatchID)
-		if err != nil {
-			return nil, err
-		}
-		if batch == nil {
-			return nil, i18n.NewError(ctx, coremsgs.MsgBatchNotFound, msg.BatchID)
-		}
-		txID = batch.TX.ID
-		if txID == nil {
-			return nil, i18n.NewError(ctx, coremsgs.MsgBatchTXNotSet, msg.BatchID)
-		}
-	} else {
-		return nil, i18n.NewError(ctx, coremsgs.MsgNoTransaction)
+	if msg.TransactionID == nil {
+		return nil, i18n.NewError(ctx, coremsgs.MsgMessageTXNotSet, msg.Header.ID)
 	}
-	return txID, nil
+	return msg.TransactionID, nil
 }
 
 func (or *orchestrator) GetMessageTransaction(ctx context.Context, id string) (*core.Transaction, error) {
diff --git a/internal/privatemessaging/privatemessaging.go b/internal/privatemessaging/privatemessaging.go
index f03f25230..937720481 100644
--- a/internal/privatemessaging/privatemessaging.go
+++ b/internal/privatemessaging/privatemessaging.go
@@ -165,22 +165,22 @@ func (pm *privateMessaging) Name() string {
 	return "PrivateMessaging"
 }
 
-func (pm *privateMessaging) dispatchPinnedBatch(ctx context.Context, state *batch.DispatchState) error {
-	err := pm.dispatchBatchCommon(ctx, state)
+func (pm *privateMessaging) dispatchPinnedBatch(ctx context.Context, payload *batch.DispatchPayload) error {
+	err := pm.dispatchBatchCommon(ctx, payload)
 	if err != nil {
 		return err
 	}
 
-	log.L(ctx).Infof("Pinning private batch %s with author=%s key=%s group=%s", state.Persisted.ID, state.Persisted.Author, state.Persisted.Key, state.Persisted.Group)
-	return pm.multiparty.SubmitBatchPin(ctx, &state.Persisted, state.Pins, "" /* no payloadRef for private */)
+	log.L(ctx).Infof("Pinning private batch %s with author=%s key=%s group=%s", payload.Batch.ID, payload.Batch.Author, payload.Batch.Key, payload.Batch.Group)
+	return pm.multiparty.SubmitBatchPin(ctx, &payload.Batch, payload.Pins, "" /* no payloadRef for private */)
 }
 
-func (pm *privateMessaging) dispatchUnpinnedBatch(ctx context.Context, state *batch.DispatchState) error {
-	return pm.dispatchBatchCommon(ctx, state)
+func (pm *privateMessaging) dispatchUnpinnedBatch(ctx context.Context, payload *batch.DispatchPayload) error {
+	return pm.dispatchBatchCommon(ctx, payload)
 }
 
-func (pm *privateMessaging) dispatchBatchCommon(ctx context.Context, state *batch.DispatchState) error {
-	batch := state.Persisted.GenInflight(state.Messages, state.Data)
+func (pm *privateMessaging) dispatchBatchCommon(ctx context.Context, payload *batch.DispatchPayload) error {
+	batch := payload.Batch.GenInflight(payload.Messages, payload.Data)
 	batch.Namespace = pm.namespace.NetworkName
 	tw := &core.TransportWrapper{
 		Batch: batch,
diff --git a/internal/txcommon/txcommon.go b/internal/txcommon/txcommon.go
index 85fe7de95..af5a776db 100644
--- a/internal/txcommon/txcommon.go
+++ b/internal/txcommon/txcommon.go
@@ -36,6 +36,7 @@ type Helper interface {
 	InsertOrGetBlockchainEvent(ctx context.Context, event *core.BlockchainEvent) (existing *core.BlockchainEvent, err error)
 	GetTransactionByIDCached(ctx context.Context, id *fftypes.UUID) (*core.Transaction, error)
 	GetBlockchainEventByIDCached(ctx context.Context, id *fftypes.UUID) (*core.BlockchainEvent, error)
+	FindOperationInTransaction(ctx context.Context, tx *fftypes.UUID, opType core.OpType) (*core.Operation, error)
 }
 
 type transactionHelper struct {
@@ -211,3 +212,16 @@ func (t *transactionHelper) InsertOrGetBlockchainEvent(ctx context.Context, even
 	t.addBlockchainEventToCache(event)
 	return nil, nil
 }
+
+func (t *transactionHelper) FindOperationInTransaction(ctx context.Context, tx *fftypes.UUID, opType core.OpType) (*core.Operation, error) {
+	fb := database.OperationQueryFactory.NewFilter(ctx)
+	filter := fb.And(
+		fb.Eq("tx", tx),
+		fb.Eq("type", opType),
+	)
+	ops, _, err := t.database.GetOperations(ctx, t.namespace, filter)
+	if err != nil || len(ops) == 0 {
+		return nil, err
+	}
+	return ops[0], nil
+}
diff --git a/mocks/blockchainmocks/plugin.go b/mocks/blockchainmocks/plugin.go
index ac2fb77a5..3840a7537 100644
--- a/mocks/blockchainmocks/plugin.go
+++ b/mocks/blockchainmocks/plugin.go
@@ -38,23 +38,23 @@ func (_m *Plugin) AddContractListener(ctx context.Context, subscription *core.Co
 	return r0
 }
 
-// AddFireflySubscription provides a mock function with given fields: ctx, namespace, location, firstEvent
-func (_m *Plugin) AddFireflySubscription(ctx context.Context, namespace *core.Namespace, location *fftypes.JSONAny, firstEvent string) (string, error) {
-	ret := _m.Called(ctx, namespace, location, firstEvent)
+// AddFireflySubscription provides a mock function with given fields: ctx, namespace, contract
+func (_m *Plugin) AddFireflySubscription(ctx context.Context, namespace *core.Namespace, contract *blockchain.MultipartyContract) (string, error) {
+	ret := _m.Called(ctx, namespace, contract)
 
 	var r0 string
 	var r1 error
-	if rf, ok := ret.Get(0).(func(context.Context, *core.Namespace, *fftypes.JSONAny, string) (string, error)); ok {
-		return rf(ctx, namespace, location, firstEvent)
+	if rf, ok := ret.Get(0).(func(context.Context, *core.Namespace, *blockchain.MultipartyContract) (string, error)); ok {
+		return rf(ctx, namespace, contract)
 	}
-	if rf, ok := ret.Get(0).(func(context.Context, *core.Namespace, *fftypes.JSONAny, string) string); ok {
-		r0 = rf(ctx, namespace, location, firstEvent)
+	if rf, ok := ret.Get(0).(func(context.Context, *core.Namespace, *blockchain.MultipartyContract) string); ok {
+		r0 = rf(ctx, namespace, contract)
 	} else {
 		r0 = ret.Get(0).(string)
 	}
 
-	if rf, ok := ret.Get(1).(func(context.Context, *core.Namespace, *fftypes.JSONAny, string) error); ok {
-		r1 = rf(ctx, namespace, location, firstEvent)
+	if rf, ok := ret.Get(1).(func(context.Context, *core.Namespace, *blockchain.MultipartyContract) error); ok {
+		r1 = rf(ctx, namespace, contract)
 	} else {
 		r1 = ret.Error(1)
 	}
diff --git a/mocks/txcommonmocks/helper.go b/mocks/txcommonmocks/helper.go
index 8bce3adf5..784d4f55f 100644
--- a/mocks/txcommonmocks/helper.go
+++ b/mocks/txcommonmocks/helper.go
@@ -30,6 +30,32 @@ func (_m *Helper) AddBlockchainTX(ctx context.Context, tx *core.Transaction, blo
 	return r0
 }
 
+// FindOperationInTransaction provides a mock function with given fields: ctx, tx, opType
+func (_m *Helper) FindOperationInTransaction(ctx context.Context, tx *fftypes.UUID, opType fftypes.FFEnum) (*core.Operation, error) {
+	ret := _m.Called(ctx, tx, opType)
+
+	var r0 *core.Operation
+	var r1 error
+	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, fftypes.FFEnum) (*core.Operation, error)); ok {
+		return rf(ctx, tx, opType)
+	}
+	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.UUID, fftypes.FFEnum) *core.Operation); ok {
+		r0 = rf(ctx, tx, opType)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*core.Operation)
+		}
+	}
+
+	if rf, ok := ret.Get(1).(func(context.Context, *fftypes.UUID, fftypes.FFEnum) error); ok {
+		r1 = rf(ctx, tx, opType)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
 // GetBlockchainEventByIDCached provides a mock function with given fields: ctx, id
 func (_m *Helper) GetBlockchainEventByIDCached(ctx context.Context, id *fftypes.UUID) (*core.BlockchainEvent, error) {
 	ret := _m.Called(ctx, id)
diff --git a/pkg/blockchain/plugin.go b/pkg/blockchain/plugin.go
index 4d9beb22e..6cf6e43a9 100644
--- a/pkg/blockchain/plugin.go
+++ b/pkg/blockchain/plugin.go
@@ -107,7 +107,7 @@ type Plugin interface {
 	GetAndConvertDeprecatedContractConfig(ctx context.Context) (location *fftypes.JSONAny, fromBlock string, err error)
 
 	// AddFireflySubscription creates a FireFly BatchPin subscription for the provided location
-	AddFireflySubscription(ctx context.Context, namespace *core.Namespace, location *fftypes.JSONAny, firstEvent string) (subID string, err error)
+	AddFireflySubscription(ctx context.Context, namespace *core.Namespace, contract *MultipartyContract) (subID string, err error)
 
 	// RemoveFireFlySubscription removes the provided FireFly subscription
 	RemoveFireflySubscription(ctx context.Context, subID string)
@@ -144,6 +144,12 @@ type Callbacks interface {
 type Capabilities struct {
 }
 
+// MultipartyContract represents the location and configuration of a FireFly multiparty contract for batch pinning of messages
+type MultipartyContract struct {
+	Location   *fftypes.JSONAny
+	FirstEvent string
+}
+
 // BatchPin is the set of data pinned to the blockchain for a batch - whether it's private or broadcast.
 type BatchPin struct {
 
diff --git a/pkg/core/transaction.go b/pkg/core/transaction.go
index 18fa0f392..cb7df5c4c 100644
--- a/pkg/core/transaction.go
+++ b/pkg/core/transaction.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -98,3 +98,7 @@ type TransactionStatus struct {
 func (tx *Transaction) Size() int64 {
 	return transactionBaseSizeEstimate // currently a static size assessment for caching
 }
+
+func IsPinned(t TransactionType) bool {
+	return t.Equals(TransactionTypeBatchPin)
+}


Test patch:
diff --git a/internal/assets/token_pool_test.go b/internal/assets/token_pool_test.go
index e44d84d79..6c4e48692 100644
--- a/internal/assets/token_pool_test.go
+++ b/internal/assets/token_pool_test.go
@@ -461,11 +461,14 @@ func TestActivateTokenPool(t *testing.T) {
 		ID:        fftypes.NewUUID(),
 		Namespace: "ns1",
 		Connector: "magic-tokens",
+		TX: core.TransactionRef{
+			ID: fftypes.NewUUID(),
+		},
 	}
 
-	mdi := am.database.(*databasemocks.Plugin)
 	mom := am.operations.(*operationmocks.Manager)
-	mdi.On("GetOperations", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
+	mth := am.txHelper.(*txcommonmocks.Helper)
+	mth.On("FindOperationInTransaction", context.Background(), pool.TX.ID, core.OpTypeTokenActivatePool).Return(nil, nil)
 	mom.On("AddOrReuseOperation", context.Background(), mock.MatchedBy(func(op *core.Operation) bool {
 		return op.Type == core.OpTypeTokenActivatePool
 	})).Return(nil)
@@ -477,7 +480,7 @@ func TestActivateTokenPool(t *testing.T) {
 	err := am.ActivateTokenPool(context.Background(), pool)
 	assert.NoError(t, err)
 
-	mdi.AssertExpectations(t)
+	mth.AssertExpectations(t)
 	mom.AssertExpectations(t)
 }
 
@@ -501,11 +504,14 @@ func TestActivateTokenPoolOpInsertFail(t *testing.T) {
 	pool := &core.TokenPool{
 		Namespace: "ns1",
 		Connector: "magic-tokens",
+		TX: core.TransactionRef{
+			ID: fftypes.NewUUID(),
+		},
 	}
 
-	mdi := am.database.(*databasemocks.Plugin)
 	mom := am.operations.(*operationmocks.Manager)
-	mdi.On("GetOperations", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
+	mth := am.txHelper.(*txcommonmocks.Helper)
+	mth.On("FindOperationInTransaction", context.Background(), pool.TX.ID, core.OpTypeTokenActivatePool).Return(nil, nil)
 	mom.On("AddOrReuseOperation", context.Background(), mock.MatchedBy(func(op *core.Operation) bool {
 		return op.Type == core.OpTypeTokenActivatePool
 	})).Return(fmt.Errorf("pop"))
@@ -513,7 +519,7 @@ func TestActivateTokenPoolOpInsertFail(t *testing.T) {
 	err := am.ActivateTokenPool(context.Background(), pool)
 	assert.EqualError(t, err, "pop")
 
-	mdi.AssertExpectations(t)
+	mth.AssertExpectations(t)
 	mom.AssertExpectations(t)
 }
 
@@ -524,11 +530,14 @@ func TestActivateTokenPoolFail(t *testing.T) {
 	pool := &core.TokenPool{
 		Namespace: "ns1",
 		Connector: "magic-tokens",
+		TX: core.TransactionRef{
+			ID: fftypes.NewUUID(),
+		},
 	}
 
-	mdi := am.database.(*databasemocks.Plugin)
 	mom := am.operations.(*operationmocks.Manager)
-	mdi.On("GetOperations", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
+	mth := am.txHelper.(*txcommonmocks.Helper)
+	mth.On("FindOperationInTransaction", context.Background(), pool.TX.ID, core.OpTypeTokenActivatePool).Return(nil, nil)
 	mom.On("AddOrReuseOperation", context.Background(), mock.MatchedBy(func(op *core.Operation) bool {
 		return op.Type == core.OpTypeTokenActivatePool
 	})).Return(nil)
@@ -540,7 +549,7 @@ func TestActivateTokenPoolFail(t *testing.T) {
 	err := am.ActivateTokenPool(context.Background(), pool)
 	assert.EqualError(t, err, "pop")
 
-	mdi.AssertExpectations(t)
+	mth.AssertExpectations(t)
 	mom.AssertExpectations(t)
 }
 
@@ -551,15 +560,18 @@ func TestActivateTokenPoolExisting(t *testing.T) {
 	pool := &core.TokenPool{
 		Namespace: "ns1",
 		Connector: "magic-tokens",
+		TX: core.TransactionRef{
+			ID: fftypes.NewUUID(),
+		},
 	}
 
-	mdi := am.database.(*databasemocks.Plugin)
-	mdi.On("GetOperations", context.Background(), "ns1", mock.Anything).Return([]*core.Operation{{}}, nil, nil)
+	mth := am.txHelper.(*txcommonmocks.Helper)
+	mth.On("FindOperationInTransaction", context.Background(), pool.TX.ID, core.OpTypeTokenActivatePool).Return(&core.Operation{}, nil)
 
 	err := am.ActivateTokenPool(context.Background(), pool)
 	assert.NoError(t, err)
 
-	mdi.AssertExpectations(t)
+	mth.AssertExpectations(t)
 }
 
 func TestActivateTokenPoolExistingFail(t *testing.T) {
@@ -569,15 +581,18 @@ func TestActivateTokenPoolExistingFail(t *testing.T) {
 	pool := &core.TokenPool{
 		Namespace: "ns1",
 		Connector: "magic-tokens",
+		TX: core.TransactionRef{
+			ID: fftypes.NewUUID(),
+		},
 	}
 
-	mdi := am.database.(*databasemocks.Plugin)
-	mdi.On("GetOperations", context.Background(), "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
+	mth := am.txHelper.(*txcommonmocks.Helper)
+	mth.On("FindOperationInTransaction", context.Background(), pool.TX.ID, core.OpTypeTokenActivatePool).Return(nil, fmt.Errorf("pop"))
 
 	err := am.ActivateTokenPool(context.Background(), pool)
 	assert.EqualError(t, err, "pop")
 
-	mdi.AssertExpectations(t)
+	mth.AssertExpectations(t)
 }
 
 func TestActivateTokenPoolSyncSuccess(t *testing.T) {
@@ -587,12 +602,14 @@ func TestActivateTokenPoolSyncSuccess(t *testing.T) {
 	pool := &core.TokenPool{
 		Namespace: "ns1",
 		Connector: "magic-tokens",
+		TX: core.TransactionRef{
+			ID: fftypes.NewUUID(),
+		},
 	}
 
-	mdi := am.database.(*databasemocks.Plugin)
 	mth := am.txHelper.(*txcommonmocks.Helper)
 	mom := am.operations.(*operationmocks.Manager)
-	mdi.On("GetOperations", context.Background(), "ns1", mock.Anything).Return(nil, nil, nil)
+	mth.On("FindOperationInTransaction", context.Background(), pool.TX.ID, core.OpTypeTokenActivatePool).Return(nil, nil)
 	mom.On("AddOrReuseOperation", context.Background(), mock.MatchedBy(func(op *core.Operation) bool {
 		return op.Type == core.OpTypeTokenActivatePool
 	})).Return(nil)
@@ -604,7 +621,6 @@ func TestActivateTokenPoolSyncSuccess(t *testing.T) {
 	err := am.ActivateTokenPool(context.Background(), pool)
 	assert.NoError(t, err)
 
-	mdi.AssertExpectations(t)
 	mth.AssertExpectations(t)
 	mom.AssertExpectations(t)
 }
diff --git a/internal/batch/batch_manager_test.go b/internal/batch/batch_manager_test.go
index 407c00466..04047202e 100644
--- a/internal/batch/batch_manager_test.go
+++ b/internal/batch/batch_manager_test.go
@@ -71,8 +71,8 @@ func TestE2EDispatchBroadcast(t *testing.T) {
 	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
 
 	readyForDispatch := make(chan bool)
-	waitForDispatch := make(chan *DispatchState)
-	handler := func(ctx context.Context, state *DispatchState) error {
+	waitForDispatch := make(chan *DispatchPayload)
+	handler := func(ctx context.Context, state *DispatchPayload) error {
 		_, ok := <-readyForDispatch
 		if !ok {
 			return nil
@@ -191,10 +191,10 @@ func TestE2EDispatchPrivateUnpinned(t *testing.T) {
 	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
 
 	readyForDispatch := make(chan bool)
-	waitForDispatch := make(chan *DispatchState)
+	waitForDispatch := make(chan *DispatchPayload)
 	var groupID fftypes.Bytes32
 	_ = groupID.UnmarshalText([]byte("44dc0861e69d9bab17dd5e90a8898c2ea156ad04e5fabf83119cc010486e6c1b"))
-	handler := func(ctx context.Context, state *DispatchState) error {
+	handler := func(ctx context.Context, state *DispatchPayload) error {
 		_, ok := <-readyForDispatch
 		if !ok {
 			return nil
@@ -371,7 +371,7 @@ func TestMessageSequencerMissingMessageData(t *testing.T) {
 	txHelper, _ := txcommon.NewTransactionHelper(ctx, "ns1", mdi, mdm, cmi)
 	bm, _ := NewBatchManager(context.Background(), "ns1", mdi, mdm, mim, txHelper)
 	bm.RegisterDispatcher("utdispatcher", core.TransactionTypeNone, []core.MessageType{core.MessageTypeBroadcast},
-		func(c context.Context, state *DispatchState) error {
+		func(c context.Context, state *DispatchPayload) error {
 			return nil
 		},
 		DispatcherOptions{BatchType: core.BatchTypeBroadcast},
@@ -418,7 +418,7 @@ func TestMessageSequencerUpdateMessagesFail(t *testing.T) {
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	bm, _ := NewBatchManager(ctx, "ns1", mdi, mdm, mim, txHelper)
 	bm.RegisterDispatcher("utdispatcher", core.TransactionTypeBatchPin, []core.MessageType{core.MessageTypeBroadcast},
-		func(c context.Context, state *DispatchState) error {
+		func(c context.Context, state *DispatchPayload) error {
 			return nil
 		},
 		DispatcherOptions{BatchMaxSize: 1, DisposeTimeout: 0},
@@ -476,7 +476,7 @@ func TestMessageSequencerDispatchFail(t *testing.T) {
 	ctx, cancelCtx := context.WithCancel(context.Background())
 	bm, _ := NewBatchManager(ctx, "ns1", mdi, mdm, mim, txHelper)
 	bm.RegisterDispatcher("utdispatcher", core.TransactionTypeBatchPin, []core.MessageType{core.MessageTypeBroadcast},
-		func(c context.Context, state *DispatchState) error {
+		func(c context.Context, state *DispatchPayload) error {
 			cancelCtx()
 			return fmt.Errorf("fizzle")
 		}, DispatcherOptions{BatchMaxSize: 1, DisposeTimeout: 0},
@@ -518,7 +518,7 @@ func TestMessageSequencerUpdateBatchFail(t *testing.T) {
 	mim.On("GetLocalNode", mock.Anything).Return(&core.Identity{}, nil)
 	bm, _ := NewBatchManager(ctx, "ns1", mdi, mdm, mim, txHelper)
 	bm.RegisterDispatcher("utdispatcher", core.TransactionTypeBatchPin, []core.MessageType{core.MessageTypeBroadcast},
-		func(c context.Context, state *DispatchState) error {
+		func(c context.Context, state *DispatchPayload) error {
 			return nil
 		},
 		DispatcherOptions{BatchMaxSize: 1, DisposeTimeout: 0},
diff --git a/internal/batch/batch_processor_test.go b/internal/batch/batch_processor_test.go
index 8d692bb38..7e0b09f5f 100644
--- a/internal/batch/batch_processor_test.go
+++ b/internal/batch/batch_processor_test.go
@@ -76,8 +76,8 @@ func TestUnfilledBatch(t *testing.T) {
 	log.SetLevel("debug")
 	coreconfig.Reset()
 
-	dispatched := make(chan *DispatchState)
-	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	dispatched := make(chan *DispatchPayload)
+	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		dispatched <- state
 		return nil
 	})
@@ -125,8 +125,8 @@ func TestBatchSizeOverflow(t *testing.T) {
 	log.SetLevel("debug")
 	coreconfig.Reset()
 
-	dispatched := make(chan *DispatchState)
-	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	dispatched := make(chan *DispatchPayload)
+	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		dispatched <- state
 		return nil
 	})
@@ -175,18 +175,18 @@ func TestBatchSizeOverflow(t *testing.T) {
 }
 
 func TestCloseToUnblockDispatch(t *testing.T) {
-	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		return fmt.Errorf("pop")
 	})
 	defer cancel()
 	bp.cancelCtx()
-	bp.dispatchBatch(&DispatchState{})
+	bp.dispatchBatch(&DispatchPayload{})
 	<-bp.done
 }
 
 func TestCloseToUnblockUpsertBatch(t *testing.T) {
 
-	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		return nil
 	})
 	defer cancel()
@@ -227,7 +227,7 @@ func TestCloseToUnblockUpsertBatch(t *testing.T) {
 }
 
 func TestInsertNewNonceFail(t *testing.T) {
-	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		return nil
 	})
 	defer cancel()
@@ -241,8 +241,8 @@ func TestInsertNewNonceFail(t *testing.T) {
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 
 	gid := fftypes.NewRandB32()
-	err := bp.sealBatch(&DispatchState{
-		Persisted: core.BatchPersisted{
+	err := bp.sealBatch(&DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				Group: gid,
 			},
@@ -263,7 +263,7 @@ func TestInsertNewNonceFail(t *testing.T) {
 }
 
 func TestUpdateExistingNonceFail(t *testing.T) {
-	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		return nil
 	})
 	defer cancel()
@@ -281,8 +281,8 @@ func TestUpdateExistingNonceFail(t *testing.T) {
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 
 	gid := fftypes.NewRandB32()
-	err := bp.sealBatch(&DispatchState{
-		Persisted: core.BatchPersisted{
+	err := bp.sealBatch(&DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				Group: gid,
 			},
@@ -303,7 +303,7 @@ func TestUpdateExistingNonceFail(t *testing.T) {
 }
 
 func TestGetNonceFail(t *testing.T) {
-	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		return nil
 	})
 	defer cancel()
@@ -316,8 +316,8 @@ func TestGetNonceFail(t *testing.T) {
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 
 	gid := fftypes.NewRandB32()
-	err := bp.sealBatch(&DispatchState{
-		Persisted: core.BatchPersisted{
+	err := bp.sealBatch(&DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				Group: gid,
 			},
@@ -338,7 +338,7 @@ func TestGetNonceFail(t *testing.T) {
 }
 
 func TestGetNonceMigrationFail(t *testing.T) {
-	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		return nil
 	})
 	defer cancel()
@@ -352,8 +352,8 @@ func TestGetNonceMigrationFail(t *testing.T) {
 	mdm.On("UpdateMessageIfCached", mock.Anything, mock.Anything).Return()
 
 	gid := fftypes.NewRandB32()
-	err := bp.sealBatch(&DispatchState{
-		Persisted: core.BatchPersisted{
+	err := bp.sealBatch(&DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				Group: gid,
 			},
@@ -374,7 +374,7 @@ func TestGetNonceMigrationFail(t *testing.T) {
 }
 
 func TestAddWorkInSort(t *testing.T) {
-	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		return nil
 	})
 	defer cancel()
@@ -397,7 +397,7 @@ func TestAddWorkInSort(t *testing.T) {
 }
 
 func TestStartQuiesceNonBlocking(t *testing.T) {
-	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		return nil
 	})
 	defer cancel()
@@ -409,8 +409,8 @@ func TestMarkMessageDispatchedUnpinnedOK(t *testing.T) {
 	log.SetLevel("debug")
 	coreconfig.Reset()
 
-	dispatched := make(chan *DispatchState)
-	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	dispatched := make(chan *DispatchPayload)
+	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		dispatched <- state
 		return nil
 	})
@@ -461,8 +461,8 @@ func TestMaskContextsRetryAfterPinsAssigned(t *testing.T) {
 	log.SetLevel("debug")
 	coreconfig.Reset()
 
-	dispatched := make(chan *DispatchState)
-	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	dispatched := make(chan *DispatchPayload)
+	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		dispatched <- state
 		return nil
 	})
@@ -505,7 +505,7 @@ func TestMaskContextsRetryAfterPinsAssigned(t *testing.T) {
 		},
 	}
 
-	state := bp.initFlushState(fftypes.NewUUID(), []*batchWork{{msg: msg1}, {msg: msg2}})
+	state := bp.initPayload(fftypes.NewUUID(), []*batchWork{{msg: msg1}, {msg: msg2}})
 	err := bp.sealBatch(state)
 	assert.NoError(t, err)
 
@@ -527,8 +527,8 @@ func TestMaskContextsUpdateMessageFail(t *testing.T) {
 	log.SetLevel("debug")
 	coreconfig.Reset()
 
-	dispatched := make(chan *DispatchState)
-	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+	dispatched := make(chan *DispatchPayload)
+	cancel, mdi, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchPayload) error {
 		dispatched <- state
 		return nil
 	})
@@ -551,7 +551,7 @@ func TestMaskContextsUpdateMessageFail(t *testing.T) {
 		},
 	}
 
-	state := bp.initFlushState(fftypes.NewUUID(), []*batchWork{{msg: msg}})
+	state := bp.initPayload(fftypes.NewUUID(), []*batchWork{{msg: msg}})
 	err := bp.sealBatch(state)
 	assert.Regexp(t, "FF00154", err)
 
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index d98ebb053..27fdaf67f 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -646,13 +646,17 @@ func TestInitAllExistingStreams(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x71C7656EC7ab88b098defB751B7401B5f6d8976F",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.NoError(t, err)
 
 	assert.Equal(t, 4, httpmock.GetTotalCallCount())
@@ -688,13 +692,17 @@ func TestInitAllExistingStreamsV1(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x71C7656EC7ab88b098defB751B7401B5f6d8976F",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.NoError(t, err)
 
 	assert.Equal(t, 4, httpmock.GetTotalCallCount())
@@ -730,13 +738,17 @@ func TestInitAllExistingStreamsOld(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x71C7656EC7ab88b098defB751B7401B5f6d8976F",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.NoError(t, err)
 
 	assert.Equal(t, 4, httpmock.GetTotalCallCount())
@@ -772,13 +784,17 @@ func TestInitAllExistingStreamsInvalidName(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x71C7656EC7ab88b098defB751B7401B5f6d8976F",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, e.metrics, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.Regexp(t, "FF10416", err)
 }
 
@@ -3374,9 +3390,13 @@ func TestAddSubBadLocation(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"bad": "bad",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err := e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err := e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.Regexp(t, "FF10310", err)
 }
 
@@ -3419,9 +3439,13 @@ func TestAddAndRemoveFireflySubscription(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x123",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "newest",
+	}
 
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	subID, err := e.AddFireflySubscription(e.ctx, ns, location, "newest")
+	subID, err := e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.NoError(t, err)
 	assert.NotNil(t, e.subs.GetSubscription("sub1"))
 
@@ -3462,9 +3486,13 @@ func TestAddFireflySubscriptionV1(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x123",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "newest",
+	}
 
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "newest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.NoError(t, err)
 	assert.NotNil(t, e.subs.GetSubscription("sub1"))
 }
@@ -3500,9 +3528,13 @@ func TestAddFireflySubscriptionQuerySubsFail(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x123",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.Regexp(t, "FF10111", err)
 }
 
@@ -3538,9 +3570,13 @@ func TestAddFireflySubscriptionCreateError(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x123",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.Regexp(t, "FF10111", err)
 }
 
@@ -3575,9 +3611,13 @@ func TestAddFireflySubscriptionGetVersionError(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"address": "0x123",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.Regexp(t, "FF10111", err)
 }
 
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index 783fdc344..de62a4f4b 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -302,13 +302,17 @@ func TestInitAllExistingStreams(t *testing.T) {
 		"channel":   "firefly",
 		"chaincode": "simplestorage",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, &metricsmocks.Manager{}, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.NoError(t, err)
 
 	assert.Equal(t, 3, httpmock.GetTotalCallCount())
@@ -343,13 +347,17 @@ func TestInitAllExistingStreamsV1(t *testing.T) {
 		"channel":   "firefly",
 		"chaincode": "simplestorage",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, &metricsmocks.Manager{}, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.NoError(t, err)
 
 	assert.Equal(t, 3, httpmock.GetTotalCallCount())
@@ -382,13 +390,17 @@ func TestAddFireflySubscriptionQuerySubsFail(t *testing.T) {
 		"channel":   "firefly",
 		"chaincode": "simplestorage",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "newest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, &metricsmocks.Manager{}, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "newest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.Regexp(t, "pop", err)
 }
 
@@ -420,13 +432,17 @@ func TestAddFireflySubscriptionGetVersionError(t *testing.T) {
 		"channel":   "firefly",
 		"chaincode": "simplestorage",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "newest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, &metricsmocks.Manager{}, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "newest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.Regexp(t, "pop", err)
 }
 
@@ -458,13 +474,17 @@ func TestAddAndRemoveFireflySubscriptionDeprecatedSubName(t *testing.T) {
 		"channel":   "firefly",
 		"chaincode": "simplestorage",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, &metricsmocks.Manager{}, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	subID, err := e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	subID, err := e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.NoError(t, err)
 
 	assert.Equal(t, 3, httpmock.GetTotalCallCount())
@@ -503,13 +523,17 @@ func TestAddFireflySubscriptionInvalidSubName(t *testing.T) {
 		"channel":   "firefly",
 		"chaincode": "simplestorage",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, &metricsmocks.Manager{}, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.Regexp(t, "FF10416", err)
 }
 
@@ -518,8 +542,12 @@ func TestAddFFSubscriptionBadLocation(t *testing.T) {
 	location := fftypes.JSONAnyPtr(fftypes.JSONObject{
 		"bad": "bad",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err := e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err := e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.Regexp(t, "F10310", err)
 }
 
@@ -637,13 +665,17 @@ func TestSubQueryCreateError(t *testing.T) {
 		"channel":   "firefly",
 		"chaincode": "simplestorage",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, &metricsmocks.Manager{}, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.Regexp(t, "FF10284.*pop", err)
 
 }
@@ -680,13 +712,17 @@ func TestSubQueryCreate(t *testing.T) {
 		"channel":   "firefly",
 		"chaincode": "simplestorage",
 	}.String())
+	contract := &blockchain.MultipartyContract{
+		Location:   location,
+		FirstEvent: "oldest",
+	}
 
 	cmi := &cachemocks.Manager{}
 	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(e.ctx, 100, 5*time.Minute), nil)
 	err := e.Init(e.ctx, e.cancelCtx, utConfig, &metricsmocks.Manager{}, cmi)
 	assert.NoError(t, err)
 	ns := &core.Namespace{Name: "ns1", NetworkName: "ns1"}
-	_, err = e.AddFireflySubscription(e.ctx, ns, location, "oldest")
+	_, err = e.AddFireflySubscription(e.ctx, ns, contract)
 	assert.NoError(t, err)
 
 }
diff --git a/internal/broadcast/manager_test.go b/internal/broadcast/manager_test.go
index 5ed4d95d0..d0cfaecc0 100644
--- a/internal/broadcast/manager_test.go
+++ b/internal/broadcast/manager_test.go
@@ -176,7 +176,7 @@ func TestDispatchBatchBlobsFail(t *testing.T) {
 	defer cancel()
 
 	blobHash := fftypes.NewRandB32()
-	state := &batch.DispatchState{
+	state := &batch.DispatchPayload{
 		Data: []*core.Data{
 			{ID: fftypes.NewUUID(), Blob: &core.BlobRef{
 				Hash: blobHash,
@@ -202,7 +202,7 @@ func TestDispatchBatchInsertOpFail(t *testing.T) {
 	bm, cancel := newTestBroadcast(t)
 	defer cancel()
 
-	state := &batch.DispatchState{
+	state := &batch.DispatchPayload{
 		Pins: []*fftypes.Bytes32{fftypes.NewRandB32()},
 	}
 
@@ -219,8 +219,8 @@ func TestDispatchBatchUploadFail(t *testing.T) {
 	bm, cancel := newTestBroadcast(t)
 	defer cancel()
 
-	state := &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	state := &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				ID: fftypes.NewUUID(),
 			},
@@ -232,7 +232,7 @@ func TestDispatchBatchUploadFail(t *testing.T) {
 	mom.On("AddOrReuseOperation", mock.Anything, mock.Anything).Return(nil)
 	mom.On("RunOperation", mock.Anything, mock.MatchedBy(func(op *core.PreparedOperation) bool {
 		data := op.Data.(uploadBatchData)
-		return op.Type == core.OpTypeSharedStorageUploadBatch && data.Batch.ID.Equals(state.Persisted.ID)
+		return op.Type == core.OpTypeSharedStorageUploadBatch && data.Batch.ID.Equals(state.Batch.ID)
 	}), operations.RemainPendingOnFailure).Return(nil, fmt.Errorf("pop"))
 
 	err := bm.dispatchBatch(context.Background(), state)
@@ -245,8 +245,8 @@ func TestDispatchBatchSubmitBatchPinSucceed(t *testing.T) {
 	bm, cancel := newTestBroadcast(t)
 	defer cancel()
 
-	state := &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	state := &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				ID: fftypes.NewUUID(),
 			},
@@ -261,7 +261,7 @@ func TestDispatchBatchSubmitBatchPinSucceed(t *testing.T) {
 	mmp.On("SubmitBatchPin", mock.Anything, mock.Anything, mock.Anything, "payload1").Return(nil)
 	mom.On("RunOperation", mock.Anything, mock.MatchedBy(func(op *core.PreparedOperation) bool {
 		data := op.Data.(uploadBatchData)
-		return op.Type == core.OpTypeSharedStorageUploadBatch && data.Batch.ID.Equals(state.Persisted.ID)
+		return op.Type == core.OpTypeSharedStorageUploadBatch && data.Batch.ID.Equals(state.Batch.ID)
 	}), operations.RemainPendingOnFailure).Return(getUploadBatchOutputs("payload1"), nil)
 
 	err := bm.dispatchBatch(context.Background(), state)
@@ -276,8 +276,8 @@ func TestDispatchBatchSubmitBroadcastFail(t *testing.T) {
 	bm, cancel := newTestBroadcast(t)
 	defer cancel()
 
-	state := &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	state := &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				ID:        fftypes.NewUUID(),
 				SignerRef: core.SignerRef{Author: "wrong", Key: "wrong"},
@@ -293,7 +293,7 @@ func TestDispatchBatchSubmitBroadcastFail(t *testing.T) {
 	mmp.On("SubmitBatchPin", mock.Anything, mock.Anything, mock.Anything, "payload1").Return(fmt.Errorf("pop"))
 	mom.On("RunOperation", mock.Anything, mock.MatchedBy(func(op *core.PreparedOperation) bool {
 		data := op.Data.(uploadBatchData)
-		return op.Type == core.OpTypeSharedStorageUploadBatch && data.Batch.ID.Equals(state.Persisted.ID)
+		return op.Type == core.OpTypeSharedStorageUploadBatch && data.Batch.ID.Equals(state.Batch.ID)
 	}), operations.RemainPendingOnFailure).Return(getUploadBatchOutputs("payload1"), nil)
 
 	err := bm.dispatchBatch(context.Background(), state)
diff --git a/internal/events/dx_callbacks_test.go b/internal/events/dx_callbacks_test.go
index eacbb763f..2f356d473 100644
--- a/internal/events/dx_callbacks_test.go
+++ b/internal/events/dx_callbacks_test.go
@@ -179,7 +179,6 @@ func TestMessageReceiveOkBadBatchIgnored(t *testing.T) {
 
 	data := &core.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, core.BatchTypePrivate, core.TransactionTypeBatchPin, core.DataArray{data})
-	batch.Payload.TX.Type = core.TransactionTypeTokenPool
 	b := &core.TransportWrapper{
 		Batch: batch,
 		Group: &core.Group{
diff --git a/internal/events/token_pool_created_test.go b/internal/events/token_pool_created_test.go
index 5723d60a2..eda4725d1 100644
--- a/internal/events/token_pool_created_test.go
+++ b/internal/events/token_pool_created_test.go
@@ -35,7 +35,6 @@ func TestTokenPoolCreatedIgnore(t *testing.T) {
 	mti := &tokenmocks.Plugin{}
 
 	txID := fftypes.NewUUID()
-	operations := []*core.Operation{}
 	info := fftypes.JSONObject{"some": "info"}
 	pool := &tokens.TokenPool{
 		Type:        core.TokenTypeFungible,
@@ -53,7 +52,7 @@ func TestTokenPoolCreatedIgnore(t *testing.T) {
 	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, txID, core.OpTypeTokenCreatePool).Return(nil, nil)
 
 	err := em.TokenPoolCreated(mti, pool)
 	assert.NoError(t, err)
@@ -216,9 +215,9 @@ func TestTokenPoolCreatedConfirmFailBadSymbol(t *testing.T) {
 	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(storedPool, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return([]*core.Operation{{
+	em.mth.On("FindOperationInTransaction", em.ctx, txID, core.OpTypeTokenCreatePool).Return(&core.Operation{
 		ID: opID,
-	}}, nil, nil)
+	}, nil)
 
 	err := em.TokenPoolCreated(mti, chainPool)
 	assert.NoError(t, err)
@@ -331,14 +330,12 @@ func TestTokenPoolCreatedAnnounce(t *testing.T) {
 
 	poolID := fftypes.NewUUID()
 	txID := fftypes.NewUUID()
-	operations := []*core.Operation{
-		{
-			ID: fftypes.NewUUID(),
-			Input: fftypes.JSONObject{
-				"id":        poolID.String(),
-				"namespace": "ns1",
-				"name":      "my-pool",
-			},
+	operation := &core.Operation{
+		ID: fftypes.NewUUID(),
+		Input: fftypes.JSONObject{
+			"id":        poolID.String(),
+			"namespace": "ns1",
+			"name":      "my-pool",
 		},
 	}
 	info := fftypes.JSONObject{"some": "info"}
@@ -358,8 +355,8 @@ func TestTokenPoolCreatedAnnounce(t *testing.T) {
 	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil).Times(2)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop")).Once()
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil).Once()
+	em.mth.On("FindOperationInTransaction", em.ctx, txID, core.OpTypeTokenCreatePool).Return(nil, fmt.Errorf("pop")).Once()
+	em.mth.On("FindOperationInTransaction", em.ctx, txID, core.OpTypeTokenCreatePool).Return(operation, nil).Once()
 	em.mds.On("DefineTokenPool", em.ctx, mock.MatchedBy(func(pool *core.TokenPoolAnnouncement) bool {
 		return pool.Pool.Namespace == "ns1" && pool.Pool.Name == "my-pool" && *pool.Pool.ID == *poolID
 	}), false).Return(nil, nil)
@@ -378,16 +375,14 @@ func TestTokenPoolCreatedAnnounceBadInterface(t *testing.T) {
 	poolID := fftypes.NewUUID()
 	txID := fftypes.NewUUID()
 	interfaceID := fftypes.NewUUID()
-	operations := []*core.Operation{
-		{
-			ID: fftypes.NewUUID(),
-			Input: fftypes.JSONObject{
-				"id":        poolID.String(),
-				"namespace": "ns1",
-				"name":      "my-pool",
-				"interface": fftypes.JSONObject{
-					"id": interfaceID.String(),
-				},
+	operation := &core.Operation{
+		ID: fftypes.NewUUID(),
+		Input: fftypes.JSONObject{
+			"id":        poolID.String(),
+			"namespace": "ns1",
+			"name":      "my-pool",
+			"interface": fftypes.JSONObject{
+				"id": interfaceID.String(),
 			},
 		},
 	}
@@ -409,8 +404,8 @@ func TestTokenPoolCreatedAnnounceBadInterface(t *testing.T) {
 	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil).Times(2)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop")).Once()
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil).Once()
+	em.mth.On("FindOperationInTransaction", em.ctx, txID, core.OpTypeTokenCreatePool).Return(nil, fmt.Errorf("pop")).Once()
+	em.mth.On("FindOperationInTransaction", em.ctx, txID, core.OpTypeTokenCreatePool).Return(operation, nil).Once()
 	em.mam.On("ResolvePoolMethods", em.ctx, mock.MatchedBy(func(pool *core.TokenPool) bool {
 		return pool.Locator == "123" && pool.Name == "my-pool"
 	})).Return(fmt.Errorf("pop"))
@@ -427,12 +422,10 @@ func TestTokenPoolCreatedAnnounceBadOpInputID(t *testing.T) {
 	mti := &tokenmocks.Plugin{}
 
 	txID := fftypes.NewUUID()
-	operations := []*core.Operation{
-		{
-			ID:    fftypes.NewUUID(),
-			Type:  core.OpTypeTokenCreatePool,
-			Input: fftypes.JSONObject{},
-		},
+	operation := &core.Operation{
+		ID:    fftypes.NewUUID(),
+		Type:  core.OpTypeTokenCreatePool,
+		Input: fftypes.JSONObject{},
 	}
 	info := fftypes.JSONObject{"some": "info"}
 	pool := &tokens.TokenPool{
@@ -451,7 +444,7 @@ func TestTokenPoolCreatedAnnounceBadOpInputID(t *testing.T) {
 	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, txID, core.OpTypeTokenCreatePool).Return(operation, nil)
 
 	err := em.TokenPoolCreated(mti, pool)
 	assert.NoError(t, err)
@@ -464,13 +457,11 @@ func TestTokenPoolCreatedAnnounceBadOpInputNS(t *testing.T) {
 	mti := &tokenmocks.Plugin{}
 
 	txID := fftypes.NewUUID()
-	operations := []*core.Operation{
-		{
-			ID:   fftypes.NewUUID(),
-			Type: core.OpTypeTokenCreatePool,
-			Input: fftypes.JSONObject{
-				"id": fftypes.NewUUID().String(),
-			},
+	operation := &core.Operation{
+		ID:   fftypes.NewUUID(),
+		Type: core.OpTypeTokenCreatePool,
+		Input: fftypes.JSONObject{
+			"id": fftypes.NewUUID().String(),
 		},
 	}
 	info := fftypes.JSONObject{"some": "info"}
@@ -490,7 +481,7 @@ func TestTokenPoolCreatedAnnounceBadOpInputNS(t *testing.T) {
 	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, txID, core.OpTypeTokenCreatePool).Return(operation, nil)
 
 	err := em.TokenPoolCreated(mti, pool)
 	assert.NoError(t, err)
@@ -504,15 +495,13 @@ func TestTokenPoolCreatedAnnounceBadSymbol(t *testing.T) {
 
 	poolID := fftypes.NewUUID()
 	txID := fftypes.NewUUID()
-	operations := []*core.Operation{
-		{
-			ID: fftypes.NewUUID(),
-			Input: fftypes.JSONObject{
-				"id":        poolID.String(),
-				"namespace": "test-ns",
-				"name":      "my-pool",
-				"symbol":    "FFT",
-			},
+	operation := &core.Operation{
+		ID: fftypes.NewUUID(),
+		Input: fftypes.JSONObject{
+			"id":        poolID.String(),
+			"namespace": "test-ns",
+			"name":      "my-pool",
+			"symbol":    "FFT",
 		},
 	}
 	info := fftypes.JSONObject{"some": "info"}
@@ -533,8 +522,8 @@ func TestTokenPoolCreatedAnnounceBadSymbol(t *testing.T) {
 	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil).Times(2)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop")).Once()
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil).Once()
+	em.mth.On("FindOperationInTransaction", em.ctx, txID, core.OpTypeTokenCreatePool).Return(nil, fmt.Errorf("pop")).Once()
+	em.mth.On("FindOperationInTransaction", em.ctx, txID, core.OpTypeTokenCreatePool).Return(operation, nil).Once()
 
 	err := em.TokenPoolCreated(mti, pool)
 	assert.NoError(t, err)
diff --git a/internal/events/tokens_approved_test.go b/internal/events/tokens_approved_test.go
index a50c5c11d..489498635 100644
--- a/internal/events/tokens_approved_test.go
+++ b/internal/events/tokens_approved_test.go
@@ -124,7 +124,7 @@ func TestPersistApprovalOpFail(t *testing.T) {
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
 	em.mdi.On("GetTokenApprovalByProtocolID", em.ctx, "ns1", approval.Connector, approval.ProtocolID).Return(nil, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
+	em.mth.On("FindOperationInTransaction", em.ctx, approval.TX.ID, core.OpTypeTokenApproval).Return(nil, fmt.Errorf("pop"))
 
 	valid, err := em.persistTokenApproval(em.ctx, approval)
 	assert.False(t, valid)
@@ -141,16 +141,16 @@ func TestPersistApprovalBadOp(t *testing.T) {
 		ID:        fftypes.NewUUID(),
 		Namespace: "ns1",
 	}
-	ops := []*core.Operation{{
+	op := &core.Operation{
 		Input: fftypes.JSONObject{
 			"localId": "realbad",
 		},
 		Transaction: fftypes.NewUUID(),
-	}}
+	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
 	em.mdi.On("GetTokenApprovalByProtocolID", em.ctx, "ns1", approval.Connector, approval.ProtocolID).Return(nil, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(ops, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, approval.TX.ID, core.OpTypeTokenApproval).Return(op, nil)
 	em.mth.On("PersistTransaction", mock.Anything, approval.TX.ID, core.TransactionTypeTokenApproval, "0xffffeeee").Return(false, fmt.Errorf("pop"))
 
 	valid, err := em.persistTokenApproval(em.ctx, approval)
@@ -169,17 +169,17 @@ func TestPersistApprovalTxFail(t *testing.T) {
 		Namespace: "ns1",
 	}
 	localID := fftypes.NewUUID()
-	ops := []*core.Operation{{
+	op := &core.Operation{
 		Input: fftypes.JSONObject{
 			"localId":   localID.String(),
 			"connector": approval.Connector,
 			"pool":      pool.ID.String(),
 		},
-	}}
+	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
 	em.mdi.On("GetTokenApprovalByProtocolID", em.ctx, "ns1", approval.Connector, approval.ProtocolID).Return(nil, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(ops, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, approval.TX.ID, core.OpTypeTokenApproval).Return(op, nil)
 	em.mdi.On("GetTokenApprovalByID", em.ctx, "ns1", localID).Return(nil, nil)
 	em.mth.On("PersistTransaction", mock.Anything, approval.TX.ID, core.TransactionTypeTokenApproval, "0xffffeeee").Return(false, fmt.Errorf("pop"))
 
@@ -199,17 +199,17 @@ func TestPersistApprovalGetApprovalFail(t *testing.T) {
 		Namespace: "ns1",
 	}
 	localID := fftypes.NewUUID()
-	ops := []*core.Operation{{
+	op := &core.Operation{
 		Input: fftypes.JSONObject{
 			"localId":   localID.String(),
 			"connector": approval.Connector,
 			"pool":      pool.ID.String(),
 		},
-	}}
+	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
 	em.mdi.On("GetTokenApprovalByProtocolID", em.ctx, "ns1", approval.Connector, approval.ProtocolID).Return(nil, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(ops, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, approval.TX.ID, core.OpTypeTokenApproval).Return(op, nil)
 	em.mdi.On("GetTokenApprovalByID", em.ctx, "ns1", localID).Return(nil, fmt.Errorf("pop"))
 
 	valid, err := em.persistTokenApproval(em.ctx, approval)
@@ -245,17 +245,17 @@ func TestApprovedWithTransactionRegenerateLocalID(t *testing.T) {
 		Namespace: "ns1",
 	}
 	localID := fftypes.NewUUID()
-	ops := []*core.Operation{{
+	op := &core.Operation{
 		Input: fftypes.JSONObject{
 			"localId":   localID.String(),
 			"connector": approval.Connector,
 			"pool":      pool.ID.String(),
 		},
-	}}
+	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
 	em.mdi.On("GetTokenApprovalByProtocolID", em.ctx, "ns1", approval.Connector, approval.ProtocolID).Return(nil, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(ops, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, approval.TX.ID, core.OpTypeTokenApproval).Return(op, nil)
 	em.mth.On("PersistTransaction", mock.Anything, approval.TX.ID, core.TransactionTypeTokenApproval, "0xffffeeee").Return(true, nil)
 	em.mdi.On("GetTokenApprovalByID", em.ctx, "ns1", localID).Return(&core.TokenApproval{}, nil)
 	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
@@ -286,17 +286,17 @@ func TestApprovedBlockchainEventFail(t *testing.T) {
 		Namespace: "ns1",
 	}
 	localID := fftypes.NewUUID()
-	ops := []*core.Operation{{
+	op := &core.Operation{
 		Input: fftypes.JSONObject{
 			"localId":   localID.String(),
 			"connector": approval.Connector,
 			"pool":      pool.ID.String(),
 		},
-	}}
+	}
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
 	em.mdi.On("GetTokenApprovalByProtocolID", em.ctx, "ns1", approval.Connector, approval.ProtocolID).Return(nil, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(ops, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, approval.TX.ID, core.OpTypeTokenApproval).Return(op, nil)
 	em.mth.On("PersistTransaction", mock.Anything, approval.TX.ID, core.TransactionTypeTokenApproval, "0xffffeeee").Return(true, nil)
 	em.mdi.On("GetTokenApprovalByID", em.ctx, "ns1", localID).Return(&core.TokenApproval{}, nil)
 	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
diff --git a/internal/events/tokens_transferred_test.go b/internal/events/tokens_transferred_test.go
index f0c090a24..5f8c26f52 100644
--- a/internal/events/tokens_transferred_test.go
+++ b/internal/events/tokens_transferred_test.go
@@ -126,7 +126,7 @@ func TestPersistTransferOpFail(t *testing.T) {
 
 	em.mdi.On("GetTokenTransferByProtocolID", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
+	em.mth.On("FindOperationInTransaction", em.ctx, transfer.TX.ID, core.OpTypeTokenTransfer).Return(nil, fmt.Errorf("pop"))
 
 	valid, err := em.persistTokenTransfer(em.ctx, transfer)
 	assert.False(t, valid)
@@ -143,16 +143,16 @@ func TestPersistTransferBadOp(t *testing.T) {
 		ID:        fftypes.NewUUID(),
 		Namespace: "ns1",
 	}
-	ops := []*core.Operation{{
+	op := &core.Operation{
 		Input: fftypes.JSONObject{
 			"localId": "bad",
 		},
 		Transaction: fftypes.NewUUID(),
-	}}
+	}
 
 	em.mdi.On("GetTokenTransferByProtocolID", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(ops, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, transfer.TX.ID, core.OpTypeTokenTransfer).Return(op, nil)
 	em.mth.On("PersistTransaction", mock.Anything, transfer.TX.ID, core.TransactionTypeTokenTransfer, "0xffffeeee").Return(false, fmt.Errorf("pop"))
 
 	valid, err := em.persistTokenTransfer(em.ctx, transfer)
@@ -171,15 +171,15 @@ func TestPersistTransferTxFail(t *testing.T) {
 		Namespace: "ns1",
 	}
 	localID := fftypes.NewUUID()
-	ops := []*core.Operation{{
+	op := &core.Operation{
 		Input: fftypes.JSONObject{
 			"localId": localID.String(),
 		},
-	}}
+	}
 
 	em.mdi.On("GetTokenTransferByProtocolID", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(ops, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, transfer.TX.ID, core.OpTypeTokenTransfer).Return(op, nil)
 	em.mth.On("PersistTransaction", mock.Anything, transfer.TX.ID, core.TransactionTypeTokenTransfer, "0xffffeeee").Return(false, fmt.Errorf("pop"))
 
 	valid, err := em.persistTokenTransfer(em.ctx, transfer)
@@ -198,17 +198,17 @@ func TestPersistTransferGetTransferFail(t *testing.T) {
 		Namespace: "ns1",
 	}
 	localID := fftypes.NewUUID()
-	ops := []*core.Operation{{
+	op := &core.Operation{
 		Input: fftypes.JSONObject{
 			"localId":   localID.String(),
 			"connector": transfer.Connector,
 			"pool":      pool.ID.String(),
 		},
-	}}
+	}
 
 	em.mdi.On("GetTokenTransferByProtocolID", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(ops, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, transfer.TX.ID, core.OpTypeTokenTransfer).Return(op, nil)
 	em.mdi.On("GetTokenTransferByID", em.ctx, "ns1", localID).Return(nil, fmt.Errorf("pop"))
 
 	valid, err := em.persistTokenTransfer(em.ctx, transfer)
@@ -227,17 +227,17 @@ func TestPersistTransferBlockchainEventFail(t *testing.T) {
 		Namespace: "ns1",
 	}
 	localID := fftypes.NewUUID()
-	ops := []*core.Operation{{
+	op := &core.Operation{
 		Input: fftypes.JSONObject{
 			"localId":   localID.String(),
 			"connector": transfer.Connector,
 			"pool":      pool.ID.String(),
 		},
-	}}
+	}
 
 	em.mdi.On("GetTokenTransferByProtocolID", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(ops, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, transfer.TX.ID, core.OpTypeTokenTransfer).Return(op, nil)
 	em.mth.On("PersistTransaction", mock.Anything, transfer.TX.ID, core.TransactionTypeTokenTransfer, "0xffffeeee").Return(true, nil)
 	em.mdi.On("GetTokenTransferByID", em.ctx, "ns1", localID).Return(nil, nil)
 	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
@@ -262,17 +262,17 @@ func TestTokensTransferredWithTransactionRegenerateLocalID(t *testing.T) {
 		Namespace: "ns1",
 	}
 	localID := fftypes.NewUUID()
-	operations := []*core.Operation{{
+	op := &core.Operation{
 		Input: fftypes.JSONObject{
 			"localId":   localID.String(),
 			"connector": transfer.Connector,
 			"pool":      pool.ID.String(),
 		},
-	}}
+	}
 
 	em.mdi.On("GetTokenTransferByProtocolID", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "F1").Return(pool, nil)
-	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil)
+	em.mth.On("FindOperationInTransaction", em.ctx, transfer.TX.ID, core.OpTypeTokenTransfer).Return(op, nil)
 	em.mth.On("PersistTransaction", mock.Anything, transfer.TX.ID, core.TransactionTypeTokenTransfer, "0xffffeeee").Return(true, nil)
 	em.mdi.On("GetTokenTransferByID", em.ctx, "ns1", localID).Return(&core.TokenTransfer{}, nil)
 	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
diff --git a/internal/multiparty/manager_test.go b/internal/multiparty/manager_test.go
index 102e7a602..4495da42c 100644
--- a/internal/multiparty/manager_test.go
+++ b/internal/multiparty/manager_test.go
@@ -84,7 +84,7 @@ func TestNewMultipartyManager(t *testing.T) {
 	config := Config{
 		Org:       RootOrg{Name: "org1"},
 		Node:      LocalNode{Name: "node1"},
-		Contracts: []Contract{},
+		Contracts: []blockchain.MultipartyContract{},
 	}
 	mom.On("RegisterHandler", mock.Anything, mock.Anything, []core.OpType{
 		core.OpTypeBlockchainPinBatch,
@@ -100,7 +100,7 @@ func TestNewMultipartyManager(t *testing.T) {
 }
 
 func TestInitFail(t *testing.T) {
-	config := Config{Contracts: []Contract{}}
+	config := Config{Contracts: []blockchain.MultipartyContract{}}
 	_, err := NewMultipartyManager(context.Background(), &core.Namespace{}, config, nil, nil, nil, nil, nil)
 	assert.Regexp(t, "FF10128", err)
 }
@@ -117,7 +117,7 @@ func TestConfigureContract(t *testing.T) {
 	mp.mbi.On("AddFireflySubscription", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("test", nil)
 	mp.mdi.On("UpsertNamespace", mock.Anything, mock.AnythingOfType("*core.Namespace"), true).Return(nil)
 
-	mp.multipartyManager.config.Contracts = []Contract{{
+	mp.multipartyManager.config.Contracts = []blockchain.MultipartyContract{{
 		FirstEvent: "0",
 		Location:   location,
 	}}
@@ -147,7 +147,7 @@ func TestConfigureContractLocationChanged(t *testing.T) {
 			Location: location,
 		},
 	}
-	mp.multipartyManager.config.Contracts = []Contract{{
+	mp.multipartyManager.config.Contracts = []blockchain.MultipartyContract{{
 		FirstEvent: "0",
 		Location:   location2,
 	}}
@@ -184,7 +184,7 @@ func TestResolveContractDeprecatedConfigError(t *testing.T) {
 
 	mp.mbi.On("GetAndConvertDeprecatedContractConfig", context.Background()).Return(nil, "", fmt.Errorf("pop"))
 
-	_, _, err := mp.resolveFireFlyContract(context.Background(), 0)
+	_, err := mp.resolveFireFlyContract(context.Background(), 0)
 	assert.Regexp(t, "pop", err)
 }
 
@@ -196,7 +196,7 @@ func TestResolveContractDeprecatedConfigNewestBlock(t *testing.T) {
 		"address": "0x123",
 	}.String()), "newst", nil)
 
-	_, _, err := mp.resolveFireFlyContract(context.Background(), 0)
+	_, err := mp.resolveFireFlyContract(context.Background(), 0)
 	assert.NoError(t, err)
 }
 
@@ -211,7 +211,7 @@ func TestConfigureContractBadIndex(t *testing.T) {
 	mp.multipartyManager.namespace.Contracts = &core.MultipartyContracts{
 		Active: &core.MultipartyContract{Index: 1},
 	}
-	mp.multipartyManager.config.Contracts = []Contract{{
+	mp.multipartyManager.config.Contracts = []blockchain.MultipartyContract{{
 		FirstEvent: "0",
 		Location:   location,
 	}}
@@ -233,7 +233,7 @@ func TestConfigureContractNetworkVersionFail(t *testing.T) {
 	mp.multipartyManager.namespace.Contracts = &core.MultipartyContracts{
 		Active: &core.MultipartyContract{Index: 0},
 	}
-	mp.multipartyManager.config.Contracts = []Contract{{
+	mp.multipartyManager.config.Contracts = []blockchain.MultipartyContract{{
 		FirstEvent: "0",
 		Location:   location,
 	}}
@@ -254,7 +254,7 @@ func TestSubmitNetworkAction(t *testing.T) {
 	mp.multipartyManager.namespace.Contracts = &core.MultipartyContracts{
 		Active: &core.MultipartyContract{Index: 0},
 	}
-	mp.multipartyManager.config.Contracts = []Contract{{
+	mp.multipartyManager.config.Contracts = []blockchain.MultipartyContract{{
 		FirstEvent: "0",
 		Location:   location,
 	}}
@@ -297,7 +297,7 @@ func TestSubmitNetworkActionTXFail(t *testing.T) {
 	mp.multipartyManager.namespace.Contracts = &core.MultipartyContracts{
 		Active: &core.MultipartyContract{Index: 0},
 	}
-	mp.multipartyManager.config.Contracts = []Contract{{
+	mp.multipartyManager.config.Contracts = []blockchain.MultipartyContract{{
 		FirstEvent: "0",
 		Location:   location,
 	}}
@@ -328,7 +328,7 @@ func TestSubmitNetworkActionOpFail(t *testing.T) {
 	mp.multipartyManager.namespace.Contracts = &core.MultipartyContracts{
 		Active: &core.MultipartyContract{Index: 0},
 	}
-	mp.multipartyManager.config.Contracts = []Contract{{
+	mp.multipartyManager.config.Contracts = []blockchain.MultipartyContract{{
 		FirstEvent: "0",
 		Location:   location,
 	}}
@@ -365,7 +365,7 @@ func TestSubmitNetworkActionBadType(t *testing.T) {
 	mp.multipartyManager.namespace.Contracts = &core.MultipartyContracts{
 		Active: &core.MultipartyContract{Index: 0},
 	}
-	mp.multipartyManager.config.Contracts = []Contract{{
+	mp.multipartyManager.config.Contracts = []blockchain.MultipartyContract{{
 		FirstEvent: "0",
 		Location:   location,
 	}}
@@ -493,7 +493,7 @@ func TestGetNetworkVersion(t *testing.T) {
 	mp.multipartyManager.namespace.Contracts = &core.MultipartyContracts{
 		Active: &core.MultipartyContract{Index: 0},
 	}
-	mp.multipartyManager.config.Contracts = []Contract{{
+	mp.multipartyManager.config.Contracts = []blockchain.MultipartyContract{{
 		FirstEvent: "0",
 		Location:   location,
 	}}
@@ -521,7 +521,7 @@ func TestConfgureAndTerminateContract(t *testing.T) {
 	mp.multipartyManager.namespace.Contracts = &core.MultipartyContracts{
 		Active: &core.MultipartyContract{Index: 0},
 	}
-	mp.multipartyManager.config.Contracts = []Contract{{
+	mp.multipartyManager.config.Contracts = []blockchain.MultipartyContract{{
 		FirstEvent: "0",
 		Location:   location,
 	}, {
diff --git a/internal/orchestrator/data_query_test.go b/internal/orchestrator/data_query_test.go
index b748fb3be..8fb93b55e 100644
--- a/internal/orchestrator/data_query_test.go
+++ b/internal/orchestrator/data_query_test.go
@@ -260,12 +260,7 @@ func TestGetMessageTransactionOk(t *testing.T) {
 			Namespace: "ns",
 			TxType:    core.TransactionTypeBatchPin,
 		},
-	}, nil)
-	or.mdi.On("GetBatchByID", mock.Anything, "ns", batchID).Return(&core.BatchPersisted{
-		TX: core.TransactionRef{
-			Type: core.TransactionTypeBatchPin,
-			ID:   txID,
-		},
+		TransactionID: txID,
 	}, nil)
 	or.mth.On("GetTransactionByIDCached", mock.Anything, txID).Return(&core.Transaction{
 		ID: txID,
@@ -276,71 +271,25 @@ func TestGetMessageTransactionOk(t *testing.T) {
 	or.mdi.AssertExpectations(t)
 }
 
-func TestGetMessageTransactionNoBatchTX(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	msgID := fftypes.NewUUID()
-	batchID := fftypes.NewUUID()
-	or.mdi.On("GetMessageByID", mock.Anything, "ns", msgID).Return(&core.Message{
-		BatchID: batchID,
-		Header: core.MessageHeader{
-			Namespace: "ns",
-			TxType:    core.TransactionTypeBatchPin,
-		},
-	}, nil)
-	or.mdi.On("GetBatchByID", mock.Anything, "ns", batchID).Return(&core.BatchPersisted{}, nil)
-	_, err := or.GetMessageTransaction(context.Background(), msgID.String())
-	assert.Regexp(t, "FF10210", err)
-}
-
-func TestGetMessageTransactionNoBatch(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	msgID := fftypes.NewUUID()
-	batchID := fftypes.NewUUID()
-	or.mdi.On("GetMessageByID", mock.Anything, "ns", msgID).Return(&core.Message{
-		BatchID: batchID,
-		Header: core.MessageHeader{
-			Namespace: "ns",
-			TxType:    core.TransactionTypeBatchPin,
-		},
-	}, nil)
-	or.mdi.On("GetBatchByID", mock.Anything, "ns", batchID).Return(nil, nil)
-	_, err := or.GetMessageTransaction(context.Background(), msgID.String())
-	assert.Regexp(t, "FF10209", err)
-}
-
-func TestGetMessageTransactionBatchLookupErr(t *testing.T) {
+func TestGetMessageTransactionLookupErr(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
 	msgID := fftypes.NewUUID()
 	batchID := fftypes.NewUUID()
+	txID := fftypes.NewUUID()
 	or.mdi.On("GetMessageByID", mock.Anything, "ns", msgID).Return(&core.Message{
 		BatchID: batchID,
 		Header: core.MessageHeader{
 			Namespace: "ns",
 			TxType:    core.TransactionTypeBatchPin,
 		},
+		TransactionID: txID,
 	}, nil)
-	or.mdi.On("GetBatchByID", mock.Anything, "ns", batchID).Return(nil, fmt.Errorf("pop"))
+	or.mth.On("GetTransactionByIDCached", mock.Anything, txID).Return(nil, fmt.Errorf("pop"))
 	_, err := or.GetMessageTransaction(context.Background(), msgID.String())
 	assert.Regexp(t, "pop", err)
 }
 
-func TestGetMessageTransactionNoBatchID(t *testing.T) {
-	or := newTestOrchestrator()
-	defer or.cleanup(t)
-	msgID := fftypes.NewUUID()
-	or.mdi.On("GetMessageByID", mock.Anything, "ns", msgID).Return(&core.Message{
-		Header: core.MessageHeader{
-			Namespace: "ns",
-			TxType:    core.TransactionTypeBatchPin,
-		},
-	}, nil)
-	_, err := or.GetMessageTransaction(context.Background(), msgID.String())
-	assert.Regexp(t, "FF10208", err)
-}
-
 func TestGetMessageTransactionNoTx(t *testing.T) {
 	or := newTestOrchestrator()
 	defer or.cleanup(t)
@@ -351,7 +300,7 @@ func TestGetMessageTransactionNoTx(t *testing.T) {
 		},
 	}, nil)
 	_, err := or.GetMessageTransaction(context.Background(), msgID.String())
-	assert.Regexp(t, "FF10207", err)
+	assert.Regexp(t, "FF10210", err)
 }
 
 func TestGetMessageTransactionMessageNotFound(t *testing.T) {
diff --git a/internal/privatemessaging/message_test.go b/internal/privatemessaging/message_test.go
index ce3d54a70..3d853f285 100644
--- a/internal/privatemessaging/message_test.go
+++ b/internal/privatemessaging/message_test.go
@@ -388,8 +388,8 @@ func TestSendUnpinnedMessageGroupLookupFail(t *testing.T) {
 	mdi := pm.database.(*databasemocks.Plugin)
 	mdi.On("GetGroupByHash", pm.ctx, "ns1", groupID).Return(nil, fmt.Errorf("pop")).Once()
 
-	err := pm.dispatchUnpinnedBatch(pm.ctx, &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	err := pm.dispatchUnpinnedBatch(pm.ctx, &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				ID:    fftypes.NewUUID(),
 				Group: groupID,
@@ -665,8 +665,8 @@ func TestDispatchedUnpinnedMessageOK(t *testing.T) {
 		return op.Type == core.OpTypeDataExchangeSendBatch && *data.Node.ID == *node2.ID
 	})).Return(nil, nil)
 
-	err := pm.dispatchUnpinnedBatch(pm.ctx, &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	err := pm.dispatchUnpinnedBatch(pm.ctx, &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				ID:        fftypes.NewUUID(),
 				Group:     groupID,
diff --git a/internal/privatemessaging/privatemessaging_test.go b/internal/privatemessaging/privatemessaging_test.go
index 2db370592..9116f6f8a 100644
--- a/internal/privatemessaging/privatemessaging_test.go
+++ b/internal/privatemessaging/privatemessaging_test.go
@@ -218,8 +218,8 @@ func TestDispatchBatchWithBlobs(t *testing.T) {
 
 	mmp.On("SubmitBatchPin", pm.ctx, mock.Anything, mock.Anything, "").Return(nil)
 
-	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				ID: batchID,
 				SignerRef: core.SignerRef{
@@ -260,7 +260,7 @@ func TestDispatchErrorFindingGroup(t *testing.T) {
 	mdi := pm.database.(*databasemocks.Plugin)
 	mdi.On("GetGroupByHash", pm.ctx, "ns1", mock.Anything).Return(nil, fmt.Errorf("pop"))
 
-	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{})
+	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchPayload{})
 	assert.Regexp(t, "pop", err)
 }
 
@@ -271,8 +271,8 @@ func TestSendAndSubmitBatchBadID(t *testing.T) {
 	mdi := pm.database.(*databasemocks.Plugin)
 	mdi.On("GetGroupByHash", pm.ctx, "ns1", mock.Anything).Return(nil, fmt.Errorf("pop"))
 
-	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				SignerRef: core.SignerRef{
 					Author: "badauthor",
@@ -310,8 +310,8 @@ func TestSendAndSubmitBatchUnregisteredNode(t *testing.T) {
 
 	mim.On("GetLocalNode", pm.ctx).Return(nil, fmt.Errorf("pop"))
 
-	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				Group: groupID,
 				SignerRef: core.SignerRef{
@@ -334,8 +334,8 @@ func TestSendImmediateFail(t *testing.T) {
 	mdi := pm.database.(*databasemocks.Plugin)
 	mdi.On("GetGroupByHash", pm.ctx, "ns1", mock.Anything).Return(nil, fmt.Errorf("pop"))
 
-	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				SignerRef: core.SignerRef{
 					Author: "org1",
@@ -376,8 +376,8 @@ func TestSendSubmitInsertOperationFail(t *testing.T) {
 	mom := pm.operations.(*operationmocks.Manager)
 	mom.On("AddOrReuseOperation", pm.ctx, mock.Anything).Return(fmt.Errorf("pop"))
 
-	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				Group: groupID,
 				SignerRef: core.SignerRef{
@@ -432,8 +432,8 @@ func TestSendSubmitBlobTransferFail(t *testing.T) {
 		return op.Type == core.OpTypeDataExchangeSendBlob && *data.Node.ID == *node2.ID
 	})).Return(nil, fmt.Errorf("pop"))
 
-	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				Group: groupID,
 				SignerRef: core.SignerRef{
@@ -505,8 +505,8 @@ func TestWriteTransactionSubmitBatchPinFail(t *testing.T) {
 	mmp := pm.multiparty.(*multipartymocks.Manager)
 	mmp.On("SubmitBatchPin", pm.ctx, mock.Anything, mock.Anything, "").Return(fmt.Errorf("pop"))
 
-	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchState{
-		Persisted: core.BatchPersisted{
+	err := pm.dispatchPinnedBatch(pm.ctx, &batch.DispatchPayload{
+		Batch: core.BatchPersisted{
 			BatchHeader: core.BatchHeader{
 				Group: groupID,
 				SignerRef: core.SignerRef{
diff --git a/internal/txcommon/txcommon_test.go b/internal/txcommon/txcommon_test.go
index 6c21d1ff2..c6eaaf7ff 100644
--- a/internal/txcommon/txcommon_test.go
+++ b/internal/txcommon/txcommon_test.go
@@ -633,3 +633,43 @@ func TestInsertBlockchainEventErr(t *testing.T) {
 	mdi.AssertExpectations(t)
 
 }
+
+func TestFindOperationInTransaction(t *testing.T) {
+	mdi := &databasemocks.Plugin{}
+	mdm := &datamocks.Manager{}
+	ctx := context.Background()
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(ctx, 100, 5*time.Minute), nil)
+	txHelper, _ := NewTransactionHelper(ctx, "ns1", mdi, mdm, cmi)
+
+	txID := fftypes.NewUUID()
+	ops := []*core.Operation{{
+		ID: fftypes.NewUUID(),
+	}}
+	mdi.On("GetOperations", ctx, "ns1", mock.Anything).Return(ops, nil, nil)
+
+	result, err := txHelper.FindOperationInTransaction(ctx, txID, core.OpTypeTokenTransfer)
+
+	assert.NoError(t, err)
+	assert.Equal(t, ops[0].ID, result.ID)
+
+	mdi.AssertExpectations(t)
+}
+
+func TestFindOperationInTransactionFail(t *testing.T) {
+	mdi := &databasemocks.Plugin{}
+	mdm := &datamocks.Manager{}
+	ctx := context.Background()
+	cmi := &cachemocks.Manager{}
+	cmi.On("GetCache", mock.Anything).Return(cache.NewUmanagedCache(ctx, 100, 5*time.Minute), nil)
+	txHelper, _ := NewTransactionHelper(ctx, "ns1", mdi, mdm, cmi)
+
+	txID := fftypes.NewUUID()
+	mdi.On("GetOperations", ctx, "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
+
+	_, err := txHelper.FindOperationInTransaction(ctx, txID, core.OpTypeTokenTransfer)
+
+	assert.Regexp(t, "pop", err)
+
+	mdi.AssertExpectations(t)
+}
diff --git a/pkg/core/transaction_test.go b/pkg/core/transaction_test.go
index fcf056894..1dfd3e398 100644
--- a/pkg/core/transaction_test.go
+++ b/pkg/core/transaction_test.go
@@ -25,3 +25,8 @@ import (
 func TestTXSizeEstimate(t *testing.T) {
 	assert.Equal(t, transactionBaseSizeEstimate, (&Transaction{}).Size())
 }
+
+func TestIsPinned(t *testing.T) {
+	assert.True(t, IsPinned(TransactionTypeBatchPin))
+	assert.False(t, IsPinned(TransactionTypeUnpinned))
+}