Repository: hyperledger/firefly (go)

Issue:
Own Identity Claims Stuck in Pending
## Description

When registering orgs and new nodes, we have observed identity claims being stuck in `pending` on the node that registering its org even though registration succeeded and we see message confirmed events in the logs and on other nodes:

<img width="1725" alt="Screen Shot 2022-04-11 at 2 38 15 PM" src="https://user-images.githubusercontent.com/11233384/162824952-fba82bd7-5322-4ba6-8081-9595a46b78f0.png">
<img width="1727" alt="Screen Shot 2022-04-11 at 2 38 29 PM" src="https://user-images.githubusercontent.com/11233384/162824955-39717357-2533-4b49-8362-43b1c09a8a77.png">

This had been hopefully fixed in https://github.com/hyperledger/firefly/pull/670 however it does not appear to be since this was observed with a recent release candidate and we can see similar logs still in the registering node:

```
[2022-04-11T16:53:30.270Z] DEBUG poc4-par1-firefly: SQL-> insert INSERT INTO messages dbtx=wKFeQgiZ role=event-manager
[2022-04-11T16:53:30.272Z] DEBUG poc4-par1-firefly: SQL<- inserted sequences=[3 4] dbtx=aiTOEbd5 role=aggregator
[2022-04-11T16:53:30.272Z] DEBUG poc4-par1-firefly: SQL-> commit dbtx=aiTOEbd5 role=aggregator
[2022-04-11T16:53:30.275Z] DEBUG poc4-par1-firefly: SQL<- commit dbtx=aiTOEbd5 role=aggregator
[2022-04-11T16:53:30.276Z]  INFO poc4-par1-firefly: Emitted identity_confirmed event 7a18c9cb-8e64-44b4-84bb-53fb9085e8e5 for ff_system:8e8b6544-df58-43fe-935e-f0078b760e7d (correlator=,topic=ff_definition) dbtx=aiTOEbd5 role=aggregator
[2022-04-11T16:53:30.276Z]  INFO poc4-par1-firefly: Emitted message_confirmed event fc00e5e9-f71e-4c84-b5bf-91b8db3dea9f for ff_system:31d09e35-1b6e-4df5-bb86-483b22fdcfb4 (correlator=,topic=e5b3a0483b8312f9f68725e69cfadb0494fb9f38290be911d546e2e7abf40884) dbtx=aiTOEbd5 role=aggregator
[2022-04-11T16:53:30.276Z] DEBUG poc4-par1-firefly: SQL insert failed (conflictEmptyRequested=true): FF10375: Failed to retrieve sequence for insert row 1 (could mean duplicate insert) sql=[ INSERT INTO messages (id,cid,mtype,author,key,created,namespace,topics,tag,group_hash,datahash,hash,pins,state,confirmed,tx_type,batch_id) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17)  ON CONFLICT DO NOTHING RETURNING seq ]: FF10375: Failed to retrieve sequence for insert row 1 (could mean duplicate insert) dbtx=wKFeQgiZ role=event-manager
[2022-04-11T16:53:30.276Z] DEBUG poc4-par1-firefly: Batch message insert optimization failed for batch '175adf2a-df06-4c5b-95af-12085bdac7c0': FF10116: Database insert failed: FF10375: Failed to retrieve sequence for insert row 1 (could mean duplicate insert) dbtx=wKFeQgiZ role=event-manager
[2022-04-11T16:53:30.276Z] DEBUG poc4-par1-firefly: SQL-> update: UPDATE messages SET dbtx=wKFeQgiZ role=event-manager
```

## Other Info

FireFly Version: `1.0.0-rc3`
Ethconnect Version: `v3.1.5`
Blockchain Provider: `Quorum v21.10.0 (Geth v1.9.25)`

## Logs

