Repository: hyperledger/firefly (go)

Issue:
Messages stuck in sent state after creating invalid identity
I added a new e2e test which created an invalid custom identity (using an already registered key and with a parent in a different org). The identity did not get created, but I did not get any events or messages that indicated that anything had gone wrong. All messages sent for creating the identity are stuck in `sent` state.

e.g.
<img width="1503" alt="Screenshot 2022-08-10 at 17 54 19" src="https://user-images.githubusercontent.com/110258282/183973200-dfa44527-82f3-4145-ad36-4e61ad25d185.png">

Logs:
* [firefly_core_0.log](https://github.com/hyperledger/firefly/files/9302448/firefly_core_0.log)
* [firefly_core_1.log](https://github.com/hyperledger/firefly/files/9302449/firefly_core_1.log)


Pull request:
Messages that fail validation should be rejected
There was some confusion between "message is invalid" and "message could not yet
be dispatched" in the event aggregator. This attempts to separate those two
situations, as the former should be an immediate rejection.

Fixes #952

Patch:
diff --git a/internal/events/aggregator.go b/internal/events/aggregator.go
index 98e294749..d2e2def37 100644
--- a/internal/events/aggregator.go
+++ b/internal/events/aggregator.go
@@ -414,6 +414,11 @@ 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
@@ -426,59 +431,50 @@ 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)
-		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)
+	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
 			}
-			nextPin, err := state.checkMaskedContextReady(ctx, msg, msg.Header.Topics[i], pin.Sequence, &msgContext, nonceStr)
-			if err != nil || nextPin == nil {
-				return err
+			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)
 			}
-			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
+		} 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 {
@@ -503,61 +499,63 @@ 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, valid bool, err error) {
+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
 
 	// Check the pin signer is valid for the message
-	if valid, err := ag.checkOnchainConsistency(ctx, msg, pin); err != nil || !valid {
-		return "", false, err
-	}
-
-	// Verify we have all the blobs for the data
-	if resolved, err := ag.resolveBlobs(ctx, data); err != nil || !resolved {
+	valid, err := ag.checkOnchainConsistency(ctx, msg, pin)
+	if err != nil {
 		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)
+	if valid {
+		// Verify we have all the blobs for the data
+		if resolved, err := ag.resolveBlobs(ctx, data); err != nil || !resolved {
 			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)
+		// 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
+			}
 		}
-		customCorrelator = handlerResult.CustomCorrelator
-		valid = handlerResult.Action == definitions.ActionConfirm
 
-	case msg.Header.Type == core.MessageTypeGroupInit:
-		// Already handled as part of resolving the context - do nothing.
+		// 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 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 edc40eafc..61b3d1849 100644
--- a/internal/events/aggregator_test.go
+++ b/internal/events/aggregator_test.go
@@ -1401,6 +1401,41 @@ 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()
@@ -1634,20 +1669,38 @@ func TestDispatchBroadcastQueuesLaterDispatch(t *testing.T) {
 	mim.On("FindIdentityForVerifier", ag.ctx, mock.Anything, mock.Anything).Return(org1, nil)
 
 	mdm := ag.data.(*datamocks.Manager)
-	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()
+	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()
 
 	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
-	err := ag.processMessage(ag.ctx, manifest, &core.Pin{Sequence: 12345}, 0, manifest.Messages[0], bs)
+	pin1 := &core.Pin{Sequence: 12345, Signer: msg1.Header.Key}
+	err := ag.processMessage(ag.ctx, manifest, pin1, 0, manifest.Messages[0], bs)
 	assert.NoError(t, err)
 
-	// 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)
+	// 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)
 	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)
 }
@@ -1664,12 +1717,15 @@ func TestDispatchPrivateQueuesLaterDispatch(t *testing.T) {
 	mim.On("FindIdentityForVerifier", ag.ctx, mock.Anything, mock.Anything).Return(org1, nil)
 
 	mdm := ag.data.(*datamocks.Manager)
-	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()
+	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()
 
 	initNPG := &nextPinGroupState{topic: "topic1", groupID: groupID}
-	member1NonceOne := initNPG.calcPinHash("org1", 1)
-	member1NonceTwo := initNPG.calcPinHash("org1", 2)
+	member1NonceOne := initNPG.calcPinHash("did:firefly:org/org1", 1)
+	member1NonceTwo := initNPG.calcPinHash("did:firefly:org/org1", 2)
 	h := sha256.New()
 	h.Write([]byte("topic1"))
 	context := fftypes.HashResult(h)
@@ -1677,19 +1733,25 @@ 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)
+	}, nil).Once()
+	mdi.On("GetBlobMatchingHash", ag.ctx, data2[0].Blob.Hash).Return(nil, nil)
 
 	msg1.Pins = core.FFStringArray{member1NonceOne.String()}
 	msg2.Pins = core.FFStringArray{member1NonceTwo.String()}
 
 	// First message should dispatch
-	err := ag.processMessage(ag.ctx, manifest, &core.Pin{Masked: true, Sequence: 12345}, 0, manifest.Messages[0], bs)
+	pin1 := &core.Pin{Masked: true, Sequence: 12345, Signer: msg1.Header.Key}
+	err := ag.processMessage(ag.ctx, manifest, pin1, 0, manifest.Messages[0], bs)
 	assert.NoError(t, err)
 
-	// 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)
+	// 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)
 	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)
 }
@@ -1776,30 +1838,27 @@ 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)
 
-	_, 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)
-}
-
-func TestDefinitionBroadcastRejectBadSigner(t *testing.T) {
-	ag, cancel := newTestAggregator()
-	defer cancel()
+	mdi := ag.database.(*databasemocks.Plugin)
+	mdi.On("InsertEvent", ag.ctx, mock.MatchedBy(func(ev *core.Event) bool {
+		return ev.Type == core.EventTypeMessageRejected
+	})).Return(nil)
 
-	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, bs, &core.Pin{Signer: "0x12345"})
+	assert.NoError(t, err)
+	assert.True(t, valid)
 
-	_, valid, err := ag.attemptMessageDispatch(ag.ctx, msg1, nil, nil, &batchState{}, &core.Pin{Signer: "0x12345"})
+	err = bs.RunFinalize(ag.ctx)
 	assert.NoError(t, err)
-	assert.False(t, valid)
 
+	mim.AssertExpectations(t)
+	mdi.AssertExpectations(t)
 }
 
 func TestDefinitionBroadcastParkUnregisteredSignerIdentityClaim(t *testing.T) {
@@ -1824,20 +1883,30 @@ func TestDefinitionBroadcastParkUnregisteredSignerIdentityClaim(t *testing.T) {
 	msh.AssertExpectations(t)
 }
 
-func TestDefinitionBroadcastRootUnregisteredOk(t *testing.T) {
+func TestDefinitionBroadcastRootUnregistered(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)
 
-	_, valid, err := ag.attemptMessageDispatch(ag.ctx, msg1, nil, nil, &batchState{}, &core.Pin{Signer: "0x12345"})
+	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)
 	assert.NoError(t, err)
-	assert.False(t, valid)
 
 	mim.AssertExpectations(t)
+	mdi.AssertExpectations(t)
 }
 
 func TestDefinitionBroadcastActionWait(t *testing.T) {