EOS源码分析：transaction的一生

文彬

发布于 2019-08-23 16:09:40

1.5K0

发布于 2019-08-23 16:09:40

文章被收录于专栏：醒者呆

最近在处理智能合约的事务上链问题，发现其中仍旧有知识盲点。原有的认识是一个事务请求会从客户端设备打包签名，然后通过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:49
    expiration: 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; // 获取事务id

   auto 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_txns
   time_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方法的第一个参数是否包含异常。这些异常情况包括：

事务超时过期，通过将事务过期时间与当前最新区块时间对比即可，若小于最新区块时间则判定事务过期。
事务重复，在当前节点的db中寻找是否有相同事务id的存在，若存在则说明事务重复。
事务执行时出错:
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程序，然而非出块节点可以配置是否要只读模式，还是推测模式。所谓只读模式，是不做数据上传的，只能查询，不能新增，它的数据结构只保留不可逆区块的内容，十分简单。而推测模式是可以处理并推送事务的，它的数据结构除了不可逆区块的内容以外，还有可逆区块的内容。所以非出块节点是具备事务校验、本地执行以及广播的能力的，只是不具备区块打包的能力，到了区块层面的问题要到出块节点来解决。事务的广播和确认并不需要共识的存在，共识的发生是针对区块的，而区块打包是由出块节点来负责，因此区块共识只在出块节点之间完成。而事务的广播和确认只是单纯的接收事务，散发事务而已，可以在所有节点中完成。