Repository: hyperledger/firefly
Pull request #935 (2022-08-05 20:41:03)

Fix bugs with Operations (and other minor fixes)
* Always populate Operation namespace from Managers instead of from other objects
* Actually store pending blob upload Operations
* Other misc cleanup identified while doing E2E and perf test runs


Patch:
diff --git a/internal/assets/token_pool.go b/internal/assets/token_pool.go
index 8dd4ba65b..d0d532390 100644
--- a/internal/assets/token_pool.go
+++ b/internal/assets/token_pool.go
@@ -81,7 +81,7 @@ func (am *assetManager) createTokenPoolInternal(ctx context.Context, pool *core.
 
 		op = core.NewOperation(
 			plugin,
-			pool.Namespace,
+			am.namespace,
 			txid,
 			core.OpTypeTokenCreatePool)
 		if err = txcommon.AddTokenPoolCreateInputs(op, pool); err == nil {
@@ -119,7 +119,7 @@ func (am *assetManager) ActivateTokenPool(ctx context.Context, pool *core.TokenP
 
 		op = core.NewOperation(
 			plugin,
-			pool.Namespace,
+			am.namespace,
 			pool.TX.ID,
 			core.OpTypeTokenActivatePool)
 		txcommon.AddTokenPoolActivateInputs(op, pool.ID)
diff --git a/internal/blockchain/ethereum/config.go b/internal/blockchain/ethereum/config.go
index edb11479d..7e6289ca5 100644
--- a/internal/blockchain/ethereum/config.go
+++ b/internal/blockchain/ethereum/config.go
@@ -54,13 +54,6 @@ const (
 	// EthconnectConfigFromBlockDeprecated is the configuration of the first block to listen to when creating the listener for the FireFly contract
 	EthconnectConfigFromBlockDeprecated = "fromBlock"
 
-	// FireFlyContractConfigKey is a sub-key in the config to contain the info on the deployed FireFly contract
-	FireFlyContractConfigKey = "fireflyContract"
-	// FireFlyContractAddress is the ethereum address of the FireFly contract
-	FireFlyContractAddress = "address"
-	// FireFlyContractFromBlock is the configuration of the first block to listen to when creating the listener
-	FireFlyContractFromBlock = "fromBlock"
-
 	// AddressResolverConfigKey is a sub-key in the config to contain an address resolver config.
 	AddressResolverConfigKey = "addressResolver"
 	// AddressResolverRetainOriginal when true the original pre-resolved string is retained after the lookup, and passed down to Ethconnect as the from address
diff --git a/internal/blockchain/fabric/config.go b/internal/blockchain/fabric/config.go
index 9cb24a47d..cf1d4588d 100644
--- a/internal/blockchain/fabric/config.go
+++ b/internal/blockchain/fabric/config.go
@@ -49,13 +49,6 @@ const (
 	FabconnectPrefixLong = "prefixLong"
 	// FabconnectConfigChaincodeDeprecated is the Fabric Firefly chaincode deployed to the Firefly channels
 	FabconnectConfigChaincodeDeprecated = "chaincode"
-
-	// FireFlyContractConfigKey is a sub-key in the config to contain the info on the deployed FireFly contract
-	FireFlyContractConfigKey = "fireflyContract"
-	// FireFlyContractChaincode is the Fabric Firefly chaincode deployed to the Firefly channels
-	FireFlyContractChaincode = "chaincode"
-	// FireFlyContractFromBlock is the configuration of the first block to listen to when creating the listener
-	FireFlyContractFromBlock = "fromBlock"
 )
 
 func (f *Fabric) InitConfig(config config.Section) {
diff --git a/internal/broadcast/manager.go b/internal/broadcast/manager.go
index 46be34e9e..cb248d19e 100644
--- a/internal/broadcast/manager.go
+++ b/internal/broadcast/manager.go
@@ -129,7 +129,7 @@ func (bm *broadcastManager) dispatchBatch(ctx context.Context, state *batch.Disp
 	// Upload the batch itself
 	op := core.NewOperation(
 		bm.sharedstorage,
-		state.Persisted.Namespace,
+		bm.namespace.LocalName,
 		state.Persisted.TX.ID,
 		core.OpTypeSharedStorageUploadBatch)
 	addUploadBatchInputs(op, state.Persisted.ID)
@@ -157,10 +157,13 @@ func (bm *broadcastManager) uploadBlobs(ctx context.Context, tx *fftypes.UUID, d
 
 			op := core.NewOperation(
 				bm.sharedstorage,
-				d.Namespace,
+				bm.namespace.LocalName,
 				tx,
 				core.OpTypeSharedStorageUploadBlob)
 			addUploadBlobInputs(op, d.ID)
+			if err := bm.operations.AddOrReuseOperation(ctx, op); err != nil {
+				return err
+			}
 
 			blob, err := bm.database.GetBlobMatchingHash(ctx, d.Blob.Hash)
 			if err != nil {
diff --git a/internal/coremsgs/en_struct_descriptions.go b/internal/coremsgs/en_struct_descriptions.go
index 1921479a5..1f9f94add 100644
--- a/internal/coremsgs/en_struct_descriptions.go
+++ b/internal/coremsgs/en_struct_descriptions.go
@@ -376,9 +376,12 @@ var (
 	VerifierCreated   = ffm("Verifier.created", "The time this verifier was created on this node")
 
 	// Namespace field descriptions
+	NamespaceID                    = ffm("Namespace.id", "The UUID of the namespace. For locally established namespaces will be different on each node in the network. For broadcast namespaces, will be the same on every node")
+	NamespaceMessage               = ffm("Namespace.message", "The UUID of broadcast message used to establish the namespace. Unset for local namespaces")
 	NamespaceName                  = ffm("Namespace.name", "The local namespace name")
 	NamespaceRemoteName            = ffm("Namespace.remoteName", "The namespace name within the multiparty network")
 	NamespaceDescription           = ffm("Namespace.description", "A description of the namespace")
+	NamespaceType                  = ffm("Namespace.type", "The type of the namespace")
 	NamespaceCreated               = ffm("Namespace.created", "The time the namespace was created")
 	MultipartyContractsActive      = ffm("MultipartyContracts.active", "The currently active FireFly smart contract")
 	MultipartyContractsTerminated  = ffm("MultipartyContracts.terminated", "Previously-terminated FireFly smart contracts")
diff --git a/internal/multiparty/manager.go b/internal/multiparty/manager.go
index 7a03e677b..7d5c860f6 100644
--- a/internal/multiparty/manager.go
+++ b/internal/multiparty/manager.go
@@ -230,7 +230,7 @@ func (mm *multipartyManager) SubmitNetworkAction(ctx context.Context, signingKey
 func (mm *multipartyManager) SubmitBatchPin(ctx context.Context, batch *core.BatchPersisted, contexts []*fftypes.Bytes32, payloadRef string) error {
 	op := core.NewOperation(
 		mm.blockchain,
-		batch.Namespace,
+		mm.namespace.LocalName,
 		batch.TX.ID,
 		core.OpTypeBlockchainPinBatch)
 	addBatchPinInputs(op, batch.ID, contexts, payloadRef)
diff --git a/internal/privatemessaging/privatemessaging.go b/internal/privatemessaging/privatemessaging.go
index 3e3cff64e..b4b57ddbe 100644
--- a/internal/privatemessaging/privatemessaging.go
+++ b/internal/privatemessaging/privatemessaging.go
@@ -215,7 +215,7 @@ func (pm *privateMessaging) prepareBlobTransfers(ctx context.Context, data core.
 
 				op := core.NewOperation(
 					pm.exchange,
-					d.Namespace,
+					pm.namespace.LocalName,
 					txid,
 					core.OpTypeDataExchangeSendBlob)
 				addTransferBlobInputs(op, node.ID, blob.Hash)
@@ -291,7 +291,7 @@ func (pm *privateMessaging) sendData(ctx context.Context, tw *core.TransportWrap
 
 			op := core.NewOperation(
 				pm.exchange,
-				batch.Namespace,
+				pm.namespace.LocalName,
 				batch.Payload.TX.ID,
 				core.OpTypeDataExchangeSendBatch)
 			addBatchSendInputs(op, node.ID, batch.Group, batch.ID)
diff --git a/pkg/core/namespace.go b/pkg/core/namespace.go
index fa8efb3f8..c96276434 100644
--- a/pkg/core/namespace.go
+++ b/pkg/core/namespace.go
@@ -35,7 +35,6 @@ type Namespace struct {
 	Contracts   *MultipartyContracts `ffstruct:"Namespace" json:"-"`
 }
 
-// MultipartyContracts represent the currently active and any terminated FireFly multiparty contract(s)
 type MultipartyContracts struct {
 	Active     *MultipartyContract   `ffstruct:"MultipartyContracts" json:"active"`
 	Terminated []*MultipartyContract `ffstruct:"MultipartyContracts" json:"terminated,omitempty"`
diff --git a/pkg/core/namespace_status.go b/pkg/core/namespace_status.go
index 59f359c54..7259ea45d 100644
--- a/pkg/core/namespace_status.go
+++ b/pkg/core/namespace_status.go
@@ -60,7 +60,6 @@ type NamespaceStatusPlugin struct {
 	PluginType string `ffstruct:"NamespaceStatusPlugin" json:"pluginType"`
 }
 
-// NamespaceStatusMultiparty is information about multiparty mode and any associated multiparty contracts
 type NamespaceStatusMultiparty struct {
 	Enabled   bool                 `ffstruct:"NamespaceStatusMultiparty" json:"enabled"`
 	Contracts *MultipartyContracts `ffstruct:"NamespaceStatusMultiparty" json:"contract,omitempty"`


Test patch:
diff --git a/internal/blockchain/ethereum/ethereum_test.go b/internal/blockchain/ethereum/ethereum_test.go
index 0aa7292fd..fa0398fbf 100644
--- a/internal/blockchain/ethereum/ethereum_test.go
+++ b/internal/blockchain/ethereum/ethereum_test.go
@@ -326,7 +326,6 @@ func TestInitNewConfig(t *testing.T) {
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
-	utConfig.AddKnownKey(FireFlyContractConfigKey+".0."+FireFlyContractAddress, "0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.Equal(t, 2, httpmock.GetTotalCallCount())
@@ -351,7 +350,6 @@ func TestInitNewConfigBadIndex(t *testing.T) {
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
-	utConfig.AddKnownKey(FireFlyContractConfigKey+".0."+FireFlyContractAddress, "")
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.NoError(t, err)
@@ -381,7 +379,6 @@ func TestInitNetworkVersionNotFound(t *testing.T) {
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
-	utConfig.AddKnownKey(FireFlyContractConfigKey+".0."+FireFlyContractAddress, "0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.NoError(t, err)
@@ -2916,7 +2913,6 @@ func TestAddAndRemoveFireflySubscription(t *testing.T) {
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
-	utConfig.AddKnownKey(FireFlyContractConfigKey+".0."+FireFlyContractAddress, "0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.NoError(t, err)
@@ -2959,7 +2955,6 @@ func TestAddFireflySubscriptionV1(t *testing.T) {
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
-	utConfig.AddKnownKey(FireFlyContractConfigKey+".0."+FireFlyContractAddress, "0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.NoError(t, err)
@@ -2997,7 +2992,6 @@ func TestAddFireflySubscriptionQuerySubsFail(t *testing.T) {
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
-	utConfig.AddKnownKey(FireFlyContractConfigKey+".0."+FireFlyContractAddress, "0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.NoError(t, err)
@@ -3035,7 +3029,6 @@ func TestAddFireflySubscriptionCreateError(t *testing.T) {
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
-	utConfig.AddKnownKey(FireFlyContractConfigKey+".0."+FireFlyContractAddress, "0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.NoError(t, err)
@@ -3072,7 +3065,6 @@ func TestAddFireflySubscriptionGetVersionError(t *testing.T) {
 	utEthconnectConf.Set(ffresty.HTTPConfigURL, "http://localhost:12345")
 	utEthconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utEthconnectConf.Set(EthconnectConfigTopic, "topic1")
-	utConfig.AddKnownKey(FireFlyContractConfigKey+".0."+FireFlyContractAddress, "0x71C7656EC7ab88b098defB751B7401B5f6d8976F")
 
 	err := e.Init(e.ctx, utConfig, e.metrics)
 	assert.NoError(t, err)
diff --git a/internal/blockchain/fabric/fabric_test.go b/internal/blockchain/fabric/fabric_test.go
index fe038adbc..8e2f373de 100644
--- a/internal/blockchain/fabric/fabric_test.go
+++ b/internal/blockchain/fabric/fabric_test.go
@@ -469,7 +469,6 @@ func TestInitNewConfig(t *testing.T) {
 	utFabconnectConf.Set(ffresty.HTTPCustomClient, mockedClient)
 	utFabconnectConf.Set(FabconnectConfigSigner, "signer001")
 	utFabconnectConf.Set(FabconnectConfigTopic, "topic1")
-	utConfig.AddKnownKey(FireFlyContractConfigKey+".0."+FireFlyContractChaincode, "firefly")
 
 	err := e.Init(e.ctx, utConfig, &metricsmocks.Manager{})
 	assert.NoError(t, err)
diff --git a/internal/broadcast/manager_test.go b/internal/broadcast/manager_test.go
index f879f8c89..45ab6144f 100644
--- a/internal/broadcast/manager_test.go
+++ b/internal/broadcast/manager_test.go
@@ -168,7 +168,7 @@ func TestBroadcastMessageBad(t *testing.T) {
 
 }
 
-func TestDispatchBatchBlobsFaill(t *testing.T) {
+func TestDispatchBatchBlobsFail(t *testing.T) {
 	bm, cancel := newTestBroadcast(t)
 	defer cancel()
 
@@ -182,6 +182,9 @@ func TestDispatchBatchBlobsFaill(t *testing.T) {
 		Pins: []*fftypes.Bytes32{fftypes.NewRandB32()},
 	}
 
+	mom := bm.operations.(*operationmocks.Manager)
+	mom.On("AddOrReuseOperation", mock.Anything, mock.Anything).Return(nil)
+
 	mdi := bm.database.(*databasemocks.Plugin)
 	mdi.On("GetBlobMatchingHash", bm.ctx, blobHash).Return(nil, fmt.Errorf("pop"))
 
@@ -189,6 +192,7 @@ func TestDispatchBatchBlobsFaill(t *testing.T) {
 	assert.EqualError(t, err, "pop")
 
 	mdi.AssertExpectations(t)
+	mom.AssertExpectations(t)
 }
 
 func TestDispatchBatchInsertOpFail(t *testing.T) {
@@ -313,6 +317,7 @@ func TestUploadBlobsPublishFail(t *testing.T) {
 
 	ctx := context.Background()
 	mdi.On("GetBlobMatchingHash", ctx, blob.Hash).Return(blob, nil)
+	mom.On("AddOrReuseOperation", mock.Anything, mock.Anything).Return(nil)
 	mom.On("RunOperation", mock.Anything, mock.MatchedBy(func(op *core.PreparedOperation) bool {
 		data := op.Data.(uploadBlobData)
 		return op.Type == core.OpTypeSharedStorageUploadBlob && data.Blob == blob
@@ -348,6 +353,9 @@ func TestUploadBlobsGetBlobFail(t *testing.T) {
 	ctx := context.Background()
 	mdi.On("GetBlobMatchingHash", ctx, blob.Hash).Return(nil, fmt.Errorf("pop"))
 
+	mom := bm.operations.(*operationmocks.Manager)
+	mom.On("AddOrReuseOperation", mock.Anything, mock.Anything).Return(nil)
+
 	err := bm.uploadBlobs(ctx, fftypes.NewUUID(), core.DataArray{
 		{
 			ID: dataID,
@@ -359,6 +367,7 @@ func TestUploadBlobsGetBlobFail(t *testing.T) {
 	assert.Regexp(t, "pop", err)
 
 	mdi.AssertExpectations(t)
+	mom.AssertExpectations(t)
 
 }
 
@@ -376,6 +385,9 @@ func TestUploadBlobsGetBlobNotFound(t *testing.T) {
 	ctx := context.Background()
 	mdi.On("GetBlobMatchingHash", ctx, blob.Hash).Return(nil, nil)
 
+	mom := bm.operations.(*operationmocks.Manager)
+	mom.On("AddOrReuseOperation", mock.Anything, mock.Anything).Return(nil)
+
 	err := bm.uploadBlobs(ctx, fftypes.NewUUID(), core.DataArray{
 		{
 			ID: dataID,
@@ -387,5 +399,34 @@ func TestUploadBlobsGetBlobNotFound(t *testing.T) {
 	assert.Regexp(t, "FF10239", err)
 
 	mdi.AssertExpectations(t)
+	mom.AssertExpectations(t)
+
+}
+
+func TestUploadBlobsGetBlobInsertOpFail(t *testing.T) {
+	bm, cancel := newTestBroadcast(t)
+	defer cancel()
+
+	blob := &core.Blob{
+		Hash:       fftypes.NewRandB32(),
+		PayloadRef: "blob/1",
+	}
+	dataID := fftypes.NewUUID()
+	ctx := context.Background()
+
+	mom := bm.operations.(*operationmocks.Manager)
+	mom.On("AddOrReuseOperation", mock.Anything, mock.Anything).Return(fmt.Errorf("pop"))
+
+	err := bm.uploadBlobs(ctx, fftypes.NewUUID(), core.DataArray{
+		{
+			ID: dataID,
+			Blob: &core.BlobRef{
+				Hash: blob.Hash,
+			},
+		},
+	})
+	assert.EqualError(t, err, "pop")
+
+	mom.AssertExpectations(t)
 
 }
diff --git a/test/e2e/run.sh b/test/e2e/run.sh
index 145e46ff5..e77a2774e 100755
--- a/test/e2e/run.sh
+++ b/test/e2e/run.sh
@@ -7,7 +7,6 @@ set -o pipefail
 CWD=$(dirname "$0")
 CLI="ff -v --ansi never"
 CLI_VERSION=$(cat $CWD/../../manifest.json | jq -r .cli.tag)
-STACKS_DIR=~/.firefly/stacks
 
 create_accounts() {
   if [ "$TEST_SUITE" == "TestEthereumMultipartyE2ESuite" ]; then
@@ -36,46 +35,21 @@ checkOk() {
   mkdir -p "${WORKDIR}/containerlogs"
   $CLI logs $STACK_NAME > "${WORKDIR}/containerlogs/logs.txt"
   if [ $rc -eq 0 ]; then rc=$?; fi
-
   if [ $rc -ne 0 ]; then exit $rc; fi
 }
 
-if [ -z "${STACK_NAME}" ]; then
-  STACK_NAME=firefly_e2e
-fi
-
-if [ -z "${DOWNLOAD_CLI}" ]; then
-  DOWNLOAD_CLI=true
-fi
+STACK_NAME=${STACK_NAME:-firefly_e2e}
+STACKS_DIR=${STACKS_DIR:-~/.firefly/stacks}
+STACK_DIR=${STACK_DIR:-$STACKS_DIR/$STACK_NAME}
 
-if [ -z "${CREATE_STACK}" ]; then
-  CREATE_STACK=true
-fi
+DOWNLOAD_CLI=${DOWNLOAD_CLI:-true}
+CREATE_STACK=${CREATE_STACK:-true}
+BUILD_FIREFLY=${BUILD_FIREFLY:-true}
+MULTIPARTY_ENABLED=${MULTIPARTY_ENABLED:-true}
 
-if [ -z "${BUILD_FIREFLY}" ]; then
-  BUILD_FIREFLY=true
-fi
-
-if [ -z "${DATABASE_TYPE}" ]; then
-  # Can also set to "postgres"
-  DATABASE_TYPE=sqlite3
-fi
-
-if [ -z "${MULTIPARTY_ENABLED}" ]; then
-  MULTIPARTY_ENABLED=true
-fi
-
-if [ -z "${STACK_DIR}" ]; then
-  STACK_DIR=$STACKS_DIR/$STACK_NAME
-fi
-
-if [ -z "${BLOCKCHAIN_PROVIDER}" ]; then
-  BLOCKCHAIN_PROVIDER=geth
-fi
-
-if [ -z "${TOKENS_PROVIDER}" ]; then
-  TOKENS_PROVIDER=erc20_erc721
-fi
+DATABASE_TYPE=${DATABASE_TYPE:-sqlite3}
+BLOCKCHAIN_PROVIDER=${BLOCKCHAIN_PROVIDER:-geth}
+TOKENS_PROVIDER=${TOKENS_PROVIDER:-erc20_erc721}
 
 if [ -z "${TEST_SUITE}" ]; then
   if [ "${BLOCKCHAIN_PROVIDER}" == "fabric" ]; then
@@ -95,10 +69,6 @@ fi
 
 cd $CWD
 
-if [ "$CREATE_STACK" == "true" ]; then
-  $CLI remove -f $STACK_NAME
-fi
-
 if [ "$BUILD_FIREFLY" == "true" ]; then
   make -C ../.. docker
   checkOk $?
@@ -110,6 +80,7 @@ if [ "$DOWNLOAD_CLI" == "true" ]; then
 fi
 
 if [ "$CREATE_STACK" == "true" ]; then
+  $CLI remove -f $STACK_NAME
   $CLI init --prometheus-enabled --database $DATABASE_TYPE $STACK_NAME 2 --blockchain-provider $BLOCKCHAIN_PROVIDER --token-providers $TOKENS_PROVIDER --manifest ../../manifest.json $EXTRA_INIT_ARGS --sandbox-enabled=false --multiparty=$MULTIPARTY_ENABLED
   checkOk $?