レッスン9 — OpenHlNode と初の start_engine 呼び出し

問い

Malachite エンジンを実際に起動するには何を渡せばいいか？ そして「構築（config を保持するだけ）」と「実行（actor system が走る）」を、なぜ別の型に住み分けさせるのか？

原理（最小モデル）

• Node は handshake interface（runtime ではない）。 OpenHlNode は長命な設定（key / validator set / home dir / moniker）を保持し engine を 構築 する。走っている actor system は start() が返す OpenHlNodeHandle の中。構築と実行は別ライフサイクル → 別の型。
• Mutex<Option<Channels>> の take-once。 channel handle はちょうど 1 回だけ取り出せる。レッスン10 の app loop が消費し、2 回目は None — 所有権が移った clean なシグナル。
• address 導出を 1 箇所に集約。 SHA-256(pubkey)[12..32] を get_address にだけ書き、テストで runner helper との一致を assert。集約 + 検証テストで silent な drift を防ぐ。
• todo!() でなく型安全 placeholder。 run() は panic でなく Err("...(レッスン10)") を返す。呼んだコードは graceful に失敗し次レッスンへの pointer 付きで止まる。

具体例

ライフサイクルの分離:

OpenHlNode { private_key, validator_set, home_dir, moniker }   ← 静的 config、長命、engine 未起動
      │  .start().await   (Node trait の handshake)
      ▼
OpenHlNodeHandle {
    engine:   EngineHandle                          ← ractor cell + libp2p 起動中
    channels: Mutex<Option<Channels<OpenHlContext>>> ← レッスン10 が take() で 1 回引き抜く
}                                                    ← .kill().await まで生存

失敗例（誤解）

「#[derive(Debug)] でいい」は誤り — private key の 32 バイトがログ/Sentry に漏れる。手書き Debug で [redacted] する。「smoke test は #[tokio::test] でいい」も誤り — default の current_thread だと engine 内部の block_on が唯一のワーカーを占有して永久ハング。multi_thread を強制する。

ここまでで「Node は handshake・構築と実行を分離」は着地した。ここから node.rs（~310 行）を組み立てる。コードは完全形。

🛑 予測。 なぜ Malachite は OpenHlNode 自身に config フィールドを持たせず、別途 OpenHlConfig を要求するのか？ ヒント: config の所有者といつ変わりうるか。node はプロセス起動時に 1 回作るが、config はシグナルでディスクから再ロードされうる。

ステップで組み立てる

Step 1: crates/consensus/Cargo.toml（依存追加）

[dependencies]
openhl-types = { workspace = true }
async-trait  = { workspace = true }
thiserror    = { workspace = true }
eyre         = { workspace = true }

informalsystems-malachitebft-core-types      = { workspace = true }
informalsystems-malachitebft-core-driver     = { workspace = true }
informalsystems-malachitebft-core-consensus  = { workspace = true }
informalsystems-malachitebft-app             = { workspace = true }
informalsystems-malachitebft-app-channel     = { workspace = true }
informalsystems-malachitebft-config          = { workspace = true }
informalsystems-malachitebft-signing-ed25519 = { workspace = true, features = ["rand", "serde"] }
bytes                                         = "1"
rand                                          = "0.8"
sha2                                          = "0.10"
serde                                         = { workspace = true }
tokio                                         = { workspace = true }

[dev-dependencies]
tokio    = { workspace = true }
tempfile = "3"

[lints]
workspace = true

新規: app-channel（start_engine + Channels<Ctx>）/ config（ConsensusConfig 等）/ signing-ed25519 に serde（OpenHlPrivateKeyFile の derive）/ serde・tokio を runtime dep へ（handle が tokio::sync::Mutex を持つ）/ tempfile(dev、smoke test の home dir)。

Step 2: node.rs — imports + OpenHlConfig

//! `Node` trait implementation — describes our chain to Malachite's engine
//! and provides the [`OpenHlNode::start`] entry point that calls
//! `malachitebft_app_channel::start_engine` to spawn the actor system.

use std::path::PathBuf;

use async_trait::async_trait;
use eyre::eyre;
use informalsystems_malachitebft_app::node::{EngineHandle, Node, NodeConfig, NodeHandle};
use informalsystems_malachitebft_app::types::Keypair;
use informalsystems_malachitebft_app_channel::Channels;
use informalsystems_malachitebft_config::{ConsensusConfig, ValueSyncConfig, ValuePayload};
use informalsystems_malachitebft_core_types::Height as _;
use informalsystems_malachitebft_signing_ed25519::{PrivateKey, PublicKey};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use tokio::sync::Mutex;

