最近在处理智能合约的事务上链问题，发现其中仍旧有知识盲点。原有的认识是一个事务请求会从客户端设备打包签名，然后通过RPC传到非出块节点，广播给超级节点，校验打包到可逆区块，共识确认最后变为不可逆区块。在执行事务完毕以后给客户端一个“executed”的状态响应。基于这个认识，本文将通过最新EOS代码详细分析验证。

关键字：EOS，区块链，eosjs，transaction，签名，节点，出块节点，事务校验，事务广播

客户端的处理：打包与签名

客户端设备可以通过eosjs完成本地的事务体构建。下面以调用hello智能合约为例。

注意：eosio.cdt的hello合约中hi方法的参数名为nm，而不是user，我们下面采用与cdt相一致的方式。

方便起见，可以首先使用eosjs-api提供的transact方法，它可以帮助我们直接将事务体打包签名并推送出去。

(async () => {const result = await api.transact({actions: [{account: 'useraaaaaaaa', // 合约部署者，是一个EOS账户name: 'hi',              // 调用方法名，hello合约的一个方法。authorization: [{        // 该方法需要的权限，默认为合约部署者权限actor: 'useraaaaaaaa',permission: 'active',}],data: {                 // 方法参数nm: 'water'},}]}, {blocksBehind: 3,            // 顶部区块之前的某区块信息作为引用数据，这是TAPoS的概念。expireSeconds: 30,          // 过期时间设置，自动计算当前区块时间加上过期时间，得到截止时间。});
})();

然后我们可以进入transact方法中查看，仿照其实现逻辑，自行编写一个完整流程的版本。

“打包”在EOS中与“压缩”，“序列化”，“转hex”等是相同的，因此所有之前提到过的压缩，转化等概念都是指同一件事。例如compression：none属性，之前也提到过zlib的方式；cleos中convert命令；rpc中的abi_json_to_bin等。

①打包Actions

actions的结构与前面是相同的。

// actions结构与上面相同，这是我们与链交互的“个性化参数”
let actions = [{ account: 'useraaaaaaaa',name: 'hi',authorization: [{actor: 'useraaaaaaaa',permission: 'active'}],data: {nm: 'seawater'}
}];
// 打包Actions
let sActions = await api.serializeActions(actions);

eosjs中通过serializeActions方法将Actions对象序列化，序列化会把data的值压缩（可理解为密文传输参数以及参数的值），最终变为：

[{account: 'useraaaaaaaa',name: 'hi',authorization: [{actor: 'useraaaaaaaa',permission: 'active'}],data: '0000005765C38DC2'
}]

②打包Transaction

首先设置事务Transactions的属性字段。

let expireSeconds = 3;                                          // 设置过期时间为3秒
let blocktime = new Date(block.timestamp).getTime();            // 获得引用区块的时间:1566263146500
let timezone = new Date(blocktime + 8*60*60*1000).getTime();    // 获得+8时区时间:1566291946500
let expired = new Date(timezone + expireSeconds * 1000);        // 获得过期时间:2019-08-20T09:05:49.500Z
let expiration = expired.toISOString().split('.')[0];           // 转换一下，得到合适的值:2019-08-20T09:05:49expiration: expiration,                     // 根据延迟时间与引用区块的时间计算得到的截止时间ref_block_num: block.block_num,             // 引用区块号，来自于查询到的引用区块的属性值ref_block_prefix: block.ref_block_prefix,   // 引用区块前缀，来自于查询到的引用区块的属性值max_net_usage_words: 0,                     // 设置该事务的最大net使用量，实际执行时评估超过这个值则自动退回，0为不设限制max_cpu_usage_ms: 0,                        // 设置该事务的最大cpu使用量，实际执行时评估超过这个值则自动退回，0为不设限制compression: 'none',                        // 事务压缩格式，默认为none，除此之外还有zlib等。delay_sec: 0,                               // 设置延迟事务的延迟时间，一般不使用。context_free_actions: [],                   actions: sActions,                          // 将前面处理好的Actions对象传入。transaction_extensions: [],                 // 事务扩展字段，一般为空。
};
let sTransaction = await api.serializeTransaction(transaction); // 打包事务

注释中没有对context_free_actions进行说明，是因为这个字段在《区块链 + 大数据：EOS存储》中有详解。
eosjs中通过serializeTransaction方法将Transaction对象序列化，得到一个Uint8Array类型的数组，这就是事务压缩完成的值。

Uint8Array[198, 164, 91, 93, 21, 141, 3, 236, 69, 55, 0, 0, 0, 0, 1, 96, 140, 49, 198, 24, 115, 21, 214, 0, 0, 0, 0, 0, 0, 128, 107, 1, 96, 140, 49, 198, 24, 115, 21, 214, 0, 0, 0, 0, 168, 237, 50, 50, 8, 0, 0, 0, 87, 101, 195, 141, 194, 0]

③准备密钥

密钥的准备分两步：首先通过已处理完毕的事务体获得所需密钥requiredKeys，然后在本地密钥库中查看可用密钥availableKeys，比对找到对应密钥。

signatureProvider.getAvailableKeys().then(function (avKeys) { // 获得本地可用密钥// 查询事务必须密钥rpc.getRequiredKeys({transaction: transaction, availableKeys: avKeys}).then(function (reKeys) {// 匹配成功：本地可用密钥库中包含事务必须密钥console.log(reKeys);});
});

由于执行结果存在先后的依赖关系，因此要采用回调嵌套的方式调用。最后成功获得匹配的密钥：

[ 'PUB_K1_69X3383RzBZj41k73CSjUNXM5MYGpnDxyPnWUKPEtYQmVzqTY7' ]

小插曲：关于block.timestamp 与 expiration的处理在第②步的代码注释中分析到了，expiration的正确取值直接影响到了rpc的getRequiredKeys方法的调用，否则会报错：“Invalid Transaction”，这是由于事务体属性字段出错导致。另外时区的问题也要注意，new Date得到的是UTC时间，客户端一般可根据自己所在时区自动调整。

④本地签名

signatureProvider.sign({ // 本地签名。chainId: chainId,requiredKeys: reKeys,serializedTransaction: sTransaction
}).then(function (signedTrx) {console.log(signedTrx);
});

注意，这部分代码要代替第③步中的console.log(reKeys);，以达到回调顺序依赖的效果。得到的签名事务的结果如下：

{signatures: ['SIG_K1_Khut1qkaDDeL26VVT4nEqa6vzHf2wgy5uk3dwNF1Fei9GM1c8JvonZswMdc3W5pZmvNnQeEeLLgoCwqaYMtstV3h5YyesV'],serializedTransaction: Uint8Array[117, 185, 91, 93, 114, 182, 131, 21, 248, 224, 0, 0, 0, 0, 1, 96, 140, 49, 198, 24, 115, 21, 214, 0, 0, 0, 0, 0, 0, 128, 107, 1, 96, 140, 49, 198, 24, 115, 21, 214, 0, 0, 0, 0, 168, 237, 50, 50, 8, 0, 0, 0, 87, 101, 195, 141, 194, 0]
}

注意是由signatures和serializedTransaction两个属性构成的。

⑤推送事务

push_transaction方法的参数与第④步得到的结果结构是一致的，因此该对象可以直接被推送。

rpc.push_transaction(signedTrx).then(function (result) {console.log(result);
})

注意，这部分代码要代替第④步中的console.log(signedTrx);，以达到回调顺序依赖的效果。得到推送结果为：

{transaction_id: '4bc089165103879c4fcfc5331c8b03402e8206f8030c0c53374d31f5a1b35688',processed: {id: '4bc089165103879c4fcfc5331c8b03402e8206f8030c0c53374d31f5a1b35688',block_num: 47078,block_time: '2019-08-20T09:15:24.000',producer_block_id: null,receipt: {status: 'executed',cpu_usage_us: 800,net_usage_words: 13},elapsed: 800,net_usage: 104,scheduled: false,action_traces: [[Object]],except: null}
}

注意receipt响应值中包含了status: 'executed的内容，这个属性将是下文着重提及的。

源码位置

小结

事务的打包与签名是在客户端通过eosjs等工具完成的。从应用角度来看，直接使用api提供的transact是最简单的方法，但如果要理解其中的逻辑，可以自行编写一遍，但没必要重新做封装，毕竟transact已经有了。

节点的处理：校验、执行和广播

经过上一节，请求从客户端发出来到达了RPC供应商。RPC服务的提供者包括出块节点和非出块节点，一般来讲是非出块节点。非出块节点也会通过EOSIO/eos搭建一个nodeos服务，可以配置选择自己同步的数据区域，不具备出块能力。非出块节点如果想具备释放RPC服务的能力，需要配置chain_api_plugin，http_plugin。这部分内容可以转到《EOS行为核心：解析插件chain_plugin》详述。

push_transaction的返回结构体与上一节的响应数据体是一致的。

struct push_transaction_results {chain::transaction_id_type  transaction_id;fc::variant                 processed;
};

记住这两个字段，然后向上滑动一点点，观察具体的响应数据内容。

关于RPC的push_transaction方法的论述链接。继承这篇文章的内容，下面进行补充。

transaction_async

事务的同步是通过transaction_async方法完成的，调用关系是chain_plugin插件通过method机制跳转到producer_plugin中。
此时事务停留在非出块节点的chain_plugin.cpp的void read_write::push_transaction方法中。除了传入的事务体对象参数外，还有作为回调接收响应的push_transaction_results结构的实例next。进入函数体，首先针对传入的参数对象params（具体内容参见上一节④本地签名最后的签名事务），转为transaction_metadata的实例ptrx。接下来调用

app().get_method<incoming::methods::transaction_async>()

这是method模板的语法，方法后紧跟传入等待同步的参数ptrx等以及一个result接收结果的对象（result由非出块节点接收，这部分将在下一小节展开）。transaction_async作为method的Key值，被声明在incoming::methods::transaction_async命名空间下。app应用实例的method集合中曾经注册过该Key值，注册的方式是关联一个handle provider。这段注册的代码位于producer_plugin.cpp，

incoming::methods::transaction_async::method_type::handle _incoming_transaction_async_provider;

该provider内容实际上是调用了producer_plugin.cpp的on_incoming_transaction_async方法，正在同步进来的事务。接下来调用process_incoming_transaction_async方法，处理正在进入的事务同步。这个方法首先会判断当前节点是否正在出块，如果未出块则进入_pending_incoming_transactions容器，这是一个双队列结构。

这些等待中的事务将会在出块节点开始出块时通过start_block方法触发重新回到process_incoming_transaction_async方法进行打包。

transaction_ack

当接收全节点同步过来的事务的出块节点处于当值轮次时，会将接收的事务立即向其他节点（包括非出块节点）进行广播，主要通过channel机制跳转到net_plugin中。
目前事务停留在当值出块节点的producer_plugin的process_incoming_transaction_async方法中。transaction_ack作为channel号被声明在producer插件的compat::channels::transaction_ack命名空间下。这个channel是由net_plugin订阅。

channels::transaction_ack::channel_type::handle  incoming_transaction_ack_subscription;

这个频道的订阅器是net插件确认正在进来的事务。订阅器的实现方法绑定在net_plugin_impl::transaction_ack方法上。

my->incoming_transaction_ack_subscription = app().get_channel<channels::transaction_ack>().subscribe(boost::bind(&net_plugin_impl::transaction_ack, my.get(), _1));

进入net_plugin_impl::transaction_ack方法。

/*** @brief 出块节点确认事务* * @param results 二元组pair类型，第一个元素为异常信息，第二个元素为事务数据。*/
void net_plugin_impl::transaction_ack(const std::pair<fc::exception_ptr, transaction_metadata_ptr>& results) {const auto& id = results.second->id; // 从事务体中得到事务id。if (results.first) { //如果存在异常情况则拒绝广播该事务。fc_ilog(logger,"signaled NACK, trx-id = ${id} : ${why}",("id", id)("why", results.first->to_detail_string()));dispatcher->rejected_transaction(id);} else { // 无异常情况，广播该事务。打印事务确认消息，到这一步就说明当前节点完成了确认fc_ilog(logger,"signaled ACK, trx-id = ${id}",("id", id));dispatcher->bcast_transaction(results.second);}
}

成功确认以后，调用bcast_transaction方法继续广播该事务。

/*** @brief 事务广播给其他节点* * @param ptrx 事务体*/
void dispatch_manager::bcast_transaction(const transaction_metadata_ptr& ptrx) {std::set<connection_ptr> skips; // 相当于连接黑名单，从连接集合中跳过广播。const auto& id = ptrx->id; // 获取事务idauto range = received_transactions.equal_range(id); // 已接收事务集是接收其他节点广播的事务，而不是自己发起广播的事务for (auto org = range.first; org != range.second; ++org) {skips.insert(org->second); // 如果找到该事务，说明该事务已被其他节点优先广播，则自己不必额外处理。将事务连接插入skips集合。}received_transactions.erase(range.first, range.second); // 删除已接收事务集中该事务，逻辑清空。// 在本地事务集local_txns中查询，若找到则直接退出，说明该事务已完成广播共识。if( my_impl->local_txns.get<by_id>().find( id ) != my_impl->local_txns.end() ) {fc_dlog(logger, "found trxid in local_trxs" );return;}// 将事务插入到本地事务集local_txnstime_point_sec trx_expiration = ptrx->packed_trx->expiration();const packed_transaction& trx = *ptrx->packed_trx;auto buff = create_send_buffer( trx );node_transaction_state nts = {id, trx_expiration, 0, buff};my_impl->local_txns.insert(std::move(nts));// 符合广播条件，开始广播。my_impl->send_transaction_to_all( buff, [&id, &skips, trx_expiration](const connection_ptr& c) -> bool {if( skips.find(c) != skips.end() || c->syncing ) {return false; // 若该事务已被其他节点优先广播，则自己不做处理。}const auto& bs = c->trx_state.find(id);bool unknown = bs == c->trx_state.end();if( unknown ) { // trx_state未找到事务，则插入。c->trx_state.insert(transaction_state({id,0,trx_expiration}));fc_dlog(logger, "sending trx to ${n}", ("n",c->peer_name() ) );}return unknown;});
}

继续，进入send_transaction_to_all方法，查看广播的具体实现。net插件维护了一个connections集合，该集合动态维护了全网节点的p2p连接情况。

/*** @brief 模板方法：发送事务给全体成员* * @tparam VerifierFunc 模板类* @param send_buffer 事务数据* @param verify 模板类实例*/
template<typename VerifierFunc>
void net_plugin_impl::send_transaction_to_all(const std::shared_ptr<std::vector<char>>& send_buffer, VerifierFunc verify) {for( auto &c : connections) {if( c->current() && verify( c )) { // 在上面的使用中，就是检查是否在skips集合中。// 进入连接队列，建立连接，发送消息。c->enqueue_buffer( send_buffer, true, priority::low, no_reason ); // enqueue_buffer->queue_write->do_queue_write->boost::asio::async_write}}
}

最终的建立socket连接并发送数据的过程在注释中已体现：enqueue_buffer -> queue_write -> do_queue_write -> boost::asio::async_write，不再深入源码详细讨论。

process_incoming_transaction_async

void net_plugin_impl::transaction_ack方法中的参数二元组对象results是由process_incoming_transaction_async方法体中对transaction_ack频道发布的数据。上一小节详细分析了transaction_ack频道的订阅处理，这一小节回到process_incoming_transaction_async方法分析transaction_ack频道的信息发布。该方法体内部首先定义了一个send_response方法。

auto send_response = [this, &trx, &chain, &next](const fc::static_variant<fc::exception_ptr, transaction_trace_ptr>& response) {next(response); // 通过next方法将response传回客户端。if (response.contains<fc::exception_ptr>()) { // 响应内容中有异常情况出现，则发布数据中的第一个元素为异常对象，作为transaction_ack在net插件中的result.first数据。_transaction_ack_channel.publish(priority::low, std::pair<fc::exception_ptr, transaction_metadata_ptr>(response.get<fc::exception_ptr>(), trx));if (_pending_block_mode == pending_block_mode::producing) { // 如果当前节点正在出块，则打印日志区块拒绝该事务。fc_dlog(_trx_trace_log, "[TRX_TRACE] Block ${block_num} for producer ${prod} is REJECTING tx: ${txid} : ${why} ",("block_num", chain.head_block_num() + 1)("prod", chain.pending_block_producer())("txid", trx->id)("why",response.get<fc::exception_ptr>()->what())); // why的值为拒绝该事务的原因，即打印出异常对象的可读信息。} else { // 如果当前节点尚未出块，则打印未出块节点的推测执行：拒绝该事务。fc_dlog(_trx_trace_log, "[TRX_TRACE] Speculative execution is REJECTING tx: ${txid} : ${why} ",("txid", trx->id)("why",response.get<fc::exception_ptr>()->what())); // 同样打印异常}} else { // 如果响应内容中无异常，说明成功执行，则第一个元素为空。_transaction_ack_channel.publish(priority::low, std::pair<fc::exception_ptr, transaction_metadata_ptr>(nullptr, trx));if (_pending_block_mode == pending_block_mode::producing) { // 如果当前节点正在出块，则打印日志区块接收该事务。fc_dlog(_trx_trace_log, "[TRX_TRACE] Block ${block_num} for producer ${prod} is ACCEPTING tx: ${txid}",("block_num", chain.head_block_num() + 1)("prod", chain.pending_block_producer())("txid", trx->id));} else { // 如果当前节点尚未出块，则打印未出块节点的推测执行：接收该事务。fc_dlog(_trx_trace_log, "[TRX_TRACE] Speculative execution is ACCEPTING tx: ${txid}",("txid", trx->id));}}
};

从send_response方法的定义可以看出，第二个参数永远是事务体本身，这是不变的。而第一个参数是否包含异常信息是不确定的，取决于调用者的传入情况。所以接下来实际上是对事务状态的判断，从而影响传给send_response方法的第一个参数是否包含异常。这些异常情况包括：

1. 事务超时过期，通过将事务过期时间与当前最新区块时间对比即可，若小于最新区块时间则判定事务过期。
2. 事务重复，在当前节点的db中寻找是否有相同事务id的存在，若存在则说明事务重复。
3. 事务执行时出错:
   1. 全节点配置为只读模式的，不可以处理推送事务。
   2. 不允许忽略检查以及延迟事务。
   3. 内部执行错误，例如权限问题，资源问题，事务进入合约内部校验错误等，详细内容看下面对controller::push_transaction方法的分析。

controller::push_transaction

/*** @brief 这是新事务进入区块状态的进入点。将会检查权限，是否立即执行或延迟执行。*        最后，将事务返回体插入到等待中的区块。* * @param trx 事务体* @param deadline 截止时间* @param billed_cpu_time_us CPU抵押时间* @param explicit_billed_cpu_time CPU抵押时间是否明确，一般是false，未显式指定* * @return transaction_trace_ptr 事务跟踪，返回的结构体对象*/
transaction_trace_ptr push_transaction( const transaction_metadata_ptr& trx,fc::time_point deadline,uint32_t billed_cpu_time_us,bool explicit_billed_cpu_time = false )
{EOS_ASSERT(deadline != fc::time_point(), transaction_exception, "deadline cannot be uninitialized"); // 截止时间的格式出现问题transaction_trace_ptr trace; // 定义事务跟踪实例。try {auto start = fc::time_point::now();const bool check_auth = !self.skip_auth_check() && !trx->implicit; // implicit事务会忽略检查也可以自己设置跳过auth检查，则check_auth 为false。// 得到要使用的cpu的时间值。const fc::microseconds sig_cpu_usage = check_auth ? std::get<0>( trx->recover_keys( chain_id ) ) : fc::microseconds();// 得到权限的公钥const flat_set<public_key_type>& recovered_keys = check_auth ? std::get<1>( trx->recover_keys( chain_id ) ) : flat_set<public_key_type>();if( !explicit_billed_cpu_time ) { // 未显式指定CPU抵押时间。// 计算已消费CPU时间fc::microseconds already_consumed_time( EOS_PERCENT(sig_cpu_usage.count(), conf.sig_cpu_bill_pct) );if( start.time_since_epoch() <  already_consumed_time ) {start = fc::time_point();} else {start -= already_consumed_time;}}const signed_transaction& trn = trx->packed_trx->get_signed_transaction();transaction_context trx_context(self, trn, trx->id, start);if ((bool)subjective_cpu_leeway && pending->_block_status == controller::block_status::incomplete) {trx_context.leeway = *subjective_cpu_leeway;}trx_context.deadline = deadline;trx_context.explicit_billed_cpu_time = explicit_billed_cpu_time;trx_context.billed_cpu_time_us = billed_cpu_time_us;trace = trx_context.trace;try {if( trx->implicit ) { // 忽略检查的事务的处理办法trx_context.init_for_implicit_trx(); // 检查事务资源（CPU和NET）可用性。trx_context.enforce_whiteblacklist = false;} else {bool skip_recording = replay_head_time && (time_point(trn.expiration) <= *replay_head_time);// 检查事务资源（CPU和NET）可用性。trx_context.init_for_input_trx( trx->packed_trx->get_unprunable_size(),trx->packed_trx->get_prunable_size(),skip_recording);}trx_context.delay = fc::seconds(trn.delay_sec);if( check_auth ) {authorization.check_authorization( // 权限校验trn.actions,recovered_keys,{},trx_context.delay,[&trx_context](){ trx_context.checktime(); },false);}trx_context.exec(); // 执行事务上下文，合约方法内部的校验错误会在这里抛出，使事务行为在当前节点的链上生效。trx_context.finalize(); // 资源处理，四舍五入，自动扣除并更新账户的资源情况。auto restore = make_block_restore_point();if (!trx->implicit) {transaction_receipt::status_enum s = (trx_context.delay == fc::seconds(0))? transaction_receipt::executed: transaction_receipt::delayed;trace->receipt = push_receipt(*trx->packed_trx, s, trx_context.billed_cpu_time_us, trace->net_usage);pending->_block_stage.get<building_block>()._pending_trx_metas.emplace_back(trx);} else { // 以上代码段都包含在try异常监控的作用域中，因此如果到此仍未发生异常而中断，则判断执行成功。transaction_receipt_header r;r.status = transaction_receipt::executed; // 注意：这就是客户端接收到的那个非常重要的状态executed。r.cpu_usage_us = trx_context.billed_cpu_time_us;r.net_usage_words = trace->net_usage / 8;trace->receipt = r;}fc::move_append(pending->_block_stage.get<building_block>()._actions, move(trx_context.executed));if (!trx->accepted) {trx->accepted = true;emit( self.accepted_transaction, trx); // 发射接收事务的信号}emit(self.applied_transaction, std::tie(trace, trn));if ( read_mode != db_read_mode::SPECULATIVE && pending->_block_status == controller::block_status::incomplete ) {trx_context.undo(); // 析构器，undo撤销操作。} else {restore.cancel();trx_context.squash(); // 上下文刷新}if (!trx->implicit) {unapplied_transactions.erase( trx->signed_id );}return trace;} catch( const disallowed_transaction_extensions_bad_block_exception& ) {throw;} catch( const protocol_feature_bad_block_exception& ) {throw;} catch (const fc::exception& e) {trace->error_code = controller::convert_exception_to_error_code( e );trace->except = e;trace->except_ptr = std::current_exception();}if (!failure_is_subjective(*trace->except)) {unapplied_transactions.erase( trx->signed_id );}emit( self.accepted_transaction, trx ); // 发射接收事务的信号，触发controller相关信号操作emit( self.applied_transaction, std::tie(trace, trn) ); // 发射应用事务的信号，触发controller相关信号操作return trace;} FC_CAPTURE_AND_RETHROW((trace))
} /// push_transaction

信号方面的内容请转到controller的信号。

小结

我们知道，非出块节点和出块节点使用的是同一套代码部署的nodeos程序，然而非出块节点可以配置是否要只读模式，还是推测模式。所谓只读模式，是不做数据上传的，只能查询，不能新增，它的数据结构只保留不可逆区块的内容，十分简单。而推测模式是可以处理并推送事务的，它的数据结构除了不可逆区块的内容以外，还有可逆区块的内容。所以非出块节点是具备事务校验、本地执行以及广播的能力的，只是不具备区块打包的能力，到了区块层面的问题要到出块节点来解决。事务的广播和确认并不需要共识的存在，共识的发生是针对区块的，而区块打包是由出块节点来负责，因此区块共识只在出块节点之间完成。而事务的广播和确认只是单纯的接收事务，散发事务而已，可以在所有节点中完成。

出块节点的处理：打包区块、共识、不可逆

本节请参考文章EOS生产区块：解析插件producer_plugin。

前面介绍了事务的产生、执行、散发的过程，而事务被打包进区块的过程没有说明，可以参照start_block函数。这样，事务在区块链中就走完了完整过程。

本文仅代表作者观点，有疏漏部分欢迎讨论，经讨论正确的会自行更正。

更多文章请转到一面千人的博客园。