Repository: hyperledger/firefly (go)

Issue:
Pool activation happens more than once
Token pool activation is sometimes happening twice, and only one of the operations gets resolved.

This is partly because activation operations are resolved only by TX ID, not by the actual operation ID.
https://github.com/hyperledger/firefly/blob/5afbecf2563d290f014e45b2542e037947ecd3ec/internal/events/token_pool_created.go#L68

Need to revisit this flow.

Pull request:
Resolve pool activation operations more cleanly
Fixes #743

Patch:
diff --git a/internal/assets/token_pool.go b/internal/assets/token_pool.go
index 7b2a9819c..8018b0f08 100644
--- a/internal/assets/token_pool.go
+++ b/internal/assets/token_pool.go
@@ -24,6 +24,7 @@ import (
 	"github.com/hyperledger/firefly/pkg/database"
 	"github.com/hyperledger/firefly/pkg/fftypes"
 	"github.com/hyperledger/firefly/pkg/i18n"
+	"github.com/hyperledger/firefly/pkg/log"
 )
 
 func (am *assetManager) CreateTokenPool(ctx context.Context, ns string, pool *fftypes.TokenPool, waitConfirm bool) (*fftypes.TokenPool, error) {
@@ -104,13 +105,29 @@ func (am *assetManager) ActivateTokenPool(ctx context.Context, pool *fftypes.Tok
 		return err
 	}
 
-	op := fftypes.NewOperation(
-		plugin,
-		pool.Namespace,
-		pool.TX.ID,
-		fftypes.OpTypeTokenActivatePool)
-	txcommon.AddTokenPoolActivateInputs(op, pool.ID)
-	if err := am.database.InsertOperation(ctx, op); err != nil {
+	var op *fftypes.Operation
+	err = am.database.RunAsGroup(ctx, func(ctx context.Context) (err error) {
+		fb := database.OperationQueryFactory.NewFilter(ctx)
+		filter := fb.And(
+			fb.Eq("tx", pool.TX.ID),
+			fb.Eq("type", fftypes.OpTypeTokenActivatePool),
+		)
+		if existing, _, err := am.database.GetOperations(ctx, filter); err != nil {
+			return err
+		} else if len(existing) > 0 {
+			log.L(ctx).Debugf("Dropping duplicate token pool activation request for pool %s", pool.ID)
+			return nil
+		}
+
+		op = fftypes.NewOperation(
+			plugin,
+			pool.Namespace,
+			pool.TX.ID,
+			fftypes.OpTypeTokenActivatePool)
+		txcommon.AddTokenPoolActivateInputs(op, pool.ID)
+		return am.database.InsertOperation(ctx, op)
+	})
+	if err != nil || op == nil {
 		return err
 	}
 
diff --git a/internal/events/token_pool_created.go b/internal/events/token_pool_created.go
index 6ec847a33..868c2d259 100644
--- a/internal/events/token_pool_created.go
+++ b/internal/events/token_pool_created.go
@@ -61,13 +61,6 @@ func (em *eventManager) confirmPool(ctx context.Context, pool *fftypes.TokenPool
 		}
 		em.emitBlockchainEventMetric(ev)
 	}
-	if op, err := em.findTXOperation(ctx, pool.TX.ID, fftypes.OpTypeTokenActivatePool); err != nil {
-		return err
-	} else if op == nil {
-		log.L(ctx).Warnf("No activate operation found for token pool transaction=%s", pool.TX.ID)
-	} else if err := em.database.ResolveOperation(ctx, op.ID, fftypes.OpStatusSucceeded, "", nil); err != nil {
-		return err
-	}
 	if _, err := em.txHelper.PersistTransaction(ctx, pool.Namespace, pool.TX.ID, pool.TX.Type, ev.BlockchainTXID); err != nil {
 		return err
 	}
diff --git a/internal/tokens/fftokens/fftokens.go b/internal/tokens/fftokens/fftokens.go
index 31ba21497..d186fd195 100644
--- a/internal/tokens/fftokens/fftokens.go
+++ b/internal/tokens/fftokens/fftokens.go
@@ -497,13 +497,14 @@ func (ft *FFTokens) CreateTokenPool(ctx context.Context, opID *fftypes.UUID, poo
 		return false, ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgTokensRESTErr)
 	}
 	if res.StatusCode() == 200 {
-		// Handle synchronous response (202 will be handled by later websocket listener)
+		// HTTP 200: Creation was successful, and pool details are in response body
 		var obj fftypes.JSONObject
 		if err := json.Unmarshal(res.Body(), &obj); err != nil {
 			return false, i18n.WrapError(ctx, err, i18n.MsgJSONObjectParseFailed, res.Body())
 		}
 		return true, ft.handleTokenPoolCreate(ctx, obj)
 	}
