レッスン10 — run_engine_app と actor pipeline 経由の最初のブロック
問い
L9 でエンジンは起動するようになった。だが silent だ — start_engine が返ると actor は AppMsg::ConsensusReady を送って reply を待ち、誰も応えないので parked になる。どうやって engine と EL を繋ぎ、実際に block を 1 つ生成させるか？
原理（最小モデル）
AppMsgのルーティングループ。 engine が 1 本の channel にConsensusReady / GetValue / Decided / …を流す。app loop はwhile let Some(msg) = recv().awaitで各 variant に match し、replyに応えるか bridge を駆動する。これが Malachite ↔ EL の唯一の接着剤。- bridge に generic な多相性。
run_engine_app<B: ConsensusBridge>はStubBridge/InMemoryEvmBridge/RethEvmBridge/LiveRethEvmBridge全てで動く。ルーティング 1 つに backend 4 つ — L3 の trait surface がここで効く。 stop_after_decisionsは test ergonomics。 production はusize::MAX、テストは1。「テスト可能であるためだけ」の引数も正当な API 設計。- closed reply channel はログして伝播しない。 reply 前に engine が死ぬと
oneshot::Sender::send()が err。伝播すると本物のエラーがノイズに埋もれる →tracing::warn!でログ。
具体例
中継ループの 1 サイクル:
[Malachite engine actors] ──AppMsg(+oneshot reply)──► [run_engine_app loop] ──► [ConsensusBridge]
① ConsensusReady → app: reply.send((height, validator_set))
② GetValue → app: bridge.build_payload + payload_ready → reply.send(LocallyProposedValue)
③ Decided → app: bridge.commit(hash) → reply.send(Next::Start) / decided.push(hash)
if decided.len() >= stop_after_decisions { return Ok(decided) }
app は engine と bridge の 中継地点（ロジック本体でない）。重い計算は bridge、合意駆動は engine。
失敗例（誤解）
「Decided の早期 return で reply は省略していい」は誤り — engine actor が drop された sender を await し続けて parked になり tear-down が遅れる。return 前に必ず reply。また「各 arm を個別 unit test する」も誤り — arm は決まった順序で届くので、fake engine を作るより real engine を 1 block 回す方が安い（integration > unit）。
ここまでで「AppMsg ルーティング・generic bridge・reply 必須」は着地した。ここから engine_app.rs（~282 行）を組み立てる。コードは完全形。
🛑 予測。 engine が
GetValueを送ったとき、app はなぜBlockHashでなくLocallyProposedValue(height, round, value)で reply するのか？ ヒント: engine は propose した value を peer に gossip / certificate に含める必要がある。hash だけでは BFT machine 内で value が first-class にならない（solo で走っていることを engine は知らない）。
ステップで組み立てる
Step 1: crates/consensus/Cargo.toml に tracing
tracing = { workspace = true }
closed reply channel のケースで tracing::warn! を 1 箇所だけ使う（これはバグでなく上流が諦めたサイン — ログするが伝播しない）。
Step 2: engine_app.rs — imports + signature
//! Engine app loop — consumes `AppMsg` from the Malachite engine and routes
//! every consensus-relevant event through a [`ConsensusBridge`].
//!
//! This is the missing half of レッスン 9: with `OpenHlNode::start()` spinning
//! up the actor system, this loop is what makes those actors do useful work.
//! Once a `Decided` arrives we commit through the bridge, increment height,
//! and (optionally) stop after N decisions for tests.
use std::sync::Arc;
use eyre::eyre;
use informalsystems_malachitebft_app::engine::host::Next;
use informalsystems_malachitebft_app_channel::{AppMsg, Channels};
use informalsystems_malachitebft_core_types::Height as _;
use openhl_types::{BlockHash, PayloadAttrs};
use crate::bridge::ConsensusBridge;
use crate::context::OpenHlContext;
use crate::types::{OpenHlHeight, OpenHlValidatorSet, OpenHlValue};
const APP_REPLY_WAIT_LOG: &str = "engine_app: peer replied unsuccessfully (channel closed)";
/// Drive the engine app loop until `stop_after_decisions` decisions have been
/// committed through the bridge, or the consensus channel closes.
///
/// Returns the `BlockHash`es that were decided, in order. Single-validator mode
/// uses this with `stop_after_decisions = 1` to exit after the first block.
#[allow(clippy::too_many_lines)] // 12 AppMsg arms — laid out flat for lesson レッスン 11's match-by-match walk
pub async fn run_engine_app<B>(
bridge: Arc<B>,
mut channels: Channels<OpenHlContext>,
validator_set: OpenHlValidatorSet,
stop_after_decisions: usize,
) -> eyre::Result<Vec<BlockHash>>
where
B: ConsensusBridge + 'static,
{
let mut decided: Vec<BlockHash> = Vec::new();
let mut current_parent = BlockHash([0u8; 32]);
let mut current_height = OpenHlHeight::INITIAL;
while let Some(msg) = channels.consensus.recv().await {
match msg {
AppMsg::ConsensusReady { reply, .. } => {
if reply
.send((current_height, validator_set.clone()))
.is_err()
{
tracing::warn!("{APP_REPLY_WAIT_LOG} (ConsensusReady)");
}
}
AppMsg::StartedRound {
height,
round: _,
reply_value,
..
} => {
current_height = height;
if reply_value.send(Vec::new()).is_err() {
tracing::warn!("{APP_REPLY_WAIT_LOG} (StartedRound)");
}
}
AppMsg::GetValue {
height,
round,
timeout: _,
reply,
} => {
let attrs = default_attrs();
let id = bridge.build_payload(current_parent, attrs).await?;
let block = bridge.payload_ready(id).await?;
let value = OpenHlValue(block.hash);
let lpv =
informalsystems_malachitebft_app_channel::app::types::LocallyProposedValue::new(
height, round, value,
);
if reply.send(lpv).is_err() {
tracing::warn!("{APP_REPLY_WAIT_LOG} (GetValue)");
}
}
AppMsg::Decided {
certificate, reply, ..
} => {
let hash = certificate.value_id;
bridge.commit(hash).await?;
decided.push(hash);
current_parent = hash;
if decided.len() >= stop_after_decisions {
// Send a reply so consensus doesn't hang waiting on us before
// we drop the channel.
let next_height = certificate.height.increment();
let _ = reply.send(Next::Start(next_height, validator_set.clone()));
return Ok(decided);
}
let next_height = certificate.height.increment();
current_height = next_height;
if reply
.send(Next::Start(next_height, validator_set.clone()))
.is_err()
{
tracing::warn!("{APP_REPLY_WAIT_LOG} (Decided)");
}
}
AppMsg::ExtendVote { reply, .. } => {
if reply.send(None).is_err() {
tracing::warn!("{APP_REPLY_WAIT_LOG} (ExtendVote)");
}
}
AppMsg::VerifyVoteExtension { reply, .. } => {
if reply.send(Ok(())).is_err() {
tracing::warn!("{APP_REPLY_WAIT_LOG} (VerifyVoteExtension)");
}
}
AppMsg::RestreamProposal { .. } => {
// Single-validator mode never re-streams.
}
AppMsg::GetHistoryMinHeight { reply } => {
if reply.send(OpenHlHeight::INITIAL).is_err() {
tracing::warn!("{APP_REPLY_WAIT_LOG} (GetHistoryMinHeight)");
}
}
AppMsg::ReceivedProposalPart { reply, .. } => {
// ProposalOnly value-payload mode — proposal parts never arrive.
if reply.send(None).is_err() {
tracing::warn!("{APP_REPLY_WAIT_LOG} (ReceivedProposalPart)");
}
}
AppMsg::GetValidatorSet { reply, .. } => {
if reply.send(Some(validator_set.clone())).is_err() {
tracing::warn!("{APP_REPLY_WAIT_LOG} (GetValidatorSet)");
}
}
AppMsg::GetDecidedValue { reply, .. } => {
if reply.send(None).is_err() {
tracing::warn!("{APP_REPLY_WAIT_LOG} (GetDecidedValue)");
}
}
AppMsg::ProcessSyncedValue { reply, .. } => {
if reply.send(None).is_err() {
tracing::warn!("{APP_REPLY_WAIT_LOG} (ProcessSyncedValue)");
}
}
}
}
Err(eyre!(
"consensus channel closed after {n} decisions (wanted {stop_after_decisions})",
n = decided.len()
))
}
fn default_attrs() -> PayloadAttrs {
PayloadAttrs {
timestamp: 0,
fee_recipient: [0u8; 20],
prev_randao: [0u8; 32],
}
}
12 arm = substantive 5 + trivial 7。要点:
- GetValue（load-bearing）:
build_payload(current_parent, attrs)→payload_ready(id)→OpenHlValue→LocallyProposedValue::new(height,round,value)で reply（予測の答え: engine が gossip/certificate に使う）。 - Decided（load-bearing）:
certificate.value_idをcommit→decided.push→current_parent更新（次の GetValue がこの hash の上に build）→ exit 条件チェック（return 前に reply）→ なければNext::Start(next_height)。 - trivial 7: vote extension（
None/Ok(())、v0 未使用）、RestreamProposal(no-op)、history/sync（peer catch-up）、ReceivedProposalPart（ProposalOnly なので来ない、None）。 default_attrsは全ゼロ（レッスン12 で real になる）。
Step 3: lib.rs
//! Consensus layer — Malachite BFT.
pub mod bridge;
pub mod codec;
pub mod context;
pub mod engine_app;
pub mod node;
pub mod signing;
pub mod signing_provider;
pub mod types;
pub use context::OpenHlContext;
Step 4: StubBridge + integration test（engine_app.rs 末尾）
#[cfg(test)]
mod tests {
use super::*;
use crate::bridge::BridgeError;
use crate::node::OpenHlNode;
use crate::types::{OpenHlAddress, OpenHlValidator};
use async_trait::async_trait;
use informalsystems_malachitebft_app::node::{Node as _, NodeHandle as _};
use informalsystems_malachitebft_signing_ed25519::PrivateKey;
use openhl_types::{ExecutedBlock, PayloadId, PayloadStatus};
use rand::rngs::OsRng;
use sha2::{Digest, Sha256};
use std::sync::Mutex;
use std::time::Duration;
#[derive(Debug, Default)]
struct StubBridge {
last_built: Mutex<Option<BlockHash>>,
committed: Mutex<Vec<BlockHash>>,
}
#[async_trait]
impl ConsensusBridge for StubBridge {
async fn build_payload(
&self,
_parent: BlockHash,
_attrs: PayloadAttrs,
) -> Result<PayloadId, BridgeError> {
let hash = BlockHash([0x42u8; 32]);
*self.last_built.lock().expect("poisoned") = Some(hash);
Ok(PayloadId(1))
}
async fn payload_ready(
&self,
_id: PayloadId,
) -> Result<ExecutedBlock, BridgeError> {
Ok(ExecutedBlock {
hash: BlockHash([0x42u8; 32]),
parent_hash: BlockHash([0u8; 32]),
number: 1,
state_root: [0u8; 32],
})
}
async fn validate_payload(
&self,
_block: &ExecutedBlock,
) -> Result<PayloadStatus, BridgeError> {
Ok(PayloadStatus::Valid)
}
async fn commit(&self, block_hash: BlockHash) -> Result<(), BridgeError> {
self.committed.lock().expect("poisoned").push(block_hash);
Ok(())
}
}
fn make_test_node(home_dir: std::path::PathBuf) -> OpenHlNode {
let sk = PrivateKey::generate(OsRng);
let pk = sk.public_key();
let digest = Sha256::digest(pk.as_bytes());
let mut addr_bytes = [0u8; 20];
addr_bytes.copy_from_slice(&digest[12..32]);
let address = OpenHlAddress(addr_bytes);
let validator_set = OpenHlValidatorSet::new(vec![OpenHlValidator::new(address, pk, 1)]);
OpenHlNode::new(sk, validator_set, home_dir, "openhl-engine-test")
}
/// End-to-end: spawn the engine actor system, drive one block through the
/// `AppMsg` loop, assert the bridge built+committed exactly the hash the
/// engine decided on.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn first_block_via_engine_actors() {
let tmp = tempfile::tempdir().unwrap();
let node = make_test_node(tmp.path().to_path_buf());
let validator_set = node.validator_set.clone();
let handle = node.start().await.expect("start_engine failed");
let channels = handle
.take_channels()
.await
.expect("channels available exactly once");
let bridge = Arc::new(StubBridge::default());
let bridge_for_check = bridge.clone();
let app_task = tokio::spawn(run_engine_app(bridge, channels, validator_set, 1));
let decisions = tokio::time::timeout(Duration::from_secs(15), app_task)
.await
.expect("app loop timed out")
.expect("app task panicked")
.expect("app loop returned error");
assert_eq!(decisions.len(), 1, "expected exactly one decided block");
let decided_hash = decisions[0];
let committed = bridge_for_check.committed.lock().unwrap().clone();
assert_eq!(committed, vec![decided_hash], "bridge must commit decided hash");
assert_eq!(
*bridge_for_check.last_built.lock().unwrap(),
Some(decided_hash),
"decided hash must match what we built",
);
handle.kill(None).await.unwrap();
}
}
StubBridge は全てに BlockHash([0x42;32]) を返す in-memory fixture。integration test は spawn → take_channels → run_engine_app(..., stop_after_decisions=1) を tokio::spawn → timeout(15s) で bound → 3 段 unwrap（timeout/panic/loop-error）→ 3 assert（decisions=1 / commit が decided hash / built が decided hash）→ kill。worker_threads = 4 は engine 内部 actor + app loop + bridge async を同時に物理コアに散らし、スレッド枯渇型 deadlock を構造的に防ぐため。
答え合わせ
cd ~/code/openhl-reference && git checkout 708472c
diff -u ~/code/my-openhl/crates/consensus/src/engine_app.rs ./crates/consensus/src/engine_app.rs
git checkout main
708472c に 282 行の engine_app.rs。12 arm・StubBridge・integration test は厳密一致するべき。
合格基準
cargo test -p openhl-consensus
→ 21 テスト pass（レッスン9 の 20 + first_block_via_engine_actors）。test 本体 ~0.02 秒。よくあるミス（多くは 15 秒 hang = timeout 発火）: reply.send 忘れ / Decided の早期 return で reply 忘れ / GetValue 未 handle で engine が進まず decisions が空。
まとめ（3行）
- app loop は engine の
AppMsgをwhile let recv()で受け、reply するか bridge を駆動する中継地点。 run_engine_app<B>は generic — 同じループが Stub/InMemory/Reth/LiveReth で動く（L3 trait surface の配当）。first_block_via_engine_actorsが engine→app→bridge→engine の完全パイプラインを 0.02 秒で証明 — block 生成の milestone。