Merge pull request #1386 from kaleido-io/blockchain-existing

Prevent inserting token transfers with invalid blockchain event refs

internal/events/blockchain_event.go

@@ -72,11 +72,14 @@ func (em *eventManager) getChainListenerByProtocolIDCached(ctx context.Context,
 }
 
 func (em *eventManager) maybePersistBlockchainEvent(ctx context.Context, chainEvent *core.BlockchainEvent, listener *core.ContractListener) error {
-	inserted, err := em.txHelper.InsertNewBlockchainEvents(ctx, []*core.BlockchainEvent{chainEvent})
+	existing, err := em.txHelper.InsertOrGetBlockchainEvent(ctx, chainEvent)
 	if err != nil {
 		return err
 	}
-	if len(inserted) < 1 {
+	if existing != nil {
+		log.L(ctx).Debugf("Ignoring duplicate blockchain event %s", chainEvent.ProtocolID)
+		// Return the ID of the existing event
+		chainEvent.ID = existing.ID
 		return nil
 	}
 	topic := em.getTopicForChainListener(listener)

internal/events/blockchain_event_test.go

@@ -164,12 +164,14 @@ func TestPersistBlockchainEventDuplicate(t *testing.T) {
 		},
 		Listener: fftypes.NewUUID(),
 	}
+	existingID := fftypes.NewUUID()
 
-	em.mth.On("InsertNewBlockchainEvents", mock.Anything, []*core.BlockchainEvent{ev}).
-		Return([]*core.BlockchainEvent{ /* duplicate, so empty insert array */ }, nil)
+	em.mth.On("InsertOrGetBlockchainEvent", mock.Anything, ev).
+		Return(&core.BlockchainEvent{ID: existingID}, nil)
 
 	err := em.maybePersistBlockchainEvent(em.ctx, ev, nil)
 	assert.NoError(t, err)
+	assert.Equal(t, existingID, ev.ID)
 
 }
 

internal/events/token_pool_created_test.go

@@ -122,9 +122,9 @@ func TestTokenPoolCreatedConfirm(t *testing.T) {
 
 	em.mam.On("GetTokenPoolByLocator", em.ctx, "erc1155", "123").Return(nil, fmt.Errorf("pop")).Once()
 	em.mam.On("GetTokenPoolByLocator", em.ctx, "erc1155", "123").Return(storedPool, nil).Once()
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Name == chainPool.Event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Name == chainPool.Event.Name
+	})).Return(nil, nil)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *core.Event) bool {
 		return e.Type == core.EventTypeBlockchainEventReceived
 	})).Return(nil).Once()
@@ -246,7 +246,9 @@ func TestConfirmPoolBlockchainEventFail(t *testing.T) {
 		ProtocolID:     "tx1",
 	}
 
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.Anything).Return(nil, fmt.Errorf("pop"))
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Name == event.Name
+	})).Return(nil, fmt.Errorf("pop"))
 
 	err := em.confirmPool(em.ctx, storedPool, event)
 	assert.EqualError(t, err, "pop")
@@ -274,9 +276,9 @@ func TestConfirmPoolTxFail(t *testing.T) {
 		ProtocolID:     "tx1",
 	}
 
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Name == event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Name == event.Name
+	})).Return(nil, nil)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *core.Event) bool {
 		return e.Type == core.EventTypeBlockchainEventReceived
 	})).Return(nil)
@@ -308,9 +310,9 @@ func TestConfirmPoolUpsertFail(t *testing.T) {
 		ProtocolID:     "tx1",
 	}
 
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Name == event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Name == event.Name
+	})).Return(nil, nil)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *core.Event) bool {
 		return e.Type == core.EventTypeBlockchainEventReceived
 	})).Return(nil)

internal/events/tokens_approved_test.go