+	// Default (HTTP 202): Request was accepted, and success/failure status will be delivered via websocket
 	return false, nil
 }
 
@@ -519,13 +520,18 @@ func (ft *FFTokens) ActivateTokenPool(ctx context.Context, opID *fftypes.UUID, p
 		return false, ffresty.WrapRestErr(ctx, res, err, coremsgs.MsgTokensRESTErr)
 	}
 	if res.StatusCode() == 200 {
-		// Handle synchronous response (202 will be handled by later websocket listener)
+		// HTTP 200: Activation was successful, and pool details are in response body
 		var obj fftypes.JSONObject
 		if err := json.Unmarshal(res.Body(), &obj); err != nil {
 			return false, i18n.WrapError(ctx, err, i18n.MsgJSONObjectParseFailed, res.Body())
 		}
 		return true, ft.handleTokenPoolCreate(ctx, obj)
+	} else if res.StatusCode() == 204 {
+		// HTTP 204: Activation was successful, but pool details are not available
+		// This will resolve the operation, but connector is responsible for re-delivering pool details on the websocket.
+		return true, nil
 	}
+	// Default (HTTP 202): Request was accepted, and success/failure status will be delivered via websocket
 	return false, nil
 }
 


Test patch:
diff --git a/internal/assets/token_pool_test.go b/internal/assets/token_pool_test.go
index c665ef44e..2f05e1a08 100644
--- a/internal/assets/token_pool_test.go
+++ b/internal/assets/token_pool_test.go
@@ -459,6 +459,7 @@ func TestActivateTokenPool(t *testing.T) {
 
 	mdi := am.database.(*databasemocks.Plugin)
 	mom := am.operations.(*operationmocks.Manager)
+	mdi.On("GetOperations", context.Background(), mock.Anything).Return(nil, nil, nil)
 	mdi.On("InsertOperation", context.Background(), mock.MatchedBy(func(op *fftypes.Operation) bool {
 		return op.Type == fftypes.OpTypeTokenActivatePool
 	})).Return(nil)
@@ -497,6 +498,7 @@ func TestActivateTokenPoolOpInsertFail(t *testing.T) {
 	}
 
 	mdi := am.database.(*databasemocks.Plugin)
+	mdi.On("GetOperations", context.Background(), mock.Anything).Return(nil, nil, nil)
 	mdi.On("InsertOperation", context.Background(), mock.MatchedBy(func(op *fftypes.Operation) bool {
 		return op.Type == fftypes.OpTypeTokenActivatePool
 	})).Return(fmt.Errorf("pop"))
@@ -517,8 +519,8 @@ func TestActivateTokenPoolFail(t *testing.T) {
 	}
 
 	mdi := am.database.(*databasemocks.Plugin)
-	mth := am.txHelper.(*txcommonmocks.Helper)
 	mom := am.operations.(*operationmocks.Manager)
+	mdi.On("GetOperations", context.Background(), mock.Anything).Return(nil, nil, nil)
 	mdi.On("InsertOperation", context.Background(), mock.MatchedBy(func(op *fftypes.Operation) bool {
 		return op.Type == fftypes.OpTypeTokenActivatePool
 	})).Return(nil)
@@ -531,10 +533,45 @@ func TestActivateTokenPoolFail(t *testing.T) {
 	assert.EqualError(t, err, "pop")
 
 	mdi.AssertExpectations(t)
-	mth.AssertExpectations(t)
 	mom.AssertExpectations(t)
 }
 
+func TestActivateTokenPoolExisting(t *testing.T) {
+	am, cancel := newTestAssets(t)
+	defer cancel()
+
+	pool := &fftypes.TokenPool{
+		Namespace: "ns1",
+		Connector: "magic-tokens",
+	}
+
+	mdi := am.database.(*databasemocks.Plugin)
+	mdi.On("GetOperations", context.Background(), mock.Anything).Return([]*fftypes.Operation{{}}, nil, nil)
+
+	err := am.ActivateTokenPool(context.Background(), pool)
+	assert.NoError(t, err)
+
+	mdi.AssertExpectations(t)
+}
+
+func TestActivateTokenPoolExistingFail(t *testing.T) {
+	am, cancel := newTestAssets(t)
+	defer cancel()
+
+	pool := &fftypes.TokenPool{
+		Namespace: "ns1",
+		Connector: "magic-tokens",
+	}
+
+	mdi := am.database.(*databasemocks.Plugin)
+	mdi.On("GetOperations", context.Background(), mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
+
+	err := am.ActivateTokenPool(context.Background(), pool)
+	assert.EqualError(t, err, "pop")
+
+	mdi.AssertExpectations(t)
+}
+
 func TestActivateTokenPoolSyncSuccess(t *testing.T) {
 	am, cancel := newTestAssets(t)
 	defer cancel()
@@ -547,6 +584,7 @@ func TestActivateTokenPoolSyncSuccess(t *testing.T) {
 	mdi := am.database.(*databasemocks.Plugin)
 	mth := am.txHelper.(*txcommonmocks.Helper)
 	mom := am.operations.(*operationmocks.Manager)
+	mdi.On("GetOperations", context.Background(), mock.Anything).Return(nil, nil, nil)
 	mdi.On("InsertOperation", context.Background(), mock.MatchedBy(func(op *fftypes.Operation) bool {
 		return op.Type == fftypes.OpTypeTokenActivatePool
 	})).Return(nil)
diff --git a/internal/events/token_pool_created_test.go b/internal/events/token_pool_created_test.go
index 1268df460..9fd5bba6f 100644
--- a/internal/events/token_pool_created_test.go
+++ b/internal/events/token_pool_created_test.go
@@ -98,7 +98,6 @@ func TestTokenPoolCreatedConfirm(t *testing.T) {
 	mth := em.txHelper.(*txcommonmocks.Helper)
 	mdm := em.data.(*datamocks.Manager)
 
-	opID := fftypes.NewUUID()
 	txID := fftypes.NewUUID()
 	info1 := fftypes.JSONObject{"pool": "info"}
 	info2 := fftypes.JSONObject{"block": "info"}
@@ -140,10 +139,6 @@ func TestTokenPoolCreatedConfirm(t *testing.T) {
 	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *fftypes.Event) bool {
 		return e.Type == fftypes.EventTypeBlockchainEventReceived
 	})).Return(nil).Once()
-	mdi.On("GetOperations", em.ctx, mock.Anything).Return([]*fftypes.Operation{{
-		ID: opID,
-	}}, nil, nil)
-	mdi.On("ResolveOperation", em.ctx, opID, fftypes.OpStatusSucceeded, "", mock.Anything).Return(nil)
 	mth.On("PersistTransaction", mock.Anything, "ns1", txID, fftypes.TransactionTypeTokenPool, "0xffffeeee").Return(true, nil).Once()
 	mdi.On("UpsertTokenPool", em.ctx, storedPool).Return(nil).Once()
 	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *fftypes.Event) bool {
@@ -289,7 +284,6 @@ func TestTokenPoolCreatedMigrate(t *testing.T) {
 	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *fftypes.Event) bool {
 		return e.Type == fftypes.EventTypeBlockchainEventReceived
 	})).Return(nil).Once()
-	mdi.On("GetOperations", em.ctx, mock.Anything).Return([]*fftypes.Operation{}, nil, nil)
 	mth.On("PersistTransaction", mock.Anything, "ns1", txID, fftypes.TransactionTypeTokenPool, "0xffffeeee").Return(true, nil).Once()
 	mdi.On("UpsertTokenPool", em.ctx, storedPool).Return(nil).Once()
 	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *fftypes.Event) bool {
@@ -341,88 +335,6 @@ func TestConfirmPoolBlockchainEventFail(t *testing.T) {
 	mth.AssertExpectations(t)
 }
 
-func TestConfirmPoolGetOpsFail(t *testing.T) {
-	em, cancel := newTestEventManager(t)
-	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
-	mth := em.txHelper.(*txcommonmocks.Helper)
-
-	txID := fftypes.NewUUID()
-	storedPool := &fftypes.TokenPool{
-		Namespace: "ns1",
-		ID:        fftypes.NewUUID(),
-		Key:       "0x0",
-		State:     fftypes.TokenPoolStatePending,
-		TX: fftypes.TransactionRef{
-			Type: fftypes.TransactionTypeTokenPool,
-			ID:   txID,
-		},
-	}
-	event := &blockchain.Event{
-		BlockchainTXID: "0xffffeeee",
-		Name:           "TokenPool",
-		ProtocolID:     "tx1",
-	}
-
-	mdi.On("GetBlockchainEventByProtocolID", mock.Anything, "ns1", (*fftypes.UUID)(nil), event.ProtocolID).Return(nil, nil)
-	mth.On("InsertBlockchainEvent", em.ctx, mock.MatchedBy(func(e *fftypes.BlockchainEvent) bool {
-		return e.Name == event.Name
-	})).Return(nil)
-	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *fftypes.Event) bool {
-		return e.Type == fftypes.EventTypeBlockchainEventReceived
-	})).Return(nil)
-	mdi.On("GetOperations", em.ctx, mock.Anything).Return(nil, nil, fmt.Errorf("pop"))
-
-	err := em.confirmPool(em.ctx, storedPool, event)
-	assert.EqualError(t, err, "pop")
-
-	mdi.AssertExpectations(t)
-	mth.AssertExpectations(t)
-}
-
-func TestConfirmPoolResolveOpFail(t *testing.T) {
-	em, cancel := newTestEventManager(t)
-	defer cancel()
-	mdi := em.database.(*databasemocks.Plugin)
-	mth := em.txHelper.(*txcommonmocks.Helper)
-
-	opID := fftypes.NewUUID()
-	txID := fftypes.NewUUID()
-	storedPool := &fftypes.TokenPool{
-		Namespace: "ns1",
-		ID:        fftypes.NewUUID(),
-		Key:       "0x0",
-		State:     fftypes.TokenPoolStatePending,
-		TX: fftypes.TransactionRef{
-			Type: fftypes.TransactionTypeTokenPool,
-			ID:   txID,
-		},
-	}
-	event := &blockchain.Event{
-		BlockchainTXID: "0xffffeeee",
-		Name:           "TokenPool",
-		ProtocolID:     "tx1",
-	}
-
-	mdi.On("GetBlockchainEventByProtocolID", mock.Anything, "ns1", (*fftypes.UUID)(nil), event.ProtocolID).Return(nil, nil)
-	mth.On("InsertBlockchainEvent", em.ctx, mock.MatchedBy(func(e *fftypes.BlockchainEvent) bool {
-		return e.Name == event.Name
-	})).Return(nil)
-	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *fftypes.Event) bool {
-		return e.Type == fftypes.EventTypeBlockchainEventReceived
-	})).Return(nil)
-	mdi.On("GetOperations", em.ctx, mock.Anything).Return([]*fftypes.Operation{{
-		ID: opID,
-	}}, nil, nil)
-	mdi.On("ResolveOperation", em.ctx, opID, fftypes.OpStatusSucceeded, "", mock.Anything).Return(fmt.Errorf("pop"))
-
-	err := em.confirmPool(em.ctx, storedPool, event)
-	assert.EqualError(t, err, "pop")
-
-	mdi.AssertExpectations(t)
-	mth.AssertExpectations(t)
-}
-
 func TestConfirmPoolTxFail(t *testing.T) {
 	em, cancel := newTestEventManager(t)
 	defer cancel()
@@ -453,7 +365,6 @@ func TestConfirmPoolTxFail(t *testing.T) {
 	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *fftypes.Event) bool {
 		return e.Type == fftypes.EventTypeBlockchainEventReceived
 	})).Return(nil)