use crate::codec::OpenHlCodec;
use crate::context::OpenHlContext;
use crate::signing_provider::OpenHlSigningProvider;
use crate::types::{OpenHlAddress, OpenHlHeight, OpenHlValidatorSet};

#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct OpenHlConfig {
    pub moniker: String,
    #[serde(flatten)]
    pub consensus: ConsensusConfig,
    pub value_sync: ValueSyncConfig,
}

impl OpenHlConfig {
    #[must_use]
    pub fn new(moniker: impl Into<String>) -> Self {
        // OpenHL runs ProposalOnly (no streaming proposal parts) — must match
        // our `Context::ProposalPart` shape.
        let consensus = ConsensusConfig {
            value_payload: ValuePayload::ProposalOnly,
            ..ConsensusConfig::default()
        };
        Self {
            moniker: moniker.into(),
            consensus,
            value_sync: ValueSyncConfig::default(),
        }
    }
}

impl NodeConfig for OpenHlConfig {
    fn moniker(&self) -> &str {
        &self.moniker
    }
    fn consensus(&self) -> &ConsensusConfig {
        &self.consensus
    }
    fn value_sync(&self) -> &ValueSyncConfig {
        &self.value_sync
    }
}

#[serde(flatten)] で consensus フィールドを親に inline。new() が value_payload: ProposalOnly を強制 — Context::ProposalPart = OpenHlProposalPart(unit) と必ず合致させる（後でデバッグするより構築時に強制）。

Step 3: OpenHlGenesis + OpenHlPrivateKeyFile

/// Genesis is a unit struct at v0 — the validator set is passed directly to
/// `start_engine` rather than read from disk. When `OpenHL` grows a real
/// on-disk genesis format this becomes the `load_genesis()` return.
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct OpenHlGenesis;

/// Wire-friendly wrapper around the raw 32-byte Ed25519 private key.
#[derive(Clone, Serialize, Deserialize)]
pub struct OpenHlPrivateKeyFile {
    pub bytes: [u8; 32],
}

impl OpenHlPrivateKeyFile {
    #[must_use]
    pub fn from_private_key(sk: &PrivateKey) -> Self {
        Self {
            bytes: sk.inner().to_bytes(),
        }
    }

    #[must_use]
    pub fn into_private_key(self) -> PrivateKey {
        PrivateKey::from(self.bytes)
    }
}

impl std::fmt::Debug for OpenHlPrivateKeyFile {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("OpenHlPrivateKeyFile")
            .field("bytes", &"[redacted]")
            .finish()
    }
}

PrivateKey(malachite 由来)は default で serde を impl しないので wrapper が担う。手書き Debug がバイトを redact — #[derive(Debug)] だと 32 バイトが print されて key リーク（予測の失敗例）。

Step 4: OpenHlNodeHandle — start() の戻り値

/// Handle returned by [`OpenHlNode::start`]. Owns the engine actor system
/// and the channel handles for the (yet-to-be-implemented) app loop.
pub struct OpenHlNodeHandle {
    engine: EngineHandle,
    channels: Mutex<Option<Channels<OpenHlContext>>>,
}

impl std::fmt::Debug for OpenHlNodeHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("OpenHlNodeHandle")
            .field("engine", &"<EngineHandle>")
            .field("channels", &"<Channels>")
            .finish()
    }
}

impl OpenHlNodeHandle {
    /// Take ownership of the engine→app message channels. Returns None on
    /// the second call. レッスン 10 will consume from this to drive the bridge.
    pub async fn take_channels(&self) -> Option<Channels<OpenHlContext>> {
        self.channels.lock().await.take()
    }
}

#[async_trait]
impl NodeHandle<OpenHlContext> for OpenHlNodeHandle {
    fn subscribe(&self) -> informalsystems_malachitebft_app::events::RxEvent<OpenHlContext> {
        // No event subscription in Stage 6c — caller can't yet observe engine
        // events. レッスン 10 wires the TxEvent from the engine to here.
        informalsystems_malachitebft_app::events::TxEvent::new().subscribe()
    }