[poc4-par1-registration.log](https://github.com/hyperledger/firefly/files/8467463/poc4-par1-registration.log)



Pull request description:
Ensuring Senders Do Not Download Their Own Batches in Response to Pin Events
Fixes #694, we were seeing the sender's aggregator process pins before a batch was ever persisted. So although message confirmed events were still being sent by the dispatchers, the database updates to mark the message pending (what is done as part of persisting a batch) were occurring after the messages were already confirmed.

When persisting the batch, we guarantee that the batch is not cached until its successfully persisted:
```
	valid, err = em.validateAndPersistBatchContent(ctx, batch)
	if err != nil || !valid {
		return nil, valid, err
	}
	em.aggregator.cacheBatch(em.aggregator.getBatchCacheKey(persistedBatch.ID, persistedBatch.Hash), persistedBatch, manifest)
	return persistedBatch, true, err
```

so if the aggregator repolls any pins who's batch is not in the cache this should avoid the issue. Other option is instead of repolling, letting the aggregators rewind handle processing the pins.

See https://github.com/hyperledger/firefly/pull/745/files#discussion_r854706368 for a more up-to-date explanation of the cause and fix.

Patch:
diff --git a/README.md b/README.md
index 0933d1e83..6ece58005 100644
--- a/README.md
+++ b/README.md
@@ -68,7 +68,8 @@ Other repositories you might be interested in containing those microservice comp
   - Public EVM compatible chains: Learn more in the [documentation](https://hyperledger.github.io/firefly)
 - Corda connector starter: https://github.com/hyperledger/firefly-cordaconnect
   - CorDapp specific customization is required
-- Firefly Performance CLI: https://github.com/hyperledger/firefly-perf-cli
+- FireFly Performance CLI: https://github.com/hyperledger/firefly-perf-cli
+- Helm Charts for Deploying to Kubernetes: https://github.com/hyperledger/firefly-helm-charts
 
 ## FireFly Core code hierarchy
 
diff --git a/internal/database/sqlcommon/message_sql.go b/internal/database/sqlcommon/message_sql.go
index 0badc95c1..59a2cc578 100644
--- a/internal/database/sqlcommon/message_sql.go
+++ b/internal/database/sqlcommon/message_sql.go
@@ -116,7 +116,7 @@ func (s *SQLCommon) attemptMessageInsert(ctx context.Context, tx *txWrapper, mes
 	return err
 }
 
-func (s *SQLCommon) UpsertMessage(ctx context.Context, message *fftypes.Message, optimization database.UpsertOptimization) (err error) {
+func (s *SQLCommon) UpsertMessage(ctx context.Context, message *fftypes.Message, optimization database.UpsertOptimization, hooks ...database.PostCompletionHook) (err error) {
 	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
 	if err != nil {
 		return err
@@ -183,6 +183,10 @@ func (s *SQLCommon) UpsertMessage(ctx context.Context, message *fftypes.Message,
 		}
 	}
 
+	for _, hook := range hooks {
+		s.postCommitEvent(tx, hook)
+	}
+
 	return s.commitTx(ctx, tx, autoCommit)
 }
 
diff --git a/internal/events/batch_pin_complete.go b/internal/events/batch_pin_complete.go
index 3d639c279..5039f52a2 100644
--- a/internal/events/batch_pin_complete.go
+++ b/internal/events/batch_pin_complete.go
@@ -35,6 +35,7 @@ func (em *eventManager) BatchPinComplete(bi blockchain.Plugin, batchPin *blockch
 		log.L(em.ctx).Errorf("Invalid BatchPin transaction - ID is nil")
 		return nil // move on
 	}
+
 	if err := fftypes.ValidateFFNameField(em.ctx, batchPin.Namespace, "namespace"); err != nil {
 		log.L(em.ctx).Errorf("Invalid transaction ID='%s' - invalid namespace '%s': %a", batchPin.TransactionID, batchPin.Namespace, err)
 		return nil // move on
@@ -68,8 +69,16 @@ func (em *eventManager) BatchPinComplete(bi blockchain.Plugin, batchPin *blockch
 			if err := em.persistContexts(ctx, batchPin, signingKey, private); err != nil {
 				return err
 			}
-			// Kick off a download for broadcast batches
-			if !private {
+
+			batch, _, err := em.aggregator.GetBatchForPin(ctx, &fftypes.Pin{
+				Batch:     batchPin.BatchID,
+				BatchHash: batchPin.BatchHash,
+			})
+			if err != nil {
+				return err
+			}
+			// Kick off a download for broadcast batches if the batch isn't already persisted
+			if !private && batch == nil {
 				if err := em.sharedDownload.InitiateDownloadBatch(ctx, batchPin.Namespace, batchPin.TransactionID, batchPin.BatchPayloadRef); err != nil {
 					return err
 				}
diff --git a/internal/events/persist_batch.go b/internal/events/persist_batch.go
index dece7591a..c93a796ed 100644
--- a/internal/events/persist_batch.go
+++ b/internal/events/persist_batch.go
@@ -246,6 +246,7 @@ func (em *eventManager) persistBatchContent(ctx context.Context, batch *fftypes.
 	// If we are sure we wrote the batch, then we do a cached lookup of each in turn - which is efficient
 	// because all of those should be in the cache as we wrote them recently.
 	if em.sentByUs(ctx, batch) {
+		log.L(ctx).Debugf("Batch %s sent by us", batch.ID)
 		allStored, err := em.verifyAlreadyStored(ctx, batch)
 		if err != nil {
 			return false, err
@@ -290,7 +291,11 @@ func (em *eventManager) persistBatchContent(ctx context.Context, batch *fftypes.
 		log.L(ctx).Debugf("Batch message insert optimization failed for batch '%s': %s", batch.ID, err)
 		// Fall back to individual upserts
 		for i, msg := range batch.Payload.Messages {
-			if err = em.database.UpsertMessage(ctx, msg, database.UpsertOptimizationExisting); err != nil {
+			postHookUpdateMessageCache := func() {
+				mm := matchedMsgs[i]
+				em.data.UpdateMessageCache(mm.message, mm.data)
+			}
+			if err = em.database.UpsertMessage(ctx, msg, database.UpsertOptimizationExisting, postHookUpdateMessageCache); err != nil {
 				if err == database.HashMismatch {
 					log.L(ctx).Errorf("Invalid message entry %d in batch'%s'. Hash mismatch with existing record with same UUID '%s' Hash=%s", i, batch.ID, msg.Header.ID, msg.Hash)
 					return false, nil // This is not retryable. skip this data entry
diff --git a/mocks/databasemocks/plugin.go b/mocks/databasemocks/plugin.go
index d27582785..0add623c6 100644
--- a/mocks/databasemocks/plugin.go
+++ b/mocks/databasemocks/plugin.go
@@ -2939,8 +2939,15 @@ func (_m *Plugin) UpsertIdentity(ctx context.Context, data *fftypes.Identity, op
 }
 
 // UpsertMessage provides a mock function with given fields: ctx, message, optimization
-func (_m *Plugin) UpsertMessage(ctx context.Context, message *fftypes.Message, optimization database.UpsertOptimization) error {
-	ret := _m.Called(ctx, message, optimization)
+func (_m *Plugin) UpsertMessage(ctx context.Context, message *fftypes.Message, optimization database.UpsertOptimization, hooks ...database.PostCompletionHook) error {
+	_va := make([]interface{}, len(hooks))
+	for _i := range hooks {
+		_va[_i] = hooks[_i]
+	}
+	var _ca []interface{}
+	_ca = append(_ca, ctx, message, optimization)
+	_ca = append(_ca, _va...)
+	ret := _m.Called(_ca...)
 
 	var r0 error
 	if rf, ok := ret.Get(0).(func(context.Context, *fftypes.Message, database.UpsertOptimization) error); ok {
diff --git a/pkg/database/plugin.go b/pkg/database/plugin.go
index 1852bd731..d3bb3c135 100644
--- a/pkg/database/plugin.go
+++ b/pkg/database/plugin.go
@@ -79,7 +79,7 @@ type iMessageCollection interface {
 	// UpsertMessage - Upsert a message, with all the embedded data references.
 	//                 The database layer must ensure that if a record already exists, the hash of that existing record
 	//                 must match the hash of the record that is being inserted.
-	UpsertMessage(ctx context.Context, message *fftypes.Message, optimization UpsertOptimization) (err error)
+	UpsertMessage(ctx context.Context, message *fftypes.Message, optimization UpsertOptimization, hooks ...PostCompletionHook) (err error)
 
 	// InsertMessages performs a batch insert of messages assured to be new records - fails if they already exist, so caller can fall back to upsert individually
 	InsertMessages(ctx context.Context, messages []*fftypes.Message, hooks ...PostCompletionHook) (err error)


Test patch:
diff --git a/internal/database/sqlcommon/message_sql_test.go b/internal/database/sqlcommon/message_sql_test.go
index 5886aee64..e88a01c10 100644
--- a/internal/database/sqlcommon/message_sql_test.go
+++ b/internal/database/sqlcommon/message_sql_test.go
@@ -127,7 +127,11 @@ func TestUpsertE2EWithDB(t *testing.T) {
 	assert.Equal(t, database.HashMismatch, err)
 
 	msgUpdated.Hash = msg.Hash
-	err = s.UpsertMessage(context.Background(), msgUpdated, database.UpsertOptimizationExisting)
+	hookCalled := make(chan struct{}, 1)
+	err = s.UpsertMessage(context.Background(), msgUpdated, database.UpsertOptimizationExisting, func() {
+		close(hookCalled)
+	})
+	<-hookCalled
 	assert.NoError(t, err)
 
 	// Check we get the exact same message back - note the removal of one of the data elements
diff --git a/internal/events/batch_pin_complete_test.go b/internal/events/batch_pin_complete_test.go
index 246478d36..7607c2317 100644
--- a/internal/events/batch_pin_complete_test.go
+++ b/internal/events/batch_pin_complete_test.go
@@ -18,6 +18,7 @@ package events
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"testing"
 
@@ -133,6 +134,7 @@ func TestBatchPinCompleteOkBroadcast(t *testing.T) {
 	})).Return(nil).Times(1)
 	mdi.On("InsertPins", mock.Anything, mock.Anything).Return(nil).Once()
 	msd := em.sharedDownload.(*shareddownloadmocks.Manager)
+	mdi.On("GetBatchByID", mock.Anything, mock.Anything).Return(nil, nil)
 	msd.On("InitiateDownloadBatch", mock.Anything, "ns1", batchPin.TransactionID, batchPin.BatchPayloadRef).Return(nil)
 	mbi := &blockchainmocks.Plugin{}
 
@@ -146,6 +148,76 @@ func TestBatchPinCompleteOkBroadcast(t *testing.T) {
 	mth.AssertExpectations(t)
 }
 
+func TestBatchPinCompleteOkBroadcastExistingBatch(t *testing.T) {
+	em, cancel := newTestEventManager(t)
+	defer cancel()
+
+	data := &fftypes.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
+	batch := sampleBatch(t, fftypes.BatchTypeBroadcast, fftypes.TransactionTypeBatchPin, fftypes.DataArray{data})
+	batchPin := &blockchain.BatchPin{
+		Namespace:       "ns1",
+		TransactionID:   batch.Payload.TX.ID,
+		BatchID:         batch.ID,
+		BatchPayloadRef: "Qmf412jQZiuVUtdgnB36FXFX7xg5V6KEbSJ4dpQuhkLyfD",
+		Contexts:        []*fftypes.Bytes32{fftypes.NewRandB32()},
+		Event: blockchain.Event{
+			Name:           "BatchPin",
+			BlockchainTXID: "0x12345",
+			ProtocolID:     "10/20/30",
+		},
+	}
+	payloadBinary, jsonErr := json.Marshal(&batch.Payload)
+	assert.NoError(t, jsonErr)
+	batchPersisted := &fftypes.BatchPersisted{
+		TX:          batch.Payload.TX,
+		BatchHeader: batch.BatchHeader,
+		Manifest:    fftypes.JSONAnyPtr(string(payloadBinary)),
+		Hash:        batch.Payload.Hash(),
+	}
+
+	batch.Hash = batch.Payload.Hash()
+	batchPin.BatchHash = batch.Hash
+
+	mth := em.txHelper.(*txcommonmocks.Helper)
+	mth.On("PersistTransaction", mock.Anything, "ns1", batchPin.TransactionID, fftypes.TransactionTypeBatchPin, "0x12345").
+		Return(false, fmt.Errorf("pop")).Once()
+	mth.On("PersistTransaction", mock.Anything, "ns1", batchPin.TransactionID, fftypes.TransactionTypeBatchPin, "0x12345").
+		Return(true, nil)
+
+	mdi := em.database.(*databasemocks.Plugin)
+	rag := mdi.On("RunAsGroup", mock.Anything, mock.Anything).Return(nil)
+	rag.RunFn = func(a mock.Arguments) {
+		// Call through to persistBatch - the hash of our batch will be invalid,
+		// which is swallowed without error as we cannot retry (it is logged of course)
+		rag.ReturnArguments = mock.Arguments{
+			a[1].(func(ctx context.Context) error)(a[0].(context.Context)),
+		}
+	}
+
+	mdi.On("GetBlockchainEventByProtocolID", mock.Anything, "ns1", (*fftypes.UUID)(nil), batchPin.Event.ProtocolID).Return(nil, nil)
+	mth.On("InsertBlockchainEvent", mock.Anything, mock.MatchedBy(func(e *fftypes.BlockchainEvent) bool {
+		return e.Name == batchPin.Event.Name
+	})).Return(fmt.Errorf("pop")).Once()
+	mth.On("InsertBlockchainEvent", mock.Anything, mock.MatchedBy(func(e *fftypes.BlockchainEvent) bool {
+		return e.Name == batchPin.Event.Name
+	})).Return(nil).Times(1)
+	mdi.On("InsertEvent", mock.Anything, mock.MatchedBy(func(e *fftypes.Event) bool {
+		return e.Type == fftypes.EventTypeBlockchainEventReceived
+	})).Return(nil).Times(1)
+	mdi.On("InsertPins", mock.Anything, mock.Anything).Return(nil).Once()
+	mdi.On("GetBatchByID", mock.Anything, mock.Anything).Return(batchPersisted, nil)
+	mbi := &blockchainmocks.Plugin{}
+
+	err := em.BatchPinComplete(mbi, batchPin, &fftypes.VerifierRef{
+		Type:  fftypes.VerifierTypeEthAddress,
+		Value: "0x12345",
+	})
+	assert.NoError(t, err)
+
+	mdi.AssertExpectations(t)
+	mth.AssertExpectations(t)
+}
+
 func TestBatchPinCompleteOkPrivate(t *testing.T) {
 	em, cancel := newTestEventManager(t)
 	defer cancel()
@@ -171,6 +243,7 @@ func TestBatchPinCompleteOkPrivate(t *testing.T) {
 	mdi.On("GetBlockchainEventByProtocolID", mock.Anything, "ns1", (*fftypes.UUID)(nil), batchPin.Event.ProtocolID).Return(nil, nil)
 	mth.On("InsertBlockchainEvent", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil)
+	mdi.On("GetBatchByID", mock.Anything, mock.Anything).Return(nil, nil)
 
 	mbi := &blockchainmocks.Plugin{}
 
@@ -227,6 +300,45 @@ func TestBatchPinCompleteInsertPinsFail(t *testing.T) {
 	mdi.AssertExpectations(t)
 	mth.AssertExpectations(t)
 }
+
+func TestBatchPinCompleteGetBatchByIDFails(t *testing.T) {
+	em, cancel := newTestEventManager(t)
+	cancel()
+
+	batchPin := &blockchain.BatchPin{
+		Namespace:     "ns1",
+		TransactionID: fftypes.NewUUID(),
+		BatchID:       fftypes.NewUUID(),
+		Contexts:      []*fftypes.Bytes32{fftypes.NewRandB32()},
+		Event: blockchain.Event{
+			BlockchainTXID: "0x12345",
+			ProtocolID:     "10/20/30",
+		},
+	}
+
+	mth := em.txHelper.(*txcommonmocks.Helper)
+	mth.On("PersistTransaction", mock.Anything, "ns1", batchPin.TransactionID, fftypes.TransactionTypeBatchPin, "0x12345").Return(true, nil)
+
+	mdi := em.database.(*databasemocks.Plugin)
+	mdi.On("RunAsGroup", mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertPins", mock.Anything, mock.Anything).Return(nil)
+	mdi.On("GetBlockchainEventByProtocolID", mock.Anything, "ns1", (*fftypes.UUID)(nil), batchPin.Event.ProtocolID).Return(nil, nil)
+	mth.On("InsertBlockchainEvent", mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil)
+	mdi.On("GetBatchByID", mock.Anything, mock.Anything).Return(nil, fmt.Errorf("batch lookup failed"))
+
+	mbi := &blockchainmocks.Plugin{}
+
+	err := em.BatchPinComplete(mbi, batchPin, &fftypes.VerifierRef{
+		Type:  fftypes.VerifierTypeEthAddress,
+		Value: "0xffffeeee",
+	})
+	assert.Regexp(t, "FF10158", err)
+
+	mdi.AssertExpectations(t)
+	mth.AssertExpectations(t)
+}
+
 func TestSequencedBroadcastInitiateDownloadFail(t *testing.T) {
 	em, cancel := newTestEventManager(t)
 
@@ -253,6 +365,7 @@ func TestSequencedBroadcastInitiateDownloadFail(t *testing.T) {
 	mth.On("InsertBlockchainEvent", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertEvent", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertPins", mock.Anything, mock.Anything).Return(nil)
+	mdi.On("GetBatchByID", mock.Anything, mock.Anything).Return(nil, nil)
 	msd := em.sharedDownload.(*shareddownloadmocks.Manager)
 	msd.On("InitiateDownloadBatch", mock.Anything, "ns1", batchPin.TransactionID, batchPin.BatchPayloadRef).Return(fmt.Errorf("pop"))
 
@@ -522,7 +635,7 @@ func TestPersistBatchGoodDataMessageFail(t *testing.T) {
 	mdi.On("UpsertBatch", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertDataArray", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertMessages", mock.Anything, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("optimzation miss"))
-	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationExisting).Return(fmt.Errorf("pop"))
+	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationExisting, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("pop"))
 
 	bp, valid, err := em.persistBatch(context.Background(), batch)
 	assert.False(t, valid)
diff --git a/internal/events/dx_callbacks_test.go b/internal/events/dx_callbacks_test.go
index 5c77fbe93..1bbf114fe 100644
--- a/internal/events/dx_callbacks_test.go
+++ b/internal/events/dx_callbacks_test.go
@@ -611,7 +611,7 @@ func TestMessageReceiveMessagePersistMessageFail(t *testing.T) {
 	mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil, nil)
 	mdi.On("InsertDataArray", em.ctx, mock.Anything).Return(nil)
 	mdi.On("InsertMessages", em.ctx, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("optimization fail"))
-	mdi.On("UpsertMessage", em.ctx, mock.Anything, database.UpsertOptimizationExisting).Return(fmt.Errorf("pop"))
+	mdi.On("UpsertMessage", em.ctx, mock.Anything, database.UpsertOptimizationExisting, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("pop"))
 
 	// no ack as we are simulating termination mid retry
 	mde := newMessageReceivedNoAck("peer1", b)
diff --git a/internal/events/persist_batch_test.go b/internal/events/persist_batch_test.go
index 7c499153e..e7ac3b162 100644
--- a/internal/events/persist_batch_test.go
+++ b/internal/events/persist_batch_test.go
@@ -97,7 +97,7 @@ func TestPersistBatchNoCacheDataNotInBatch(t *testing.T) {
 
 	mdi := em.database.(*databasemocks.Plugin)
 	mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil)
-	mdi.On("UpsertMessage", em.ctx, mock.Anything, database.UpsertOptimizationSkip).Return(nil)
+	mdi.On("UpsertMessage", em.ctx, mock.Anything, database.UpsertOptimizationSkip, mock.AnythingOfType("database.PostCompletionHook")).Return(nil)
 
 	data := &fftypes.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, fftypes.BatchTypeBroadcast, fftypes.TransactionTypeBatchPin, fftypes.DataArray{data})
@@ -119,7 +119,7 @@ func TestPersistBatchExtraDataInBatch(t *testing.T) {
 
 	mdi := em.database.(*databasemocks.Plugin)
 	mdi.On("UpsertBatch", em.ctx, mock.Anything).Return(nil)
-	mdi.On("UpsertMessage", em.ctx, mock.Anything, database.UpsertOptimizationSkip).Return(nil)
+	mdi.On("UpsertMessage", em.ctx, mock.Anything, database.UpsertOptimizationSkip, mock.AnythingOfType("database.PostCompletionHook")).Return(nil)
 
 	data := &fftypes.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
 	batch := sampleBatch(t, fftypes.BatchTypeBroadcast, fftypes.TransactionTypeBatchPin, fftypes.DataArray{data})
@@ -233,7 +233,7 @@ func TestPersistBatchContentSentByUsFoundMismatch(t *testing.T) {
 	mdi := em.database.(*databasemocks.Plugin)
 	mdi.On("InsertDataArray", mock.Anything, mock.Anything).Return(nil)
 	mdi.On("InsertMessages", mock.Anything, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("optimization miss"))
-	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationExisting).Return(database.HashMismatch)
+	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationExisting, mock.AnythingOfType("database.PostCompletionHook")).Return(database.HashMismatch)
 
 	ok, err := em.persistBatchContent(em.ctx, batch, []*messageAndData{})
 	assert.NoError(t, err)
@@ -244,6 +244,36 @@ func TestPersistBatchContentSentByUsFoundMismatch(t *testing.T) {
 
 }
 
+func TestPersistBatchContentInsertMessagesFail(t *testing.T) {
+
+	em, cancel := newTestEventManager(t)
+	defer cancel()
+
+	data := &fftypes.Data{ID: fftypes.NewUUID(), Value: fftypes.JSONAnyPtr(`"test"`)}
+	batch := sampleBatch(t, fftypes.BatchTypeBroadcast, fftypes.TransactionTypeBatchPin, fftypes.DataArray{data})
+
+	mdi := em.database.(*databasemocks.Plugin)
+	mdi.On("InsertDataArray", mock.Anything, mock.Anything).Return(nil)
+	mdi.On("InsertMessages", mock.Anything, mock.Anything, mock.AnythingOfType("database.PostCompletionHook")).Return(fmt.Errorf("optimization miss"))
+	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationExisting, mock.AnythingOfType("database.PostCompletionHook")).Return(nil).Run(func(args mock.Arguments) {
+		args[3].(database.PostCompletionHook)()
+	})
+
+	mdm := em.data.(*datamocks.Manager)
+	msgData := &messageAndData{
+		message: batch.Payload.Messages[0],
+		data:    batch.Payload.Data,
+	}
+	mdm.On("UpdateMessageCache", msgData.message, msgData.data).Return()
+
+	ok, err := em.persistBatchContent(em.ctx, batch, []*messageAndData{msgData})
+	assert.NoError(t, err)
+	assert.True(t, ok)
+
+	mdm.AssertExpectations(t)
+	mdi.AssertExpectations(t)
+}
+
 func TestPersistBatchContentSentByUsFoundError(t *testing.T) {
 
 	em, cancel := newTestEventManager(t)