-	mdi.On("GetOperations", em.ctx, mock.Anything).Return([]*fftypes.Operation{}, nil, nil)
 	mth.On("PersistTransaction", mock.Anything, "ns1", txID, fftypes.TransactionTypeTokenPool, "0xffffeeee").Return(false, fmt.Errorf("pop"))
 
 	err := em.confirmPool(em.ctx, storedPool, event)
@@ -492,7 +403,6 @@ func TestConfirmPoolUpsertFail(t *testing.T) {
 	mdi.On("InsertEvent", em.ctx, mock.MatchedBy(func(e *fftypes.Event) bool {
 		return e.Type == fftypes.EventTypeBlockchainEventReceived
 	})).Return(nil)
-	mdi.On("GetOperations", em.ctx, mock.Anything).Return([]*fftypes.Operation{}, nil, nil)
 	mth.On("PersistTransaction", mock.Anything, "ns1", txID, fftypes.TransactionTypeTokenPool, "0xffffeeee").Return(true, nil).Once()
 	mdi.On("UpsertTokenPool", em.ctx, storedPool).Return(fmt.Errorf("pop"))
 
diff --git a/internal/tokens/fftokens/fftokens_test.go b/internal/tokens/fftokens/fftokens_test.go
index fe02b7d70..ee97ba627 100644
--- a/internal/tokens/fftokens/fftokens_test.go
+++ b/internal/tokens/fftokens/fftokens_test.go
@@ -418,6 +418,41 @@ func TestActivateTokenPoolSynchronousBadResponse(t *testing.T) {
 	assert.Regexp(t, "FF00127", err)
 }
 
+func TestActivateTokenPoolNoContent(t *testing.T) {
+	h, _, _, httpURL, done := newTestFFTokens(t)
+	defer done()
+
+	opID := fftypes.NewUUID()
+	poolConfig := map[string]interface{}{
+		"foo": "bar",
+	}
+	pool := &fftypes.TokenPool{
+		Locator: "N1",
+		Config:  poolConfig,
+	}
+
+	httpmock.RegisterResponder("POST", fmt.Sprintf("%s/api/v1/activatepool", httpURL),
+		func(req *http.Request) (*http.Response, error) {
+			body := make(fftypes.JSONObject)
+			err := json.NewDecoder(req.Body).Decode(&body)
+			assert.NoError(t, err)
+			assert.Equal(t, fftypes.JSONObject{
+				"requestId":   opID.String(),
+				"poolLocator": "N1",
+				"config":      poolConfig,
+			}, body)
+
+			res := &http.Response{
+				StatusCode: 204,
+			}
+			return res, nil
+		})
+
+	complete, err := h.ActivateTokenPool(context.Background(), opID, pool)
+	assert.True(t, complete)
+	assert.NoError(t, err)
+}
+
 func TestMintTokens(t *testing.T) {
 	h, _, _, httpURL, done := newTestFFTokens(t)
 	defer done()