Repository: hyperledger/firefly
Pull request #1338 (2023-06-07 18:14:04)

v1.2: Backport fixes
Backport of the following fixes:

https://github.com/hyperledger/firefly/pull/1249
https://github.com/hyperledger/firefly/pull/1275
https://github.com/hyperledger/firefly/pull/1313
https://github.com/hyperledger/firefly/pull/1316


Patch:
diff --git a/internal/apiserver/route_post_new_contract_api.go b/internal/apiserver/route_post_new_contract_api.go
index f5bf7a9f2..21a6c3526 100644
--- a/internal/apiserver/route_post_new_contract_api.go
+++ b/internal/apiserver/route_post_new_contract_api.go
@@ -1,4 +1,4 @@
-// Copyright © 2022 Kaleido, Inc.
+// Copyright © 2023 Kaleido, Inc.
 //
 // SPDX-License-Identifier: Apache-2.0
 //
@@ -46,6 +46,7 @@ var postNewContractAPI = &ffapi.Route{
 			waitConfirm := strings.EqualFold(r.QP["confirm"], "true")
 			r.SuccessStatus = syncRetcode(waitConfirm)
 			api := r.Input.(*core.ContractAPI)
+			api.ID = nil
 			err = cr.or.DefinitionSender().DefineContractAPI(cr.ctx, cr.apiBaseURL, api, waitConfirm)
 			return api, err
 		},
diff --git a/internal/batch/batch_manager.go b/internal/batch/batch_manager.go
index de561ff15..81d62e809 100644
--- a/internal/batch/batch_manager.go
+++ b/internal/batch/batch_manager.go
@@ -134,8 +134,8 @@ type dispatcher struct {
 	options    DispatcherOptions
 }
 