    async fn kill(&self, _reason: Option<String>) -> eyre::Result<()> {
        self.engine.actor.kill_and_wait(None).await?;
        self.engine.handle.abort();
        Ok(())
    }
}

tokio::sync::Mutex（std でない）を使うのは take_channels が async で lock が .await 境界をまたぐから（std::sync::Mutex だと executor スレッドをブロック）。subscribe は placeholder（producer 未 attach の空ストリーム、レッスン10 で本物）、kill は本物で smoke test が exercise する。

Step 5: OpenHlNode struct + Node impl

#[derive(Clone, Debug)]
pub struct OpenHlNode {
    pub private_key: PrivateKey,
    pub validator_set: OpenHlValidatorSet,
    pub home_dir: PathBuf,
    pub moniker: String,
}

impl OpenHlNode {
    #[must_use]
    pub fn new(
        private_key: PrivateKey,
        validator_set: OpenHlValidatorSet,
        home_dir: PathBuf,
        moniker: impl Into<String>,
    ) -> Self {
        Self {
            private_key,
            validator_set,
            home_dir,
            moniker: moniker.into(),
        }
    }
}

#[async_trait]
impl Node for OpenHlNode {
    type Context = OpenHlContext;
    type Config = OpenHlConfig;
    type Genesis = OpenHlGenesis;
    type PrivateKeyFile = OpenHlPrivateKeyFile;
    type SigningProvider = OpenHlSigningProvider;
    type NodeHandle = OpenHlNodeHandle;

    fn get_home_dir(&self) -> PathBuf {
        self.home_dir.clone()
    }

    fn load_config(&self) -> eyre::Result<Self::Config> {
        let mut cfg = OpenHlConfig::new(&self.moniker);
        // Bind to an ephemeral port on localhost so tests and devnets don't
        // step on each other. Real deployments override this in their config.
        cfg.consensus.p2p.listen_addr = "/ip4/127.0.0.1/tcp/0"
            .parse()
            .map_err(|e| eyre!("invalid listen_addr: {e}"))?;
        Ok(cfg)
    }

    fn get_address(&self, pk: &PublicKey) -> OpenHlAddress {
        let digest = Sha256::digest(pk.as_bytes());
        let mut addr = [0u8; 20];
        addr.copy_from_slice(&digest[12..32]);
        OpenHlAddress(addr)
    }

    fn get_public_key(&self, pk: &PrivateKey) -> PublicKey {
        pk.public_key()
    }

    fn get_keypair(&self, pk: PrivateKey) -> Keypair {
        Keypair::ed25519_from_bytes(pk.inner().to_bytes())
            .expect("ed25519 private key is always 32 bytes")
    }

    fn load_private_key(&self, file: Self::PrivateKeyFile) -> PrivateKey {
        file.into_private_key()
    }

    fn load_private_key_file(&self) -> eyre::Result<Self::PrivateKeyFile> {
        Ok(OpenHlPrivateKeyFile::from_private_key(&self.private_key))
    }

    fn load_genesis(&self) -> eyre::Result<Self::Genesis> {
        // Validator set is passed directly to start_engine; genesis carries
        // nothing else at v0.
        Ok(OpenHlGenesis)
    }

    fn get_signing_provider(&self, private_key: PrivateKey) -> Self::SigningProvider {
        OpenHlSigningProvider::new(private_key)
    }

    async fn start(&self) -> eyre::Result<Self::NodeHandle> {
        let cfg = self.load_config()?;
        let validator_set = self.validator_set.clone();

        let (channels, engine) = informalsystems_malachitebft_app_channel::start_engine(
            OpenHlContext,
            self.clone(),
            cfg,
            OpenHlCodec, // WAL
            OpenHlCodec, // Network
            Some(OpenHlHeight::INITIAL),
            validator_set,
        )
        .await?;

        Ok(OpenHlNodeHandle {
            engine,
            channels: Mutex::new(Some(channels)),
        })
    }

    async fn run(self) -> eyre::Result<()> {
        // レッスン 10 will consume from channels here and run the app loop.
        Err(eyre!("OpenHlNode::run is not yet implemented (レッスン 10)"))
    }
}

6 関連型がハンドシェイクの各スロットの具象型を宣言。start() がハイライト — start_engine を context / node(self.clone()) / config / codec ×2（WAL 用と Network 用、別々に渡すので一方だけ差し替え可能）/ 初期 height / validator set の 7 引数で呼び、戻り値 (Channels, EngineHandle) を handle に wrap。run() は app loop（レッスン10）未実装なので型安全エラーを返す。get_address は SHA-256(pubkey)[12..32]。