@@ -73,9 +73,9 @@ func TestTokensApprovedSucceedWithRetries(t *testing.T) {
 	em.mam.On("GetTokenPoolByID", em.ctx, pool.ID).Return(pool, nil).Times(3)
 	em.mdi.On("GetTokenApprovalByProtocolID", em.ctx, "ns1", pool.ID, approval.ProtocolID).Return(nil, fmt.Errorf("pop")).Once()
 	em.mdi.On("GetTokenApprovalByProtocolID", em.ctx, "ns1", pool.ID, approval.ProtocolID).Return(nil, nil).Times(3)
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Name == approval.Event.Name && events[0].Namespace == approval.Namespace
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == approval.Event.Name
+	})).Return(nil, nil).Times(3)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *core.Event) bool {
 		return ev.Type == core.EventTypeBlockchainEventReceived && ev.Namespace == pool.Namespace
 	})).Return(nil).Times(3)
@@ -258,9 +258,9 @@ func TestApprovedWithTransactionRegenerateLocalID(t *testing.T) {
 	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("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Namespace == pool.Namespace && events[0].Name == approval.Event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == approval.Event.Name
+	})).Return(nil, nil)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *core.Event) bool {
 		return ev.Type == core.EventTypeBlockchainEventReceived && ev.Namespace == pool.Namespace
 	})).Return(nil)
@@ -299,8 +299,8 @@ func TestApprovedBlockchainEventFail(t *testing.T) {
 	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("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Namespace == pool.Namespace && events[0].Name == approval.Event.Name
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == approval.Event.Name
 	})).Return(nil, fmt.Errorf("pop"))
 
 	valid, err := em.persistTokenApproval(em.ctx, approval)
@@ -341,9 +341,9 @@ func TestTokensApprovedWithMessageReceived(t *testing.T) {
 	em.mdi.On("GetTokenApprovalByProtocolID", em.ctx, "ns1", pool.ID, "123").Return(nil, nil).Times(2)
 	em.mam.On("GetTokenPoolByLocator", em.ctx, "erc1155", "F1").Return(pool, nil).Once()
 	em.mam.On("GetTokenPoolByID", em.ctx, pool.ID).Return(pool, nil).Once()
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Namespace == pool.Namespace && events[0].Name == approval.Event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil).Times(2)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == approval.Event.Name
+	})).Return(nil, nil).Times(2)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *core.Event) bool {
 		return ev.Type == core.EventTypeBlockchainEventReceived && ev.Namespace == pool.Namespace
 	})).Return(nil).Times(2)
@@ -394,9 +394,9 @@ func TestTokensApprovedWithMessageSend(t *testing.T) {
 	em.mdi.On("GetTokenApprovalByProtocolID", em.ctx, "ns1", pool.ID, "123").Return(nil, nil).Times(2)
 	em.mam.On("GetTokenPoolByLocator", em.ctx, "erc1155", "F1").Return(pool, nil).Once()
 	em.mam.On("GetTokenPoolByID", em.ctx, pool.ID).Return(pool, nil).Once()
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Namespace == pool.Namespace && events[0].Name == approval.Event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil).Times(2)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == approval.Event.Name
+	})).Return(nil, nil).Times(2)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *core.Event) bool {
 		return ev.Type == core.EventTypeBlockchainEventReceived && ev.Namespace == pool.Namespace
 	})).Return(nil).Times(2)

internal/events/tokens_transferred_test.go

@@ -72,9 +72,9 @@ func TestTokensTransferredSucceedWithRetries(t *testing.T) {
 	em.mam.On("GetTokenPoolByLocator", em.ctx, "erc1155", "F1").Return(nil, fmt.Errorf("pop")).Once()
 	em.mam.On("GetTokenPoolByLocator", em.ctx, "erc1155", "F1").Return(pool, nil).Once()
 	em.mam.On("GetTokenPoolByID", em.ctx, pool.ID).Return(pool, nil).Times(2)
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Namespace == pool.Namespace && events[0].Name == transfer.Event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == transfer.Event.Name
+	})).Return(nil, nil).Times(3)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *core.Event) bool {
 		return ev.Type == core.EventTypeBlockchainEventReceived && ev.Namespace == pool.Namespace
 	})).Return(nil).Times(3)
