Misc fixes for operations

internal/assets/token_pool.go

@@ -90,10 +90,10 @@ func (am *assetManager) createTokenPoolInternal(ctx context.Context, pool *fftyp
 	}
 
 	if complete, err := plugin.CreateTokenPool(ctx, op.ID, pool); err != nil {
+		am.txHelper.WriteOperationFailure(ctx, op.ID, err)
 		return nil, err
 	} else if complete {
-		update := database.OperationQueryFactory.NewUpdate(ctx).Set("status", fftypes.OpStatusSucceeded)
-		return pool, am.database.UpdateOperation(ctx, op.ID, update)
+		am.txHelper.WriteOperationSuccess(ctx, op.ID, nil)
 	}
 	return pool, nil
 }
@@ -114,10 +114,10 @@ func (am *assetManager) ActivateTokenPool(ctx context.Context, pool *fftypes.Tok
 	}
 
 	if complete, err := plugin.ActivateTokenPool(ctx, op.ID, pool, event); err != nil {
+		am.txHelper.WriteOperationFailure(ctx, op.ID, err)
 		return err
 	} else if complete {
-		update := database.OperationQueryFactory.NewUpdate(ctx).Set("status", fftypes.OpStatusSucceeded)
-		return am.database.UpdateOperation(ctx, op.ID, update)
+		am.txHelper.WriteOperationSuccess(ctx, op.ID, nil)
 	}
 	return nil
 }

internal/assets/token_pool_test.go

