Repository: hyperledger/firefly
Pull request #970 (2022-08-16 15:51:41)

Update to firefly-signer v0.9.15
Signed-off-by: Nicko Guyer <nicko.guyer@kaleido.io>


Patch:
diff --git a/go.mod b/go.mod
index ed76c3754..c002a2ece 100644
--- a/go.mod
+++ b/go.mod
@@ -14,7 +14,7 @@ require (
 	github.com/gorilla/mux v1.8.0
 	github.com/gorilla/websocket v1.5.0
 	github.com/hyperledger/firefly-common v1.1.0
-	github.com/hyperledger/firefly-signer v0.9.13
+	github.com/hyperledger/firefly-signer v0.9.15
 	github.com/jarcoal/httpmock v1.1.0
 	github.com/karlseguin/ccache v2.0.3+incompatible
 	github.com/lib/pq v1.10.6
diff --git a/go.sum b/go.sum
index 4b7fef337..93c685fde 100644
--- a/go.sum
+++ b/go.sum
@@ -729,8 +729,8 @@ github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpO
 github.com/hyperledger/firefly-common v0.1.13/go.mod h1:2NqPi5Ud9H6rSlZXkLbotxW7z4EAD89p3/8oNOpm9Gs=
 github.com/hyperledger/firefly-common v1.1.0 h1:wbfUPDJfCVU7AssSROBPZLhp0udkjiTVhFGw6BxZzR4=
 github.com/hyperledger/firefly-common v1.1.0/go.mod h1:gMlv4Iy5JjnzXmSEdb+tWVDIc/2GhL9MRcgNX+VmI4M=
-github.com/hyperledger/firefly-signer v0.9.13 h1:yvKxYTsEmE0XWl0vcZBQV353YmmePvWwIPMr4Lie67o=
-github.com/hyperledger/firefly-signer v0.9.13/go.mod h1:GPQRUZOFOAjkLmg8GDjZUjEdUD0gcar+CSVhwltIwyw=
+github.com/hyperledger/firefly-signer v0.9.15 h1:r1tJeS7R7ESRo3Khy7/gSqQ49j2VYBKTrWtEViXuBZU=
+github.com/hyperledger/firefly-signer v0.9.15/go.mod h1:GPQRUZOFOAjkLmg8GDjZUjEdUD0gcar+CSVhwltIwyw=
 github.com/iancoleman/strcase v0.2.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47ZCWhYzw7ho=
 github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
 github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
diff --git a/internal/events/aggregator.go b/internal/events/aggregator.go
index d2e2def37..98e294749 100644
--- a/internal/events/aggregator.go
+++ b/internal/events/aggregator.go
@@ -414,11 +414,6 @@ func (ag *aggregator) checkOnchainConsistency(ctx context.Context, msg *core.Mes
 func (ag *aggregator) processMessage(ctx context.Context, manifest *core.BatchManifest, pin *core.Pin, msgBaseIndex int64, msgEntry *core.MessageManifestEntry, state *batchState) (err error) {
 	l := log.L(ctx)
 
-	unmaskedContexts := make([]*fftypes.Bytes32, 0)
-	nextPins := make([]*nextPinState, 0)
-	var newState core.MessageState
-	var dispatched bool
-
 	var cro data.CacheReadOption
 	if pin.Masked {
 		cro = data.CRORequirePins
@@ -431,50 +426,59 @@ func (ag *aggregator) processMessage(ctx context.Context, manifest *core.BatchMa
 		return err
 	case msg == nil:
 		l.Debugf("Message '%s' in batch '%s' is not yet available", msgEntry.ID, manifest.ID)
+		return nil
 	case !dataAvailable:
 		l.Errorf("Message '%s' in batch '%s' is missing data", msgEntry.ID, manifest.ID)
-	default:
-		// Check if it's ready to be processed
-		if pin.Masked {
-			// Private messages have one or more masked "pin" hashes that allow us to work
-			// out if it's the next message in the sequence, given the previous messages
-			if msg.Header.Group == nil || len(msg.Pins) == 0 || len(msg.Header.Topics) != len(msg.Pins) {
-				l.Errorf("Message '%s' in batch '%s' has invalid pin data pins=%v topics=%v", msg.Header.ID, manifest.ID, msg.Pins, msg.Header.Topics)
+		return nil
+	}
+
+	// Check if it's ready to be processed
+	unmaskedContexts := make([]*fftypes.Bytes32, 0, len(msg.Header.Topics))
+	nextPins := make([]*nextPinState, 0, len(msg.Header.Topics))
+	if pin.Masked {
+		// Private messages have one or more masked "pin" hashes that allow us to work
+		// out if it's the next message in the sequence, given the previous messages
+		if msg.Header.Group == nil || len(msg.Pins) == 0 || len(msg.Header.Topics) != len(msg.Pins) {
+			l.Errorf("Message '%s' in batch '%s' has invalid pin data pins=%v topics=%v", msg.Header.ID, manifest.ID, msg.Pins, msg.Header.Topics)
+			return nil
+		}
+		for i, pinStr := range msg.Pins {
+			var msgContext fftypes.Bytes32
+			pinSplit := strings.Split(pinStr, ":")
+			nonceStr := ""
+			if len(pinSplit) > 1 {
+				// We introduced a "HASH:NONCE" syntax into the pin strings, to aid debug, but the inclusion of the
+				// nonce after the hash is not necessary.
+				nonceStr = pinSplit[1]
+			}
+			err := msgContext.UnmarshalText([]byte(pinSplit[0]))
+			if err != nil {
+				l.Errorf("Message '%s' in batch '%s' has invalid pin at index %d: '%s'", msg.Header.ID, manifest.ID, i, pinStr)
 				return nil
 			}
-			for i, pinStr := range msg.Pins {
-				var msgContext fftypes.Bytes32
-				pinSplit := strings.Split(pinStr, ":")
-				nonceStr := ""
-				if len(pinSplit) > 1 {
-					// We introduced a "HASH:NONCE" syntax into the pin strings, to aid debug, but the inclusion of the
-					// nonce after the hash is not necessary.
-					nonceStr = pinSplit[1]
-				}
-				err := msgContext.UnmarshalText([]byte(pinSplit[0]))
-				if err != nil {
-					l.Errorf("Message '%s' in batch '%s' has invalid pin at index %d: '%s'", msg.Header.ID, manifest.ID, i, pinStr)
-					return nil
-				}
-				nextPin, err := state.checkMaskedContextReady(ctx, msg, msg.Header.Topics[i], pin.Sequence, &msgContext, nonceStr)
-				if err != nil || nextPin == nil {
-					return err
-				}
-				nextPins = append(nextPins, nextPin)
+			nextPin, err := state.checkMaskedContextReady(ctx, msg, msg.Header.Topics[i], pin.Sequence, &msgContext, nonceStr)
+			if err != nil || nextPin == nil {
+				return err
 			}
-		} else {
-			for _, topic := range msg.Header.Topics {
-				h := sha256.New()
-				h.Write([]byte(topic))
-				msgContext := fftypes.HashResult(h)
-				unmaskedContexts = append(unmaskedContexts, msgContext)
-				ready, err := state.checkUnmaskedContextReady(ctx, msgContext, msg, pin.Sequence)
-				if err != nil || !ready {
-					return err
-				}
+			nextPins = append(nextPins, nextPin)
+		}
+	} else {
+		for _, topic := range msg.Header.Topics {
+			h := sha256.New()
+			h.Write([]byte(topic))
+			msgContext := fftypes.HashResult(h)
+			unmaskedContexts = append(unmaskedContexts, msgContext)
+			ready, err := state.checkUnmaskedContextReady(ctx, msgContext, msg, pin.Sequence)
+			if err != nil || !ready {
+				return err
 			}
 		}
 
+	}
+
+	dispatched := false
+	var newState core.MessageState
+	if dataAvailable {
 		l.Debugf("Attempt dispatch msg=%s broadcastContexts=%v privatePins=%v", msg.Header.ID, unmaskedContexts, msg.Pins)
 		newState, dispatched, err = ag.attemptMessageDispatch(ctx, msg, data, manifest.TX.ID, state, pin)
 		if err != nil {
@@ -499,63 +503,61 @@ func (ag *aggregator) processMessage(ctx context.Context, manifest *core.BatchMa
 	return nil
 }
 
-func (ag *aggregator) attemptMessageDispatch(ctx context.Context, msg *core.Message, data core.DataArray, tx *fftypes.UUID, state *batchState, pin *core.Pin) (newState core.MessageState, dispatched bool, err error) {
-	var customCorrelator *fftypes.UUID
+func (ag *aggregator) attemptMessageDispatch(ctx context.Context, msg *core.Message, data core.DataArray, tx *fftypes.UUID, state *batchState, pin *core.Pin) (newState core.MessageState, valid bool, err error) {
 
 	// Check the pin signer is valid for the message
-	valid, err := ag.checkOnchainConsistency(ctx, msg, pin)
-	if err != nil {
+	if valid, err := ag.checkOnchainConsistency(ctx, msg, pin); err != nil || !valid {
 		return "", false, err
 	}
 
-	if valid {
-		// Verify we have all the blobs for the data
-		if resolved, err := ag.resolveBlobs(ctx, data); err != nil || !resolved {
+	// Verify we have all the blobs for the data
+	if resolved, err := ag.resolveBlobs(ctx, data); err != nil || !resolved {
+		return "", false, err
+	}
+
+	// For transfers, verify the transfer has come through
+	if msg.Header.Type == core.MessageTypeTransferBroadcast || msg.Header.Type == core.MessageTypeTransferPrivate {
+		fb := database.TokenTransferQueryFactory.NewFilter(ctx)
+		filter := fb.And(
+			fb.Eq("message", msg.Header.ID),
+		)
+		if transfers, _, err := ag.database.GetTokenTransfers(ctx, ag.namespace, filter); err != nil || len(transfers) == 0 {
+			log.L(ctx).Debugf("Transfer for message %s not yet available", msg.Header.ID)
 			return "", false, err
+		} else if !msg.Hash.Equals(transfers[0].MessageHash) {
+			log.L(ctx).Errorf("Message hash %s does not match hash recorded in transfer: %s", msg.Hash, transfers[0].MessageHash)
+			return "", false, nil
 		}
+	}
 
-		// For transfers, verify the transfer has come through
-		if msg.Header.Type == core.MessageTypeTransferBroadcast || msg.Header.Type == core.MessageTypeTransferPrivate {
-			fb := database.TokenTransferQueryFactory.NewFilter(ctx)
-			filter := fb.And(
-				fb.Eq("message", msg.Header.ID),
-			)
-			if transfers, _, err := ag.database.GetTokenTransfers(ctx, ag.namespace, filter); err != nil || len(transfers) == 0 {
-				log.L(ctx).Debugf("Transfer for message %s not yet available", msg.Header.ID)
-				return "", false, err
-			} else if !msg.Hash.Equals(transfers[0].MessageHash) {
-				log.L(ctx).Errorf("Message hash %s does not match hash recorded in transfer: %s", msg.Hash, transfers[0].MessageHash)
-				return "", false, nil
-			}
+	// Validate the message data
+	valid = true
+	var customCorrelator *fftypes.UUID
+	switch {
+	case msg.Header.Type == core.MessageTypeDefinition:
+		// We handle definition events in-line on the aggregator, as it would be confusing for apps to be
+		// dispatched subsequent events before we have processed the definition events they depend on.
+		handlerResult, err := ag.definitions.HandleDefinitionBroadcast(ctx, &state.BatchState, msg, data, tx)
+		log.L(ctx).Infof("Result of definition broadcast '%s' [%s]: %s", msg.Header.Tag, msg.Header.ID, handlerResult.Action)
+		if handlerResult.Action == definitions.ActionRetry {
+			return "", false, err
 		}
+		if handlerResult.Action == definitions.ActionWait {
+			return "", false, nil
+		}
+		if handlerResult.Action == definitions.ActionReject {
+			log.L(ctx).Warnf("Definition broadcast rejected: %s", err)
+		}
+		customCorrelator = handlerResult.CustomCorrelator
+		valid = handlerResult.Action == definitions.ActionConfirm
 
-		// Validate the message data
-		switch {
-		case msg.Header.Type == core.MessageTypeDefinition:
-			// We handle definition events in-line on the aggregator, as it would be confusing for apps to be
-			// dispatched subsequent events before we have processed the definition events they depend on.
-			handlerResult, err := ag.definitions.HandleDefinitionBroadcast(ctx, &state.BatchState, msg, data, tx)
-			log.L(ctx).Infof("Result of definition broadcast '%s' [%s]: %s", msg.Header.Tag, msg.Header.ID, handlerResult.Action)
-			if handlerResult.Action == definitions.ActionRetry {
-				return "", false, err
-			}
-			if handlerResult.Action == definitions.ActionWait {
-				return "", false, nil
-			}
-			if handlerResult.Action == definitions.ActionReject {
-				log.L(ctx).Warnf("Definition broadcast rejected: %s", err)
-			}
-			customCorrelator = handlerResult.CustomCorrelator
-			valid = handlerResult.Action == definitions.ActionConfirm
-
-		case msg.Header.Type == core.MessageTypeGroupInit:
-			// Already handled as part of resolving the context - do nothing.
+	case msg.Header.Type == core.MessageTypeGroupInit:
+		// Already handled as part of resolving the context - do nothing.
 
-		case len(msg.Data) > 0:
-			valid, err = ag.data.ValidateAll(ctx, data)
-			if err != nil {
-				return "", false, err
-			}
+	case len(msg.Data) > 0:
+		valid, err = ag.data.ValidateAll(ctx, data)
+		if err != nil {
+			return "", false, err
 		}
 	}
 


Test patch:
diff --git a/internal/events/aggregator_test.go b/internal/events/aggregator_test.go
index 61b3d1849..edc40eafc 100644
--- a/internal/events/aggregator_test.go
+++ b/internal/events/aggregator_test.go
@@ -1401,41 +1401,6 @@ func TestAttemptMessageDispatchFailValidateData(t *testing.T) {
 
 }
 
-func TestAttemptMessageDispatchBadSigner(t *testing.T) {
-	ag, cancel := newTestAggregator()
-	defer cancel()
-	bs := newBatchState(ag)
-
-	blobHash := fftypes.NewRandB32()
-
-	org1 := newTestOrg("org1")
-
-	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("InsertEvent", ag.ctx, mock.MatchedBy(func(ev *core.Event) bool {
-		return ev.Type == core.EventTypeMessageRejected
-	})).Return(nil)
-
-	_, dispatched, err := ag.attemptMessageDispatch(ag.ctx, &core.Message{
-		Header: core.MessageHeader{
-			ID:        fftypes.NewUUID(),
-			SignerRef: core.SignerRef{Key: "0x12345", Author: org1.DID},
-			Topics:    []string{"topic1"},
-		},
-	}, core.DataArray{
-		{ID: fftypes.NewUUID(), Hash: fftypes.NewRandB32(), Blob: &core.BlobRef{
-			Hash:   blobHash,
-			Public: "public-ref",
-		}},
-	}, nil, bs, &core.Pin{Signer: ""})
-	assert.NoError(t, err)
-	assert.True(t, dispatched)
-
-	err = bs.RunFinalize(ag.ctx)
-	assert.NoError(t, err)
-
-	mdi.AssertExpectations(t)
-}
-
 func TestAttemptMessageDispatchMissingBlobs(t *testing.T) {
 	ag, cancel := newTestAggregator()
 	defer cancel()
@@ -1669,38 +1634,20 @@ func TestDispatchBroadcastQueuesLaterDispatch(t *testing.T) {
 	mim.On("FindIdentityForVerifier", ag.ctx, mock.Anything, mock.Anything).Return(org1, nil)
 
 	mdm := ag.data.(*datamocks.Manager)
-	data1 := core.DataArray{}
-	data2 := core.DataArray{}
-	mdm.On("GetMessageWithDataCached", ag.ctx, msg1.Header.ID, data.CRORequirePublicBlobRefs).Return(msg1, data1, true, nil).Once()
-	mdm.On("GetMessageWithDataCached", ag.ctx, msg2.Header.ID, data.CRORequirePublicBlobRefs).Return(msg2, data2, true, nil).Once()
+	mdm.On("GetMessageWithDataCached", ag.ctx, msg1.Header.ID, data.CRORequirePublicBlobRefs).Return(msg1, core.DataArray{}, true, nil).Once()
+	mdm.On("GetMessageWithDataCached", ag.ctx, msg2.Header.ID, data.CRORequirePublicBlobRefs).Return(msg2, core.DataArray{}, true, nil).Once()
 
 	mdi := ag.database.(*databasemocks.Plugin)
 	mdi.On("GetPins", ag.ctx, "ns1", mock.Anything).Return([]*core.Pin{}, nil, nil)
 
-	mdh := ag.definitions.(*definitionsmocks.Handler)
-	result1 := definitions.HandlerResult{Action: definitions.ActionConfirm}
-	mdh.On("HandleDefinitionBroadcast", ag.ctx, mock.Anything, msg1, data1, manifest.TX.ID).Return(result1, nil).Once()
-	result2 := definitions.HandlerResult{Action: definitions.ActionWait}
-	mdh.On("HandleDefinitionBroadcast", ag.ctx, mock.Anything, msg2, data2, manifest.TX.ID).Return(result2, nil).Once()
-
 	// First message should dispatch
-	pin1 := &core.Pin{Sequence: 12345, Signer: msg1.Header.Key}
-	err := ag.processMessage(ag.ctx, manifest, pin1, 0, manifest.Messages[0], bs)
+	err := ag.processMessage(ag.ctx, manifest, &core.Pin{Sequence: 12345}, 0, manifest.Messages[0], bs)
 	assert.NoError(t, err)
 
-	// Second message should block
-	pin2 := &core.Pin{Sequence: 12346, Signer: msg2.Header.Key}
-	err = ag.processMessage(ag.ctx, manifest, pin2, 0, manifest.Messages[1], bs)
+	// Second message should not (mocks have Once limit on GetMessageData to confirm)
+	err = ag.processMessage(ag.ctx, manifest, &core.Pin{Sequence: 12346}, 0, manifest.Messages[1], bs)
 	assert.NoError(t, err)
 
-	assert.Len(t, bs.dispatchedMessages, 1)
-	assert.Equal(t, msg1.Header.ID, bs.dispatchedMessages[0].msgID)
-
-	h := sha256.New()
-	h.Write([]byte(msg2.Header.Topics[0]))
-	msgContext := fftypes.HashResult(h)
-	assert.Equal(t, pin2.Sequence, bs.unmaskedContexts[*msgContext].blockedBy)
-
 	mdi.AssertExpectations(t)
 	mdm.AssertExpectations(t)
 }
@@ -1717,15 +1664,12 @@ func TestDispatchPrivateQueuesLaterDispatch(t *testing.T) {
 	mim.On("FindIdentityForVerifier", ag.ctx, mock.Anything, mock.Anything).Return(org1, nil)
 
 	mdm := ag.data.(*datamocks.Manager)
-	data1 := core.DataArray{}
-	data2 := core.DataArray{{Blob: &core.BlobRef{Hash: fftypes.NewRandB32()}}}
-	mdm.On("GetMessageWithDataCached", ag.ctx, msg1.Header.ID, data.CRORequirePins).Return(msg1, data1, true, nil).Once()
-	mdm.On("ValidateAll", ag.ctx, data1).Return(true, nil)
-	mdm.On("GetMessageWithDataCached", ag.ctx, msg2.Header.ID, data.CRORequirePins).Return(msg2, data2, true, nil).Once()
+	mdm.On("GetMessageWithDataCached", ag.ctx, msg1.Header.ID, data.CRORequirePins).Return(msg1, core.DataArray{}, true, nil).Once()
+	mdm.On("GetMessageWithDataCached", ag.ctx, msg2.Header.ID, data.CRORequirePins).Return(msg2, core.DataArray{}, true, nil).Once()
 
 	initNPG := &nextPinGroupState{topic: "topic1", groupID: groupID}
-	member1NonceOne := initNPG.calcPinHash("did:firefly:org/org1", 1)
-	member1NonceTwo := initNPG.calcPinHash("did:firefly:org/org1", 2)
+	member1NonceOne := initNPG.calcPinHash("org1", 1)
+	member1NonceTwo := initNPG.calcPinHash("org1", 2)
 	h := sha256.New()
 	h.Write([]byte("topic1"))
 	context := fftypes.HashResult(h)
@@ -1733,25 +1677,19 @@ func TestDispatchPrivateQueuesLaterDispatch(t *testing.T) {
 	mdi := ag.database.(*databasemocks.Plugin)
 	mdi.On("GetNextPinsForContext", ag.ctx, "ns1", mock.Anything).Return([]*core.NextPin{
 		{Context: context, Nonce: 1 /* match member1NonceOne */, Identity: org1.DID, Hash: member1NonceOne},
-	}, nil).Once()
-	mdi.On("GetBlobMatchingHash", ag.ctx, data2[0].Blob.Hash).Return(nil, nil)
+	}, nil)
 
 	msg1.Pins = core.FFStringArray{member1NonceOne.String()}
 	msg2.Pins = core.FFStringArray{member1NonceTwo.String()}
 
 	// First message should dispatch
-	pin1 := &core.Pin{Masked: true, Sequence: 12345, Signer: msg1.Header.Key}
-	err := ag.processMessage(ag.ctx, manifest, pin1, 0, manifest.Messages[0], bs)
+	err := ag.processMessage(ag.ctx, manifest, &core.Pin{Masked: true, Sequence: 12345}, 0, manifest.Messages[0], bs)
 	assert.NoError(t, err)
 
-	// Second message should block
-	pin2 := &core.Pin{Masked: true, Sequence: 12346, Signer: msg2.Header.Key}
-	err = ag.processMessage(ag.ctx, manifest, pin2, 0, manifest.Messages[1], bs)
+	// Second message should not (mocks have Once limit on GetMessageData to confirm)
+	err = ag.processMessage(ag.ctx, manifest, &core.Pin{Masked: true, Sequence: 12346}, 0, manifest.Messages[1], bs)
 	assert.NoError(t, err)
 
-	assert.Len(t, bs.dispatchedMessages, 1)
-	assert.Equal(t, msg1.Header.ID, bs.dispatchedMessages[0].msgID)
-
 	mdi.AssertExpectations(t)
 	mdm.AssertExpectations(t)
 }
@@ -1838,27 +1776,30 @@ func TestDefinitionBroadcastRejectSignerLookupFail(t *testing.T) {
 func TestDefinitionBroadcastRejectSignerLookupWrongOrg(t *testing.T) {
 	ag, cancel := newTestAggregator()
 	defer cancel()
-	bs := newBatchState(ag)
 
 	msg1, _, _, _ := newTestManifest(core.MessageTypeDefinition, nil)
 
 	mim := ag.identity.(*identitymanagermocks.Manager)
 	mim.On("FindIdentityForVerifier", ag.ctx, mock.Anything, mock.Anything).Return(newTestOrg("org2"), nil)
 
-	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("InsertEvent", ag.ctx, mock.MatchedBy(func(ev *core.Event) bool {
-		return ev.Type == core.EventTypeMessageRejected
-	})).Return(nil)
-
-	_, valid, err := ag.attemptMessageDispatch(ag.ctx, msg1, nil, nil, bs, &core.Pin{Signer: "0x12345"})
+	_, valid, err := ag.attemptMessageDispatch(ag.ctx, msg1, nil, nil, &batchState{}, &core.Pin{Signer: "0x12345"})
 	assert.NoError(t, err)
-	assert.True(t, valid)
+	assert.False(t, valid)
 
-	err = bs.RunFinalize(ag.ctx)
+	mim.AssertExpectations(t)
+}
+
+func TestDefinitionBroadcastRejectBadSigner(t *testing.T) {
+	ag, cancel := newTestAggregator()
+	defer cancel()
+
+	msg1, _, org1, _ := newTestManifest(core.MessageTypeDefinition, nil)
+	msg1.Header.SignerRef = core.SignerRef{Key: "0x23456", Author: org1.DID}
+
+	_, valid, err := ag.attemptMessageDispatch(ag.ctx, msg1, nil, nil, &batchState{}, &core.Pin{Signer: "0x12345"})
 	assert.NoError(t, err)
+	assert.False(t, valid)
 
-	mim.AssertExpectations(t)
-	mdi.AssertExpectations(t)
 }
 
 func TestDefinitionBroadcastParkUnregisteredSignerIdentityClaim(t *testing.T) {
@@ -1883,30 +1824,20 @@ func TestDefinitionBroadcastParkUnregisteredSignerIdentityClaim(t *testing.T) {
 	msh.AssertExpectations(t)
 }
 
-func TestDefinitionBroadcastRootUnregistered(t *testing.T) {
+func TestDefinitionBroadcastRootUnregisteredOk(t *testing.T) {
 	ag, cancel := newTestAggregator()
 	defer cancel()
-	bs := newBatchState(ag)
 
 	msg1, _, _, _ := newTestManifest(core.MessageTypeDefinition, nil)
 
 	mim := ag.identity.(*identitymanagermocks.Manager)
 	mim.On("FindIdentityForVerifier", ag.ctx, mock.Anything, mock.Anything).Return(nil, nil)
 
-	mdi := ag.database.(*databasemocks.Plugin)
-	mdi.On("InsertEvent", ag.ctx, mock.MatchedBy(func(ev *core.Event) bool {
-		return ev.Type == core.EventTypeMessageRejected
-	})).Return(nil)
-
-	_, valid, err := ag.attemptMessageDispatch(ag.ctx, msg1, nil, nil, bs, &core.Pin{Signer: "0x12345"})
-	assert.NoError(t, err)
-	assert.True(t, valid)
-
-	err = bs.RunFinalize(ag.ctx)
+	_, valid, err := ag.attemptMessageDispatch(ag.ctx, msg1, nil, nil, &batchState{}, &core.Pin{Signer: "0x12345"})
 	assert.NoError(t, err)
+	assert.False(t, valid)
 
 	mim.AssertExpectations(t)
-	mdi.AssertExpectations(t)
 }
 
 func TestDefinitionBroadcastActionWait(t *testing.T) {