@@ -212,8 +212,8 @@ func TestPersistTransferBlockchainEventFail(t *testing.T) {
 	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("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Namespace == pool.Namespace && events[0].Name == transfer.Event.Name
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == transfer.Event.Name
 	})).Return(nil, fmt.Errorf("pop"))
 
 	valid, err := em.persistTokenTransfer(em.ctx, transfer)
@@ -246,9 +246,9 @@ func TestTokensTransferredWithTransactionRegenerateLocalID(t *testing.T) {
 	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("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Namespace == pool.Namespace && events[0].Name == transfer.Event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == transfer.Event.Name
+	})).Return(nil, nil)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *core.Event) bool {
 		return ev.Type == core.EventTypeBlockchainEventReceived && ev.Namespace == pool.Namespace
 	})).Return(nil)
@@ -285,9 +285,9 @@ func TestTokensTransferredWithExistingTransfer(t *testing.T) {
 	}
 
 	em.mam.On("GetTokenPoolByLocator", em.ctx, "erc1155", "F1").Return(pool, nil)
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Namespace == pool.Namespace && events[0].Name == transfer.Event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == transfer.Event.Name
+	})).Return(nil, nil)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *core.Event) bool {
 		return ev.Type == core.EventTypeBlockchainEventReceived && ev.Namespace == pool.Namespace
 	})).Return(nil)
@@ -359,9 +359,9 @@ func TestTokensTransferredWithMessageReceived(t *testing.T) {
 
 	em.mam.On("GetTokenPoolByLocator", em.ctx, "erc1155", "F1").Return(pool, nil).Once()
 	em.mam.On("GetTokenPoolByID", em.ctx, pool.ID).Return(pool, nil).Once()
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Namespace == pool.Namespace && events[0].Name == transfer.Event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil).Times(2)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == transfer.Event.Name
+	})).Return(nil, nil).Times(2)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *core.Event) bool {
 		return ev.Type == core.EventTypeBlockchainEventReceived && ev.Namespace == pool.Namespace
 	})).Return(nil).Times(2)
@@ -418,9 +418,9 @@ func TestTokensTransferredWithMessageSend(t *testing.T) {
 
 	em.mam.On("GetTokenPoolByLocator", em.ctx, "erc1155", "F1").Return(pool, nil).Once()
 	em.mam.On("GetTokenPoolByID", em.ctx, pool.ID).Return(pool, nil).Once()
-	em.mth.On("InsertNewBlockchainEvents", em.ctx, mock.MatchedBy(func(events []*core.BlockchainEvent) bool {
-		return len(events) == 1 && events[0].Namespace == pool.Namespace && events[0].Name == transfer.Event.Name
-	})).Return([]*core.BlockchainEvent{{ID: fftypes.NewUUID()}}, nil).Times(2)
+	em.mth.On("InsertOrGetBlockchainEvent", em.ctx, mock.MatchedBy(func(e *core.BlockchainEvent) bool {
+		return e.Namespace == pool.Namespace && e.Name == transfer.Event.Name
+	})).Return(nil, nil).Times(2)
 	em.mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(ev *core.Event) bool {
 		return ev.Type == core.EventTypeBlockchainEventReceived && ev.Namespace == pool.Namespace
 	})).Return(nil).Times(2)

internal/txcommon/txcommon.go