-func (bm *batchManager) getProcessorKey(identity *core.SignerRef, groupID *fftypes.Bytes32) string {
-	return fmt.Sprintf("%s|%s|%v", identity.Author, identity.Key, groupID)
+func (bm *batchManager) getProcessorKey(author string, groupID *fftypes.Bytes32) string {
+	return fmt.Sprintf("%s|%v", author, groupID)
 }
 
 func (bm *batchManager) getDispatcherKey(txType core.TransactionType, msgType core.MessageType) string {
@@ -169,7 +169,7 @@ func (bm *batchManager) NewMessages() chan<- int64 {
 	return bm.newMessages
 }
 
-func (bm *batchManager) getProcessor(txType core.TransactionType, msgType core.MessageType, group *fftypes.Bytes32, signer *core.SignerRef) (*batchProcessor, error) {
+func (bm *batchManager) getProcessor(txType core.TransactionType, msgType core.MessageType, group *fftypes.Bytes32, author string) (*batchProcessor, error) {
 	bm.dispatcherMux.Lock()
 	defer bm.dispatcherMux.Unlock()
 
@@ -178,7 +178,7 @@ func (bm *batchManager) getProcessor(txType core.TransactionType, msgType core.M
 	if !ok {
 		return nil, i18n.NewError(bm.ctx, coremsgs.MsgUnregisteredBatchType, dispatcherKey)
 	}
-	name := bm.getProcessorKey(signer, group)
+	name := bm.getProcessorKey(author, group)
 	processor, ok := dispatcher.processors[name]
 	if !ok {
 		processor = newBatchProcessor(
@@ -188,7 +188,7 @@ func (bm *batchManager) getProcessor(txType core.TransactionType, msgType core.M
 				name:              name,
 				txType:            txType,
 				dispatcherName:    dispatcher.name,
-				signer:            *signer,
+				author:            author,
 				group:             group,
 				dispatch:          dispatcher.handler,
 			},
@@ -317,7 +317,7 @@ func (bm *batchManager) messageSequencer() {
 				// the database store. Meaning we cannot rely on the sequence having been set.
 				msg.Sequence = entry.Sequence
 
-				processor, err := bm.getProcessor(msg.Header.TxType, msg.Header.Type, msg.Header.Group, &msg.Header.SignerRef)
+				processor, err := bm.getProcessor(msg.Header.TxType, msg.Header.Type, msg.Header.Group, msg.Header.SignerRef.Author)
 				if err != nil {
 					l.Errorf("Failed to dispatch message %s: %s", msg.Header.ID, err)
 					continue
diff --git a/internal/batch/batch_processor.go b/internal/batch/batch_processor.go
index c5ca52f32..ce3e354cb 100644
--- a/internal/batch/batch_processor.go
+++ b/internal/batch/batch_processor.go
@@ -47,7 +47,7 @@ type batchProcessorConf struct {
 	name           string
 	dispatcherName string
 	txType         core.TransactionType
-	signer         core.SignerRef
+	author         string
 	group          *fftypes.Bytes32
 	dispatch       DispatchHandler
 }
@@ -171,22 +171,40 @@ func (bp *batchProcessor) newAssembly(initalWork ...*batchWork) {
 func (bp *batchProcessor) addWork(newWork *batchWork) (full, overflow bool) {
 	newQueue := make([]*batchWork, 0, len(bp.assemblyQueue)+1)
 	added := false
+
+	if newWork.msg.BatchID != nil {
+		log.L(bp.ctx).Warnf("Adding message to a new batch when one was already assigned. Old batch %s is likely abandoned.", newWork.msg.BatchID)
+	}
+
+	// Check for conditions that prevent this piece of work from going into the current batch
+	// (i.e. the new work is specifically assigned a separate signing key)
+	if len(bp.assemblyQueue) > 0 {
+		full = newWork.msg.Header.Key != bp.assemblyQueue[0].msg.Header.Key
+		overflow = true
+	}
+
 	// Build the new sorted work list
-	for _, work := range bp.assemblyQueue {
-		if !added && newWork.msg.Sequence < work.msg.Sequence {
+	if full {
+		bp.assemblyQueue = append(bp.assemblyQueue, newWork)
+	} else {
+		for _, work := range bp.assemblyQueue {
+			if !added && newWork.msg.Sequence < work.msg.Sequence {
+				newQueue = append(newQueue, newWork)
+				added = true
+			}
+			newQueue = append(newQueue, work)
+		}
+		if !added {
 			newQueue = append(newQueue, newWork)
-			added = true
 		}
-		newQueue = append(newQueue, work)
-	}
-	if !added {
-		newQueue = append(newQueue, newWork)
+
+		bp.assemblyQueueBytes += newWork.estimateSize()
+		bp.assemblyQueue = newQueue
+
+		full = len(bp.assemblyQueue) >= int(bp.conf.BatchMaxSize) || bp.assemblyQueueBytes >= bp.conf.BatchMaxBytes
+		overflow = len(bp.assemblyQueue) > 1 && bp.assemblyQueueBytes > bp.conf.BatchMaxBytes
 	}
 	log.L(bp.ctx).Debugf("Added message %s sequence=%d to in-flight batch assembly %s", newWork.msg.Header.ID, newWork.msg.Sequence, bp.assemblyID)
-	bp.assemblyQueueBytes += newWork.estimateSize()
-	bp.assemblyQueue = newQueue
-	full = len(bp.assemblyQueue) >= int(bp.conf.BatchMaxSize) || (bp.assemblyQueueBytes >= bp.conf.BatchMaxBytes)
-	overflow = len(bp.assemblyQueue) > 1 && (bp.assemblyQueueBytes > bp.conf.BatchMaxBytes)
 	return full, overflow
 }
 
@@ -377,9 +395,12 @@ func (bp *batchProcessor) initFlushState(id *fftypes.UUID, flushWork []*batchWor
 				ID:        id,
 				Type:      bp.conf.DispatcherOptions.BatchType,
 				Namespace: bp.bm.namespace,
-				SignerRef: bp.conf.signer,
-				Group:     bp.conf.group,
-				Created:   fftypes.Now(),
+				SignerRef: core.SignerRef{
+					Author: bp.conf.author,
+					Key:    flushWork[0].msg.Header.Key,
+				},
+				Group:   bp.conf.group,
+				Created: fftypes.Now(),
 			},
 		},
 	}
diff --git a/internal/blockchain/ethereum/ethereum.go b/internal/blockchain/ethereum/ethereum.go
index e0fcde5d8..1b4b70800 100644
--- a/internal/blockchain/ethereum/ethereum.go
+++ b/internal/blockchain/ethereum/ethereum.go
@@ -919,7 +919,7 @@ func (e *Ethereum) queryNetworkVersion(ctx context.Context, address string) (ver
 	res, err := e.queryContractMethod(ctx, address, networkVersionMethodABI, []interface{}{}, emptyErrors, nil)
 	if err != nil || !res.IsSuccess() {
 		// "Call failed" is interpreted as "method does not exist, default to version 1"
-		if strings.Contains(err.Error(), "FFEC100148") {
+		if strings.Contains(err.Error(), "FFEC100148") || strings.Contains(err.Error(), "FF23021") {
 			return 1, nil
 		}
 		return 0, err
diff --git a/internal/definitions/sender_contracts.go b/internal/definitions/sender_contracts.go
index 667843f17..813ab57fd 100644
--- a/internal/definitions/sender_contracts.go
+++ b/internal/definitions/sender_contracts.go
@@ -55,7 +55,9 @@ func (bm *definitionSender) DefineFFI(ctx context.Context, ffi *fftypes.FFI, wai
 }
 
 func (bm *definitionSender) DefineContractAPI(ctx context.Context, httpServerURL string, api *core.ContractAPI, waitConfirm bool) error {
-	api.ID = fftypes.NewUUID()
+	if api.ID == nil {
+		api.ID = fftypes.NewUUID()
+	}
 
 	if bm.multiparty {
 		if err := bm.contracts.ResolveContractAPI(ctx, httpServerURL, api); err != nil {
diff --git a/manifest.json b/manifest.json
index 7f5f9a0c4..c521866aa 100644
--- a/manifest.json
+++ b/manifest.json
@@ -6,8 +6,8 @@
   },
   "evmconnect": {
     "image": "ghcr.io/hyperledger/firefly-evmconnect",
-    "tag": "v1.2.2",
-    "sha": "9032e45ea9cee7d881ce88eaea720f450d2ed176e96137feb495c5e7850fab45"
+    "tag": "v1.2.7",
+    "sha": "41d3aabb7d6675b34ab983341f73751b22ab2075caac2ae0cea2f5232f072632"
   },
   "fabconnect": {
     "image": "ghcr.io/hyperledger/firefly-fabconnect",
@@ -56,4 +56,4 @@
   "cli": {
     "tag": "v1.2.0"
   }
-}
\ No newline at end of file
+}


Test patch:
diff --git a/internal/batch/batch_manager_test.go b/internal/batch/batch_manager_test.go
index 407c00466..9a1028140 100644
--- a/internal/batch/batch_manager_test.go
+++ b/internal/batch/batch_manager_test.go
@@ -339,7 +339,7 @@ func TestGetInvalidBatchTypeMsg(t *testing.T) {
 	txHelper, _ := txcommon.NewTransactionHelper(ctx, "ns1", mdi, mdm, cmi)
 	bm, _ := NewBatchManager(context.Background(), "ns1", mdi, mdm, mim, txHelper)
 	defer bm.Close()
-	_, err := bm.(*batchManager).getProcessor(core.BatchTypeBroadcast, "wrong", nil, &core.SignerRef{})
+	_, err := bm.(*batchManager).getProcessor(core.BatchTypeBroadcast, "wrong", nil, "")
 	assert.Regexp(t, "FF10126", err)
 }
 
diff --git a/internal/batch/batch_processor_test.go b/internal/batch/batch_processor_test.go
index 8d692bb38..429d47058 100644
--- a/internal/batch/batch_processor_test.go
+++ b/internal/batch/batch_processor_test.go
@@ -48,7 +48,7 @@ func newTestBatchProcessor(t *testing.T, dispatch DispatchHandler) (func(), *dat
 	txHelper, _ := txcommon.NewTransactionHelper(ctx, "ns1", mdi, mdm, cmi)
 	bp := newBatchProcessor(bm, &batchProcessorConf{
 		txType:   core.TransactionTypeBatchPin,
-		signer:   core.SignerRef{Author: "did:firefly:org/abcd", Key: "0x12345"},
+		author:   "did:firefly:org/abcd",
 		dispatch: dispatch,
 		DispatcherOptions: DispatcherOptions{
 			BatchMaxSize:   10,
@@ -396,6 +396,72 @@ func TestAddWorkInSort(t *testing.T) {
 	}, bp.assemblyQueue)
 }
 
+func TestAddWorkDifferentKeys(t *testing.T) {
+	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+		return nil
+	})
+	defer cancel()
+
+	msg1 := &core.Message{
+		Sequence: 200,
+		Header: core.MessageHeader{
+			SignerRef: core.SignerRef{
+				Key: "0x1",
+			},
+		},
+	}
+	msg2 := &core.Message{
+		Sequence: 201,
+		Header: core.MessageHeader{
+			SignerRef: core.SignerRef{
+				Key: "0x2",
+			},
+		},
+	}
+	msg3 := &core.Message{
+		Sequence: 202,
+		Header: core.MessageHeader{
+			SignerRef: core.SignerRef{
+				Key: "0x1",
+			},
+		},
+	}
+
+	full, overflow := bp.addWork(&batchWork{msg: msg1})
+	assert.False(t, full)
+	assert.False(t, overflow)
+	assert.Equal(t, []*batchWork{
+		{msg: msg1},
+	}, bp.assemblyQueue)
+
+	full, overflow = bp.addWork(&batchWork{msg: msg3})
+	assert.False(t, full)
+	assert.False(t, overflow)
+	assert.Equal(t, []*batchWork{
+		{msg: msg1},
+		{msg: msg3},
+	}, bp.assemblyQueue)
+
+	full, overflow = bp.addWork(&batchWork{msg: msg2})
+	assert.True(t, full)
+	assert.True(t, overflow)
+	assert.Equal(t, []*batchWork{
+		{msg: msg1},
+		{msg: msg3},
+		{msg: msg2},
+	}, bp.assemblyQueue)
+}
+
+func TestAddWorkAbandonedBatch(t *testing.T) {
+	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
+		return nil
+	})
+	defer cancel()
+	msg := &core.Message{Sequence: 200, BatchID: fftypes.NewUUID()}
+	_, _ = bp.addWork(&batchWork{msg: msg})
+	assert.Equal(t, []*batchWork{{msg: msg}}, bp.assemblyQueue)
+}
+
 func TestStartQuiesceNonBlocking(t *testing.T) {
 	cancel, _, bp := newTestBatchProcessor(t, func(c context.Context, state *DispatchState) error {
 		return nil