如何在HLF Chaincode中捕获MVCC_READ_CONFLICT错误并反馈至NodeJS服务

场景说明 #

在Ubuntu虚拟机上运行Data61BPMNtoChaincode项目，以NodeJS作为客户端服务，向Docker容器中的Peer节点发送链码安装、部署及交易请求。项目支持BPMN并行块流程（多活动分支并行执行，无先后依赖），但并发请求过快时会触发MVCC_READ_CONFLICT错误：Peer更新变量状态后，并发请求仍读取旧值，导致交易失败。当前通过添加3秒延迟规避问题，但生产环境不可行，需在Chaincode中捕获该错误并反馈至NodeJS服务，实现自动重发。

解决方案步骤 #

1. 在Chaincode中识别并抛出MVCC冲突错误 #

Fabric Chaincode执行时，MVCC_READ_CONFLICT错误会包含特定标识字符串，可在交易处理逻辑中捕获并包装成客户端可识别的自定义错误。

Go Chaincode示例

package main

import (
	"fmt"
	"strings"

	"github.com/hyperledger/fabric-contract-api-go/contractapi"
)

type SmartContract struct {
	contractapi.Contract
}

func (t *SmartContract) Invoke(ctx contractapi.TransactionContextInterface) error {
	fn, args := ctx.GetStub().GetFunctionAndParameters()

	switch fn {
	case "updateValue":
		err := t.updateValue(ctx, args)
		if err != nil {
			// 检测MVCC冲突错误
			if strings.Contains(err.Error(), "MVCC_READ_CONFLICT") {
				return fmt.Errorf("MVCC_CONFLICT: %w", err)
			}
			return err
		}
		return nil
	default:
		return fmt.Errorf("unsupported function: %s", fn)
	}
}

func (t *SmartContract) updateValue(ctx contractapi.TransactionContextInterface, args []string) error {
	if len(args) != 2 {
		return fmt.Errorf("incorrect number of arguments. Expecting 2")
	}

	key := args[0]
	newValue := args[1]

	currentValue, err := ctx.GetStub().GetState(key)
	if err != nil {
		return err
	}
	if currentValue == nil {
		return fmt.Errorf("key %s does not exist", key)
	}

	// 业务逻辑处理（如基于当前值计算新值）
	// ...

	err = ctx.GetStub().PutState(key, []byte(newValue))
	if err != nil {
		return err
	}
	return nil
}

func main() {
	chaincode, err := contractapi.NewChaincode(&SmartContract{})
	if err != nil {
		fmt.Printf("Error creating chaincode: %v", err)
		return
	}

	if err := chaincode.Start(); err != nil {
		fmt.Printf("Error starting chaincode: %v", err)
	}
}

NodeJS Chaincode示例

'use strict';

const { Contract } = require('fabric-contract-api');

class SmartContract extends Contract {
	async invoke(ctx) {
		const { fcn, args } = ctx.stub.getFunctionAndParameters();
		try {
			switch(fcn) {
				case 'updateValue':
					await this.updateValue(ctx, args);
					return Buffer.from('Success');
				default:
					throw new Error(`Unsupported function: ${fcn}`);
			}
		} catch (err) {
			// 检测MVCC冲突错误
			if (err.message.includes('MVCC_READ_CONFLICT')) {
				throw new Error(`MVCC_CONFLICT: ${err.message}`);
			}
			throw err;
		}
	}

	async updateValue(ctx, args) {
		if (args.length !== 2) {
			throw new Error('Incorrect number of arguments. Expecting 2');
		}
		const key = args[0];
		const newValue = args[1];

		const currentValue = await ctx.stub.getState(key);
		if (!currentValue || currentValue.length === 0) {
			throw new Error(`Key ${key} does not exist`);
		}

		// 业务逻辑处理
		// ...

		await ctx.stub.putState(key, Buffer.from(newValue));
	}
}

module.exports = SmartContract;

2. NodeJS客户端捕获错误并实现重发逻辑 #

在NodeJS服务端调用Fabric SDK时，捕获Chaincode返回的MVCC_CONFLICT标识错误，实现带指数退避的重试机制。

const { Gateway, Wallets } = require('fabric-network');
const path = require('path');
const fs = require('fs');

async function submitTransactionWithRetry(contract, functionName, args, maxRetries = 3) {
	let retries = 0;
	while (retries < maxRetries) {
		try {
			const result = await contract.submitTransaction(functionName, ...args);
			console.log(`交易执行成功: ${result.toString()}`);
			return result.toString();
		} catch (err) {
			if (err.message.includes('MVCC_CONFLICT') && retries < maxRetries - 1) {
				retries++;
				const delay = 100 * Math.pow(2, retries); // 指数退避延迟
				console.log(`检测到MVCC冲突，${delay}ms后进行第${retries}次重试`);
				await new Promise(resolve => setTimeout(resolve, delay));
				continue;
			}
			console.error(`交易失败: ${err.message}`);
			throw err;
		}
	}
	throw new Error(`达到最大重试次数(${maxRetries})，交易执行失败`);
}

async function main() {
	const gateway = new Gateway();
	try {
		// 加载连接配置
		const ccpPath = path.resolve(__dirname, '..', 'connection.json');
		const ccp = JSON.parse(fs.readFileSync(ccpPath, 'utf8'));

		// 加载钱包
		const walletPath = path.join(__dirname, 'wallet');
		const wallet = await Wallets.newFileSystemWallet(walletPath);

		// 连接网关
		const connectionOptions = {
			wallet,
			identity: 'user1',
			discovery: { enabled: true, asLocalhost: true }
		};
		await gateway.connect(ccp, connectionOptions);

		// 获取通道和合约
		const network = await gateway.getNetwork('mychannel');
		const contract = network.getContract('mychaincode');

		// 提交交易并启用重试
		await submitTransactionWithRetry(contract, 'updateValue', ['key1', '1']);
	} finally {
		// 断开网关连接
		gateway.disconnect();
	}
}

main().catch(err => {
	console.error(`执行失败: ${err.message}`);
	process.exit(1);
});

3. 额外优化建议 #

• 添加乐观锁：在Chaincode中为状态数据增加版本号字段，每次更新时验证版本号，不匹配则主动抛出冲突错误，减少依赖Fabric MVCC检查的被动性。
• 缩短事务时长：优化Chaincode业务逻辑，减少状态读写的间隔时间，降低锁持有时长，从根源减少冲突概率。
• 限制并发度：在NodeJS服务端对并行请求做流量控制，避免短时间内发送过多同类交易。

内容的提问来源于stack exchange，提问作者sebi