@@ -38,6 +38,7 @@ type Helper interface {
 	SubmitNewTransactionBatch(ctx context.Context, namespace string, batch []*BatchedTransactionInsert) error
 	PersistTransaction(ctx context.Context, id *fftypes.UUID, txType core.TransactionType, blockchainTXID string) (valid bool, err error)
 	AddBlockchainTX(ctx context.Context, tx *core.Transaction, blockchainTXID string) error
+	InsertOrGetBlockchainEvent(ctx context.Context, event *core.BlockchainEvent) (existing *core.BlockchainEvent, err error)
 	InsertNewBlockchainEvents(ctx context.Context, events []*core.BlockchainEvent) (inserted []*core.BlockchainEvent, err error)
 	GetTransactionByIDCached(ctx context.Context, id *fftypes.UUID) (*core.Transaction, error)
 	GetBlockchainEventByIDCached(ctx context.Context, id *fftypes.UUID) (*core.BlockchainEvent, error)
@@ -312,6 +313,19 @@ func (t *transactionHelper) GetBlockchainEventByIDCached(ctx context.Context, id
 	return chainEvent, nil
 }
 
+func (t *transactionHelper) InsertOrGetBlockchainEvent(ctx context.Context, event *core.BlockchainEvent) (existing *core.BlockchainEvent, err error) {
+	existing, err = t.database.InsertOrGetBlockchainEvent(ctx, event)
+	if err != nil {
+		return nil, err
+	}
+	if existing != nil {
+		t.addBlockchainEventToCache(existing)
+		return existing, nil
+	}
+	t.addBlockchainEventToCache(event)
+	return nil, nil
+}
+
 func (t *transactionHelper) InsertNewBlockchainEvents(ctx context.Context, events []*core.BlockchainEvent) (inserted []*core.BlockchainEvent, err error) {
 	// First we try and insert the whole bundle using batch insert
 	err = t.database.InsertBlockchainEvents(ctx, events, func() {

internal/txcommon/txcommon_test.go

@@ -570,6 +570,77 @@ func TestGetBlockchainEventByIDErr(t *testing.T) {
 
 }
 
+func TestInsertGetBlockchainEventCached(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)
+	evID := fftypes.NewUUID()
+	chainEvent := &core.BlockchainEvent{
+		ID:        evID,
+		Namespace: "ns1",
+	}
+	mdi.On("InsertOrGetBlockchainEvent", ctx, chainEvent).Return(nil, nil)
+
+	_, err := txHelper.InsertOrGetBlockchainEvent(ctx, chainEvent)
+	assert.NoError(t, err)
+
+	cached, err := txHelper.GetBlockchainEventByIDCached(ctx, evID)
+	assert.NoError(t, err)
+	assert.Equal(t, chainEvent, cached)
+	mdi.AssertExpectations(t)
+
+}
+
+func TestInsertGetBlockchainEventDuplicate(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)
+	evID := fftypes.NewUUID()
+	chainEvent := &core.BlockchainEvent{
+		ID:        evID,
+		Namespace: "ns1",
+	}
+	existingEvent := &core.BlockchainEvent{}
+	mdi.On("InsertOrGetBlockchainEvent", ctx, chainEvent).Return(existingEvent, nil)
+
+	result, err := txHelper.InsertOrGetBlockchainEvent(ctx, chainEvent)
+	assert.NoError(t, err)
+	assert.Equal(t, existingEvent, result)
+
+	mdi.AssertExpectations(t)
+
+}
+
+func TestInsertGetBlockchainEventErr(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)
+	evID := fftypes.NewUUID()
+	chainEvent := &core.BlockchainEvent{
+		ID:        evID,
+		Namespace: "ns1",
+	}
+	mdi.On("InsertOrGetBlockchainEvent", ctx, chainEvent).Return(nil, fmt.Errorf("pop"))
+
+	_, err := txHelper.InsertOrGetBlockchainEvent(ctx, chainEvent)
+	assert.Regexp(t, "pop", err)
+
+	mdi.AssertExpectations(t)
+
+}
+
 func TestInsertNewBlockchainEventsOptimized(t *testing.T) {
 
 	mdi := &databasemocks.Plugin{}