@@ -210,6 +210,7 @@ func TestCreateTokenPoolFail(t *testing.T) {
 	mth.On("SubmitNewTransaction", context.Background(), "ns1", fftypes.TransactionTypeTokenPool).Return(fftypes.NewUUID(), nil)
 	mdi.On("InsertOperation", context.Background(), mock.Anything).Return(nil)
 	mti.On("CreateTokenPool", context.Background(), mock.Anything, mock.Anything, mock.Anything).Return(false, fmt.Errorf("pop"))
+	mth.On("WriteOperationFailure", context.Background(), mock.Anything, fmt.Errorf("pop"))
 
 	_, err := am.CreateTokenPool(context.Background(), "ns1", pool, false)
 	assert.Regexp(t, "pop", err)
@@ -257,7 +258,7 @@ func TestCreateTokenPoolOpInsertFail(t *testing.T) {
 	assert.Regexp(t, "pop", err)
 }
 
-func TestCreateTokenPoolOpUpdateFail(t *testing.T) {
+func TestCreateTokenPoolSyncSuccess(t *testing.T) {
 	am, cancel := newTestAssets(t)
 	defer cancel()
 
@@ -276,13 +277,13 @@ func TestCreateTokenPoolOpUpdateFail(t *testing.T) {
 	mth.On("SubmitNewTransaction", context.Background(), "ns1", fftypes.TransactionTypeTokenPool).Return(fftypes.NewUUID(), nil)
 	mdi.On("InsertOperation", context.Background(), mock.Anything).Return(nil)
 	mti.On("CreateTokenPool", context.Background(), mock.Anything, mock.Anything, mock.Anything).Return(true, nil)
-	mdi.On("UpdateOperation", context.Background(), mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	mth.On("WriteOperationSuccess", context.Background(), mock.Anything, mock.Anything)
 
 	_, err := am.CreateTokenPool(context.Background(), "ns1", pool, false)
-	assert.Regexp(t, "pop", err)
+	assert.NoError(t, err)
 }
 
-func TestCreateTokenPoolSuccess(t *testing.T) {
+func TestCreateTokenPoolAsyncSuccess(t *testing.T) {
 	am, cancel := newTestAssets(t)
 	defer cancel()
 
@@ -411,17 +412,19 @@ func TestActivateTokenPoolFail(t *testing.T) {
 	mdm := am.data.(*datamocks.Manager)
 	mdi := am.database.(*databasemocks.Plugin)
 	mti := am.tokens["magic-tokens"].(*tokenmocks.Plugin)
+	mth := am.txHelper.(*txcommonmocks.Helper)
 	mdm.On("VerifyNamespaceExists", context.Background(), "ns1").Return(nil)
 	mdi.On("InsertOperation", context.Background(), mock.MatchedBy(func(op *fftypes.Operation) bool {
 		return op.Type == fftypes.OpTypeTokenActivatePool
 	})).Return(nil)
 	mti.On("ActivateTokenPool", context.Background(), mock.Anything, pool, ev).Return(false, fmt.Errorf("pop"))
+	mth.On("WriteOperationFailure", context.Background(), mock.Anything, fmt.Errorf("pop"))
 
 	err := am.ActivateTokenPool(context.Background(), pool, ev)
 	assert.EqualError(t, err, "pop")
 }
 
-func TestActivateTokenPoolOpUpdateFail(t *testing.T) {
+func TestActivateTokenPoolSyncSuccess(t *testing.T) {
 	am, cancel := newTestAssets(t)
 	defer cancel()
 
@@ -434,15 +437,16 @@ func TestActivateTokenPoolOpUpdateFail(t *testing.T) {
 	mdm := am.data.(*datamocks.Manager)
 	mdi := am.database.(*databasemocks.Plugin)
 	mti := am.tokens["magic-tokens"].(*tokenmocks.Plugin)
+	mth := am.txHelper.(*txcommonmocks.Helper)
 	mdm.On("VerifyNamespaceExists", context.Background(), "ns1").Return(nil)
 	mdi.On("InsertOperation", context.Background(), mock.MatchedBy(func(op *fftypes.Operation) bool {
 		return op.Type == fftypes.OpTypeTokenActivatePool
 	})).Return(nil)
 	mti.On("ActivateTokenPool", context.Background(), mock.Anything, pool, ev).Return(true, nil)
-	mdi.On("UpdateOperation", context.Background(), mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+	mth.On("WriteOperationSuccess", context.Background(), mock.Anything, mock.Anything)
 
 	err := am.ActivateTokenPool(context.Background(), pool, ev)
-	assert.EqualError(t, err, "pop")
+	assert.NoError(t, err)
 }
 
 func TestGetTokenPool(t *testing.T) {

internal/assets/token_transfer.go

@@ -21,7 +21,6 @@ import (
 	"fmt"
 
 	"github.com/hyperledger/firefly/internal/i18n"
-	"github.com/hyperledger/firefly/internal/log"
 	"github.com/hyperledger/firefly/internal/sysmessaging"
 	"github.com/hyperledger/firefly/internal/txcommon"
 	"github.com/hyperledger/firefly/pkg/database"
@@ -272,20 +271,9 @@ func (s *transferSender) sendInternal(ctx context.Context, method sendMethod) er
 		panic(fmt.Sprintf("unknown transfer type: %v", s.transfer.Type))
 	}
 
-	// if transaction fails,  mark op as failed in DB
 	if err != nil {
-		_ = s.mgr.database.RunAsGroup(ctx, func(ctx context.Context) (err error) {
-			l := log.L(ctx)
-			update := database.OperationQueryFactory.NewUpdate(ctx).
-				Set("status", fftypes.OpStatusFailed)
-			if err = s.mgr.database.UpdateOperation(ctx, op.ID, update); err != nil {
-				l.Errorf("Operation update failed: %s update=[ %s ]", err, update)
-			}
-
-			return nil
-		})
+		s.mgr.txHelper.WriteOperationFailure(ctx, op.ID, err)
 	}
-
 	return err
 }
 

internal/assets/token_transfer_test.go

@@ -411,7 +411,7 @@ func TestMintTokensFail(t *testing.T) {
 	mti.On("MintTokens", context.Background(), mock.Anything, "F1", &mint.TokenTransfer).Return(fmt.Errorf("pop"))
 	mth.On("SubmitNewTransaction", context.Background(), "ns1", fftypes.TransactionTypeTokenTransfer).Return(fftypes.NewUUID(), nil)
 	mdi.On("InsertOperation", context.Background(), mock.Anything).Return(nil)
-	mdi.On("UpdateOperation", context.Background(), mock.Anything, mock.Anything).Return(nil)
+	mth.On("WriteOperationFailure", context.Background(), mock.Anything, fmt.Errorf("pop"))
 
 	_, err := am.MintTokens(context.Background(), "ns1", mint, false)
 	assert.EqualError(t, err, "pop")
@@ -441,7 +441,7 @@ func TestMintTokensFailAndDbFail(t *testing.T) {
 	mti.On("MintTokens", context.Background(), mock.Anything, "F1", &mint.TokenTransfer).Return(fmt.Errorf("pop"))
 	mdi.On("InsertOperation", context.Background(), mock.Anything).Return(nil)
 	mth.On("SubmitNewTransaction", context.Background(), "ns1", fftypes.TransactionTypeTokenTransfer).Return(fftypes.NewUUID(), nil)
-	mdi.On("UpdateOperation", context.Background(), mock.Anything, mock.Anything).Return(fmt.Errorf("Update fail"))
+	mth.On("WriteOperationFailure", context.Background(), mock.Anything, fmt.Errorf("pop"))
 
 	_, err := am.MintTokens(context.Background(), "ns1", mint, false)
 	assert.EqualError(t, err, "pop")

internal/contracts/manager.go

@@ -163,6 +163,21 @@ func (cm *contractManager) GetFFIs(ctx context.Context, ns string, filter databa
 	return cm.database.GetFFIs(ctx, ns, filter)
 }
 
+func (cm *contractManager) writeInvokeTransaction(ctx context.Context, ns string, input fftypes.JSONObject) (*fftypes.Operation, error) {
+	txid, err := cm.txHelper.SubmitNewTransaction(ctx, ns, fftypes.TransactionTypeContractInvoke)
+	if err != nil {
+		return nil, err
+	}
+
+	op := fftypes.NewOperation(
+		cm.blockchain,
+		ns,
+		txid,
+		fftypes.OpTypeBlockchainInvoke)
+	op.Input = input
+	return op, cm.database.InsertOperation(ctx, op)
+}
+
 func (cm *contractManager) InvokeContract(ctx context.Context, ns string, req *fftypes.ContractCallRequest) (res interface{}, err error) {
 	req.Key, err = cm.identity.ResolveSigningKey(ctx, req.Key)
 	if err != nil {
@@ -177,33 +192,31 @@ func (cm *contractManager) InvokeContract(ctx context.Context, ns string, req *f
 		if err := cm.validateInvokeContractRequest(ctx, req); err != nil {
 			return err
 		}
-
-		txid, err := cm.txHelper.SubmitNewTransaction(ctx, ns, fftypes.TransactionTypeContractInvoke)
-		if err != nil {
-			return err
+		if req.Type == fftypes.CallTypeInvoke {
+			op, err = cm.writeInvokeTransaction(ctx, ns, req.Input)
+			if err != nil {
+				return err
+			}
 		}
-
-		op = fftypes.NewOperation(
-			cm.blockchain,
-			ns,
-			txid,
-			fftypes.OpTypeBlockchainInvoke)
-		op.Input = req.Input
-		return cm.database.InsertOperation(ctx, op)
+		return nil
 	})
 	if err != nil {
 		return nil, err
 	}
 
 	switch req.Type {
 	case fftypes.CallTypeInvoke:
-		operationID := fftypes.NewUUID()
-		return cm.blockchain.InvokeContract(ctx, operationID, req.Key, req.Location, req.Method, req.Input)
+		res, err = cm.blockchain.InvokeContract(ctx, op.ID, req.Key, req.Location, req.Method, req.Input)
 	case fftypes.CallTypeQuery:
-		return cm.blockchain.QueryContract(ctx, req.Location, req.Method, req.Input)
+		res, err = cm.blockchain.QueryContract(ctx, req.Location, req.Method, req.Input)
 	default:
 		panic(fmt.Sprintf("unknown call type: %s", req.Type))
 	}
+
+	if op != nil && err != nil {
+		cm.txHelper.WriteOperationFailure(ctx, op.ID, err)
+	}
+	return res, err
 }
 
 func (cm *contractManager) InvokeContractAPI(ctx context.Context, ns, apiName, methodPath string, req *fftypes.ContractCallRequest) (interface{}, error) {

internal/contracts/manager_test.go

@@ -1004,6 +1004,41 @@ func TestInvokeContract(t *testing.T) {
 	mth.AssertExpectations(t)
 }
 
+func TestInvokeContractFail(t *testing.T) {
+	cm := newTestContractManager()
+	mbi := cm.blockchain.(*blockchainmocks.Plugin)
+	mim := cm.identity.(*identitymanagermocks.Manager)
+	mdi := cm.database.(*databasemocks.Plugin)
+	mth := cm.txHelper.(*txcommonmocks.Helper)
+
+	req := &fftypes.ContractCallRequest{
+		Type:      fftypes.CallTypeInvoke,
+		Interface: fftypes.NewUUID(),
+		Ledger:    fftypes.JSONAnyPtr(""),
+		Location:  fftypes.JSONAnyPtr(""),
+		Method: &fftypes.FFIMethod{
+			Name:    "doStuff",
+			ID:      fftypes.NewUUID(),
+			Params:  fftypes.FFIParams{},
+			Returns: fftypes.FFIParams{},
+		},
+	}
+
+	mth.On("SubmitNewTransaction", mock.Anything, "ns1", fftypes.TransactionTypeContractInvoke).Return(fftypes.NewUUID(), nil)
+
+	mim.On("ResolveSigningKey", mock.Anything, "").Return("key-resolved", nil)
+	mdi.On("InsertOperation", mock.Anything, mock.MatchedBy(func(op *fftypes.Operation) bool {
+		return op.Namespace == "ns1" && op.Type == fftypes.OpTypeBlockchainInvoke && op.Plugin == "mockblockchain"
+	})).Return(nil)
+	mbi.On("InvokeContract", mock.Anything, mock.AnythingOfType("*fftypes.UUID"), "key-resolved", req.Location, req.Method, req.Input).Return(nil, fmt.Errorf("pop"))
+	mth.On("WriteOperationFailure", mock.Anything, mock.Anything, fmt.Errorf("pop"))
+
+	_, err := cm.InvokeContract(context.Background(), "ns1", req)
+
+	assert.EqualError(t, err, "pop")
+	mth.AssertExpectations(t)
+}
+
 func TestInvokeContractFailResolveSigningKey(t *testing.T) {
 	cm := newTestContractManager()
 	mim := cm.identity.(*identitymanagermocks.Manager)

internal/database/sqlcommon/operation_sql.go

@@ -152,7 +152,7 @@ func (s *SQLCommon) GetOperations(ctx context.Context, filter database.Filter) (
 	return ops, s.queryRes(ctx, tx, "operations", fop, fi), err
 }
 
-func (s *SQLCommon) UpdateOperation(ctx context.Context, id *fftypes.UUID, update database.Update) (err error) {
+func (s *SQLCommon) updateOperation(ctx context.Context, id *fftypes.UUID, update database.Update) (err error) {
 
 	ctx, tx, autoCommit, err := s.beginOrUseTx(ctx)
 	if err != nil {
@@ -174,3 +174,13 @@ func (s *SQLCommon) UpdateOperation(ctx context.Context, id *fftypes.UUID, updat
 
 	return s.commitTx(ctx, tx, autoCommit)
 }
+
+func (s *SQLCommon) ResolveOperation(ctx context.Context, id *fftypes.UUID, status fftypes.OpStatus, errorMsg string, output fftypes.JSONObject) (err error) {
+	update := database.OperationQueryFactory.NewUpdate(ctx).
+		Set("status", status).
+		Set("error", errorMsg)
+	if output != nil {
+		update.Set("output", output)
+	}
+	return s.updateOperation(ctx, id, update)
+}

internal/database/sqlcommon/operation_sql_test.go

@@ -90,12 +90,7 @@ func TestOperationE2EWithDB(t *testing.T) {
 	assert.Equal(t, 0, len(operations))
 
 	// Update
-	updateTime := fftypes.Now()
-	up := database.OperationQueryFactory.NewUpdate(ctx).
-		Set("status", fftypes.OpStatusSucceeded).
-		Set("updated", updateTime).
-		Set("error", "")
-	err = s.UpdateOperation(ctx, operation.ID, up)
+	err = s.ResolveOperation(ctx, operation.ID, fftypes.OpStatusSucceeded, "", fftypes.JSONObject{"extra": "info"})
 	assert.NoError(t, err)
 
 	// Test find updated value
@@ -198,15 +193,15 @@ func TestOperationUpdateBeginFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin().WillReturnError(fmt.Errorf("pop"))
 	u := database.OperationQueryFactory.NewUpdate(context.Background()).Set("id", fftypes.NewUUID())
-	err := s.UpdateOperation(context.Background(), fftypes.NewUUID(), u)
+	err := s.updateOperation(context.Background(), fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10114", err)
 }
 
 func TestOperationUpdateBuildQueryFail(t *testing.T) {
 	s, mock := newMockProvider().init()
 	mock.ExpectBegin()
 	u := database.OperationQueryFactory.NewUpdate(context.Background()).Set("id", map[bool]bool{true: false})
-	err := s.UpdateOperation(context.Background(), fftypes.NewUUID(), u)
+	err := s.updateOperation(context.Background(), fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10149.*id", err)
 }
 
@@ -216,6 +211,6 @@ func TestOperationUpdateFail(t *testing.T) {
 	mock.ExpectExec("UPDATE .*").WillReturnError(fmt.Errorf("pop"))
 	mock.ExpectRollback()
 	u := database.OperationQueryFactory.NewUpdate(context.Background()).Set("id", fftypes.NewUUID())
-	err := s.UpdateOperation(context.Background(), fftypes.NewUUID(), u)
+	err := s.updateOperation(context.Background(), fftypes.NewUUID(), u)
 	assert.Regexp(t, "FF10117", err)
 }

internal/events/dx_callbacks.go

@@ -269,11 +269,9 @@ func (em *eventManager) TransferResult(dx dataexchange.Plugin, trackingID string
 			}
 		}
 
-		update := database.OperationQueryFactory.NewUpdate(em.ctx).
-			Set("status", status).
-			Set("error", update.Error).
-			Set("output", update.Info) // Note that we don't need the manifest to be kept here, as it's already in the input
-		if err := em.database.UpdateOperation(em.ctx, op.ID, update); err != nil {
+		// Resolve the operation
+		// Note that we don't need the manifest to be kept here, as it's already in the input
+		if err := em.database.ResolveOperation(em.ctx, op.ID, status, update.Error, update.Info); err != nil {
 			return true, err // this is always retryable
 		}
 		return false, nil

internal/events/dx_callbacks_test.go

@@ -19,6 +19,7 @@ package events
 import (
 	"encoding/json"
 	"fmt"
+	"strings"
 	"testing"
 
 	"github.com/hyperledger/firefly/mocks/databasemocks"
@@ -492,7 +493,9 @@ func TestTransferResultOk(t *testing.T) {
 			ID: id,
 		},
 	}, nil, nil)
-	mdi.On("UpdateOperation", mock.Anything, id, mock.Anything).Return(nil)
+	mdi.On("ResolveOperation", mock.Anything, id, fftypes.OpStatusFailed, "error info", fftypes.JSONObject{
+		"extra": "info",
+	}).Return(nil)
 
 	mdx := &dataexchangemocks.Plugin{}
 	mdx.On("Name").Return("utdx")
@@ -501,7 +504,6 @@ func TestTransferResultOk(t *testing.T) {
 		Info:  fftypes.JSONObject{"extra": "info"},
 	})
 	assert.NoError(t, err)
-
 }
 
 func TestTransferResultManifestMismatch(t *testing.T) {
@@ -518,7 +520,11 @@ func TestTransferResultManifestMismatch(t *testing.T) {
 			},
 		},
 	}, nil, nil)
-	mdi.On("UpdateOperation", mock.Anything, id, mock.Anything).Return(nil)
+	mdi.On("ResolveOperation", mock.Anything, id, fftypes.OpStatusFailed, mock.MatchedBy(func(errorMsg string) bool {
+		return strings.Contains(errorMsg, "FF10329")
+	}), fftypes.JSONObject{
+		"extra": "info",
+	}).Return(nil)
 
 	mdx := &dataexchangemocks.Plugin{}
 	mdx.On("Name").Return("utdx")
@@ -538,9 +544,7 @@ func TestTransferResultNotCorrelated(t *testing.T) {
 	defer cancel()
 
 	mdi := em.database.(*databasemocks.Plugin)
-	id := fftypes.NewUUID()
 	mdi.On("GetOperations", mock.Anything, mock.Anything).Return([]*fftypes.Operation{}, nil, nil)
-	mdi.On("UpdateOperation", mock.Anything, id, mock.Anything).Return(nil)
 
 	mdx := &dataexchangemocks.Plugin{}
 	mdx.On("Name").Return("utdx")
@@ -597,7 +601,9 @@ func TestTransferUpdateFail(t *testing.T) {
 			ID: id,
 		},
 	}, nil, nil)
-	mdi.On("UpdateOperation", mock.Anything, id, mock.Anything).Return(fmt.Errorf("pop"))
+	mdi.On("ResolveOperation", mock.Anything, id, fftypes.OpStatusFailed, "error info", fftypes.JSONObject{
+		"extra": "info",
+	}).Return(fmt.Errorf("pop"))
 
 	mdx := &dataexchangemocks.Plugin{}
 	mdx.On("Name").Return("utdx")