Step 6: lib.rs

//! Consensus layer — Malachite BFT.

pub mod bridge;
pub mod codec;
pub mod context;
pub mod node;
pub mod signing;
pub mod signing_provider;
pub mod types;

pub use context::OpenHlContext;

Step 7: 4 unit test（node.rs 末尾）

#[cfg(test)]
mod tests {
    use super::*;
    use crate::types::OpenHlValidator;
    use rand::rngs::OsRng;

    fn single_validator_node(home_dir: 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-test")
    }

    #[test]
    fn private_key_file_round_trips() {
        let sk = PrivateKey::generate(OsRng);
        let file = OpenHlPrivateKeyFile::from_private_key(&sk);
        let restored = file.into_private_key();
        assert_eq!(restored.inner().to_bytes(), sk.inner().to_bytes());
    }

    #[test]
    fn load_config_sets_proposal_only_payload_and_ephemeral_listen_addr() {
        let tmp = tempfile::tempdir().unwrap();
        let node = single_validator_node(tmp.path().to_path_buf());
        let cfg = node.load_config().unwrap();
        assert_eq!(cfg.consensus.value_payload, ValuePayload::ProposalOnly);
        // listen_addr should be /ip4/127.0.0.1/tcp/0 (ephemeral)
        let listen_str = cfg.consensus.p2p.listen_addr.to_string();
        assert!(
            listen_str.starts_with("/ip4/127.0.0.1/tcp/0"),
            "unexpected listen_addr: {listen_str}"
        );
    }

    #[test]
    fn get_address_matches_runner_derivation() {
        let tmp = tempfile::tempdir().unwrap();
        let node = single_validator_node(tmp.path().to_path_buf());
        let pk = node.private_key.public_key();
        let addr1 = node.get_address(&pk);
        // Same derivation as runner.rs (last 20 bytes of SHA-256(pubkey)).
        let digest = Sha256::digest(pk.as_bytes());
        let mut expected = [0u8; 20];
        expected.copy_from_slice(&digest[12..32]);
        assert_eq!(addr1, OpenHlAddress(expected));
    }

    /// Smoke test: spin up the actor system, get a handle back, kill cleanly.
    /// Does NOT drive consensus — that's レッスン 10.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn start_engine_smoke_spawns_and_kills() {
        let tmp = tempfile::tempdir().unwrap();
        let node = single_validator_node(tmp.path().to_path_buf());
        let handle = match node.start().await {
            Ok(h) => h,
            Err(e) => panic!("start_engine failed: {e:?}"),
        };
        // Sanity-poke the channels handle is available exactly once.
        assert!(handle.take_channels().await.is_some());
        assert!(handle.take_channels().await.is_none());
        handle.kill(None).await.unwrap();
    }
}

capstone は start_engine_smoke_spawns_and_kills — #[tokio::test(flavor = "multi_thread", worker_threads = 2)]（engine が複数 actor を spawn、current_thread だと deadlock）。node 構築 → start() → channels を 1 回 Some/2 回目 None で poke → kill()。~0.02 秒。pass すれば自分のコードが動く BFT エンジンになっている。

答え合わせ

cd ~/code/openhl-reference && git checkout d59d6cf
diff -u ~/code/my-openhl/crates/consensus/src/node.rs ./crates/consensus/src/node.rs
git checkout main

d59d6cf に 310 行の node.rs。12 method・struct レイアウト・smoke test は厳密一致するはず。doc は個人差可。

合格基準

cargo test -p openhl-consensus

→ 20 テスト pass（レッスン8 の 16 + node 4）。smoke test が最後に走る。よくあるミス: app-channel 依存忘れ / signing-ed25519 の serde feature 忘れ（PrivateKey: Deserialize 不充足）/ smoke test が current_thread で永久ハング / get_address が helper と不一致。

まとめ（3行）

• OpenHlNode(静的 config) と OpenHlNodeHandle(実行中 actor system) を分離 — 構築と実行は別ライフサイクル。
• start() が start_engine を 7 引数（context/node/config/codec×2/初期 height/validator set）で呼び engine を spawn。
• smoke test（multi_thread 必須）が spawn→channel take-once→kill を end-to-end で証明 — ここで動く BFT エンジンが立つ。