Repository: hyperledger/firefly
Pull request #971 (2022-08-16 17:53:05)

Fix group race condition, optimize transaction cache
Insert local group only after inserting groupinit message.
This ensures no other threads pick up on the group before the
message is sequenced.

Update transaction cache after initial insert.


Patch:
diff --git a/internal/privatemessaging/groupmanager.go b/internal/privatemessaging/groupmanager.go
index ebffe30f9..eab422a45 100644
--- a/internal/privatemessaging/groupmanager.go
+++ b/internal/privatemessaging/groupmanager.go
@@ -104,19 +104,6 @@ func (gm *groupManager) groupInit(ctx context.Context, signer *core.SignerRef, g
 	}
 	group.LocalNamespace = gm.namespace.LocalName
 
-	// In the case of groups, we actually write the unconfirmed group directly to our database.
-	// So it can be used straight away.
-	// We're able to do this by making the identifier of the group a hash of the identity fields
-	// (name, ledger and member list), as that is all the group contains. There's no data in there.
-	if err = gm.database.UpsertGroup(ctx, group, database.UpsertOptimizationNew /* we think we're first */); err != nil {
-		return err
-	}
-
-	// Write as data to the local store
-	if err = gm.database.UpsertData(ctx, data, database.UpsertOptimizationNew); err != nil {
-		return err
-	}
-
 	// Create a private send message referring to the data
 	msg := &core.Message{
 		State:          core.MessageStateReady,
@@ -134,18 +121,28 @@ func (gm *groupManager) groupInit(ctx context.Context, signer *core.SignerRef, g
 			{ID: data.ID, Hash: data.Hash},
 		},
 	}
-
-	// Seal the message
-	err = msg.Seal(ctx)
-	if err == nil {
-		// Store the message - this asynchronously triggers the next step in process
-		err = gm.database.UpsertMessage(ctx, msg, database.UpsertOptimizationNew)
-	}
-	if err == nil {
-		log.L(ctx).Infof("Created new group %s", group.Hash)
+	if err = msg.Seal(ctx); err == nil {
+		err = gm.database.RunAsGroup(ctx, func(ctx context.Context) error {
+			// Write as data to the local store
+			if err = gm.database.UpsertData(ctx, data, database.UpsertOptimizationNew); err != nil {
+				return err
+			}
+
+			// Store the message - this asynchronously triggers the next step in process
+			if err = gm.database.UpsertMessage(ctx, msg, database.UpsertOptimizationNew); err != nil {
+				return err
+			}
+
+			// Write the unconfirmed group directly to our database, so it can be used straight away.
+			// We're able to do this by making the identifier of the group a hash of the identity fields
+			// (name, ledger and member list), as that is all the group contains. There's no data in there.
+			return gm.database.UpsertGroup(ctx, group, database.UpsertOptimizationNew /* we think we're first */)
+		})
+		if err == nil {
+			log.L(ctx).Infof("Created new group %s", group.Hash)
+		}
 	}
 	return err
-
 }
 
 func (gm *groupManager) GetGroupByID(ctx context.Context, hash string) (*core.Group, error) {
diff --git a/internal/txcommon/txcommon.go b/internal/txcommon/txcommon.go
index 3f5dbc52d..eb3b2856c 100644
--- a/internal/txcommon/txcommon.go
+++ b/internal/txcommon/txcommon.go
@@ -105,6 +105,7 @@ func (t *transactionHelper) SubmitNewTransaction(ctx context.Context, txType cor
 		return nil, err
 	}
 
+	t.updateTransactionsCache(tx)
 	return tx.ID, nil
 }
 
@@ -144,7 +145,6 @@ func (t *transactionHelper) PersistTransaction(ctx context.Context, id *fftypes.
 	}
 
 	t.updateTransactionsCache(tx)
-
 	return true, nil
 }
 
@@ -162,8 +162,8 @@ func (t *transactionHelper) AddBlockchainTX(ctx context.Context, tx *core.Transa
 	if err != nil {
 		return err
 	}
-	t.updateTransactionsCache(tx)
 
+	t.updateTransactionsCache(tx)
 	return nil
 }
 


Test patch:
diff --git a/internal/privatemessaging/groupmanager_test.go b/internal/privatemessaging/groupmanager_test.go
index cd26248c2..1a287009e 100644
--- a/internal/privatemessaging/groupmanager_test.go
+++ b/internal/privatemessaging/groupmanager_test.go
@@ -47,6 +47,8 @@ func TestGroupInitWriteGroupFail(t *testing.T) {
 
 	mdi := pm.database.(*databasemocks.Plugin)
 	mdi.On("UpsertGroup", mock.Anything, mock.Anything, database.UpsertOptimizationNew).Return(fmt.Errorf("pop"))
+	mdi.On("UpsertData", mock.Anything, mock.Anything, database.UpsertOptimizationNew).Return(nil)
+	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationNew).Return(nil)
 
 	group := &core.Group{
 		GroupIdentity: core.GroupIdentity{
@@ -59,6 +61,32 @@ func TestGroupInitWriteGroupFail(t *testing.T) {
 	group.Seal()
 	err := pm.groupInit(pm.ctx, &core.SignerRef{}, group)
 	assert.Regexp(t, "pop", err)
+
+	mdi.AssertExpectations(t)
+}
+
+func TestGroupInitWriteMessageFail(t *testing.T) {
+
+	pm, cancel := newTestPrivateMessaging(t)
+	defer cancel()
+
+	mdi := pm.database.(*databasemocks.Plugin)
+	mdi.On("UpsertData", mock.Anything, mock.Anything, database.UpsertOptimizationNew).Return(nil)
+	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationNew).Return(fmt.Errorf("pop"))
+
+	group := &core.Group{
+		GroupIdentity: core.GroupIdentity{
+			Namespace: "ns1",
+			Members: core.Members{
+				{Identity: "id1", Node: fftypes.NewUUID()},
+			},
+		},
+	}
+	group.Seal()
+	err := pm.groupInit(pm.ctx, &core.SignerRef{}, group)
+	assert.Regexp(t, "pop", err)
+
+	mdi.AssertExpectations(t)
 }
 
 func TestGroupInitWriteDataFail(t *testing.T) {
@@ -67,7 +95,6 @@ func TestGroupInitWriteDataFail(t *testing.T) {
 	defer cancel()
 
 	mdi := pm.database.(*databasemocks.Plugin)
-	mdi.On("UpsertGroup", mock.Anything, mock.Anything, database.UpsertOptimizationNew).Return(nil)
 	mdi.On("UpsertData", mock.Anything, mock.Anything, database.UpsertOptimizationNew).Return(fmt.Errorf("pop"))
 
 	group := &core.Group{
@@ -81,6 +108,8 @@ func TestGroupInitWriteDataFail(t *testing.T) {
 	group.Seal()
 	err := pm.groupInit(pm.ctx, &core.SignerRef{}, group)
 	assert.Regexp(t, "pop", err)
+
+	mdi.AssertExpectations(t)
 }
 
 func TestResolveInitGroupMissingData(t *testing.T) {