hypercall_state_commitment/

pipeline.rs

//! Commitment pipeline: converts engine state deltas into Merkle roots.
//!
//! The pipeline is the bridge between the engine (which knows nothing about
//! Merkle trees) and the JMT store (which knows nothing about accounts or
//! positions). The integration layer in the main binary feeds `StateDelta`s
//! in; the pipeline produces `BatchCommitment`s out.
//!
//! ```text
//! Engine → apply() → events → integration layer → StateDelta
//!                                                      ↓
//!                                              CommitmentPipeline
//!                                                      ↓
//!                                              BatchCommitment { state_root, risk_root, batch_root }
//! ```

use jmt::storage::{TreeReader, TreeUpdateBatch, TreeWriter};
use jmt::{KeyHash, RootHash, Version};

use crate::batch_root::compute_batch_root;
use crate::keys::StateKey;
use crate::leaves::*;
use crate::mmr::{
    HypercallMmr, MmrHash, MmrMetadataStore, PrepareMmrStore, PreparedMmrAppend, SupportedMmrStore,
};
#[cfg(feature = "rocksdb")]
use crate::rocks_store::{RocksDbMmrStore, RocksDbStore};
use crate::state_tree::new_jmt;

/// A batch of state changes to commit.
///
/// The integration layer constructs this from engine events and state reads.
/// The pipeline doesn't know or care where the data comes from — it just
/// converts it to JMT leaf updates.
///
/// All monetary values are pre-scaled to 1e8 by the caller.
#[derive(Debug, Default, Clone)]
pub struct StateDelta {
    pub commands: Vec<[u8; 32]>,
    pub obligations: Vec<[u8; 32]>,
    pub intents: Vec<[u8; 32]>,
    pub accounts: Vec<AccountUpdate>,
    pub option_positions: Vec<OptionPositionUpdate>,
    pub perp_positions: Vec<PerpPositionUpdate>,
    pub orders: Vec<OrderUpdate>,
    pub instruments: Vec<InstrumentUpdate>,
    pub oracles: Vec<OracleUpdate>,
    pub mmp_configs: Vec<MmpConfigUpdate>,
    pub risk: Vec<RiskUpdate>,
    pub global: Option<GlobalLeaf>,
}

#[derive(Debug, Clone)]
pub struct AccountUpdate {
    pub address: [u8; 20],
    pub leaf: AccountLeaf,
}

#[derive(Debug, Clone)]
pub struct OptionPositionUpdate {
    pub address: [u8; 20],
    pub symbol: String,
    /// None = position closed (delete leaf).
    pub leaf: Option<OptionPositionLeaf>,
}

#[derive(Debug, Clone)]
pub struct PerpPositionUpdate {
    pub address: [u8; 20],
    pub coin: String,
    /// None = position closed (delete leaf).
    pub leaf: Option<PerpPositionLeaf>,
}

#[derive(Debug, Clone)]
pub struct OrderUpdate {
    pub address: [u8; 20],
    pub order_id: u64,
    /// None = order cancelled/filled (delete leaf).
    pub leaf: Option<OrderLeaf>,
}

#[derive(Debug, Clone)]
pub struct InstrumentUpdate {
    pub symbol: String,
    pub leaf: InstrumentLeaf,
}

#[derive(Debug, Clone)]
pub struct OracleUpdate {
    pub underlying: String,
    pub leaf: OracleLeaf,
}

#[derive(Debug, Clone)]
pub struct MmpConfigUpdate {
    pub address: [u8; 20],
    pub currency: String,
    pub leaf: MmpConfigLeaf,
}

#[derive(Debug, Clone)]
pub struct RiskUpdate {
    pub address: [u8; 20],
    pub leaf: RiskLeaf,
}

/// Output of a batch commitment.
#[derive(Debug, Clone)]
pub struct BatchCommitment {
    pub version: Version,
    pub state_root: RootHash,
    pub risk_root: RootHash,
    pub command_mmr_root: MmrHash,
    pub obligation_mmr_root: MmrHash,
    pub intent_mmr_root: MmrHash,
    pub batch_root: [u8; 32],
    pub state_leaves_updated: usize,
    pub risk_leaves_updated: usize,
}

#[derive(Debug)]
pub struct PreparedBatchCommitment {
    pub commitment: BatchCommitment,
    state_update: Option<TreeUpdateBatch>,
    risk_update: Option<TreeUpdateBatch>,
    command_mmr: PreparedMmrAppend,
    obligation_mmr: PreparedMmrAppend,
    intent_mmr: PreparedMmrAppend,
}

/// The commitment pipeline.
///
/// Generic over the store backend (`S`) so it works with both
/// `FastMemoryStore` (tests) and `RocksDbStore` (production).
///
/// Two separate JMTs:
/// - **State tree**: accounts, positions, orders, instruments, oracles, global
/// - **Risk tree**: per-account risk metrics (equity, margin, scanning risk)
///
/// Plus three MMRs for append-only data (command log, obligations, intents).
pub struct CommitmentPipeline<S, M = crate::mmr::MemoryMmrStore> {
    state_store: S,
    risk_store: S,
    command_mmr: HypercallMmr<M>,
    obligation_mmr: HypercallMmr<M>,
    intent_mmr: HypercallMmr<M>,
    version: Version,
    prune_window: u64,
}

#[cfg(feature = "rocksdb")]
impl CommitmentPipeline<RocksDbStore, RocksDbMmrStore> {
    pub fn checkpoint_inherited_state_root(&self, version: Version) -> anyhow::Result<bool> {
        self.state_store.copy_latest_root_to_version(version)
    }

    pub fn checkpoint_inherited_risk_root(&self, version: Version) -> anyhow::Result<bool> {
        self.risk_store.copy_latest_root_to_version(version)
    }

    pub fn prune_stores(&self, prune_to: Version) -> anyhow::Result<(usize, usize)> {
        let (state_pruned, _) = self.state_store.prune(prune_to)?;
        let (risk_pruned, _) = self.risk_store.prune(prune_to)?;
        Ok((state_pruned, risk_pruned))
    }

    pub fn compact_stores(&self) -> anyhow::Result<()> {
        self.state_store.compact_all()?;
        self.risk_store.compact_all()?;
        Ok(())
    }
}

impl<S: TreeReader + TreeWriter> CommitmentPipeline<S, crate::mmr::MemoryMmrStore> {
    /// Create a new pipeline starting at version 0.
    pub fn new(state_store: S, risk_store: S, prune_window: u64) -> Self {
        Self {
            state_store,
            risk_store,
            command_mmr: HypercallMmr::new(),
            obligation_mmr: HypercallMmr::new(),
            intent_mmr: HypercallMmr::new(),
            version: 0,
            prune_window,
        }
    }
}

impl<S, M> CommitmentPipeline<S, M>
where
    S: TreeReader + TreeWriter,
    M: MmrMetadataStore + SupportedMmrStore + PrepareMmrStore,
    for<'a> &'a M: ckb_merkle_mountain_range::MMRStoreReadOps<MmrHash>
        + ckb_merkle_mountain_range::MMRStoreWriteOps<MmrHash>,
{
    pub fn new_with_mmrs(
        state_store: S,
        risk_store: S,
        command_mmr_store: M,
        obligation_mmr_store: M,
        intent_mmr_store: M,
        prune_window: u64,
    ) -> anyhow::Result<Self> {
        Ok(Self {
            state_store,
            risk_store,
            command_mmr: HypercallMmr::from_store(command_mmr_store)?,
            obligation_mmr: HypercallMmr::from_store(obligation_mmr_store)?,
            intent_mmr: HypercallMmr::from_store(intent_mmr_store)?,
            version: 0,
            prune_window,
        })
    }

    /// Resume from a known version (e.g., after loading from disk).
    pub fn with_version(mut self, version: Version) -> Self {
        self.version = version;
        self
    }

    /// Current version (number of committed batches).
    pub fn version(&self) -> Version {
        self.version
    }

    /// Configured number of recent versions to keep available for proofs.
    pub fn prune_window(&self) -> u64 {
        self.prune_window
    }

    /// Prepare a batch commitment without mutating the backing stores.
    pub fn prepare_commit(&self, delta: &StateDelta) -> anyhow::Result<PreparedBatchCommitment> {
        let version = self.version;

        let (state_root, state_count, state_update) = self.prepare_state_delta(delta, version)?;
        let (risk_root, risk_count, risk_update) = self.prepare_risk_delta(&delta.risk, version)?;
        let command_mmr = self.command_mmr.prepare_append_many(&delta.commands)?;
        let obligation_mmr = self
            .obligation_mmr
            .prepare_append_many(&delta.obligations)?;
        let intent_mmr = self.intent_mmr.prepare_append_many(&delta.intents)?;

        let command_mmr_root = command_mmr.root();
        let obligation_mmr_root = obligation_mmr.root();
        let intent_mmr_root = intent_mmr.root();

        let batch_root = compute_batch_root(
            &state_root,
            &risk_root,
            &command_mmr_root,
            &obligation_mmr_root,
            &intent_mmr_root,
        );

        let commitment = BatchCommitment {
            version,
            state_root,
            risk_root,
            command_mmr_root,
            obligation_mmr_root,
            intent_mmr_root,
            batch_root,
            state_leaves_updated: state_count,
            risk_leaves_updated: risk_count,
        };

        Ok(PreparedBatchCommitment {
            commitment,
            state_update,
            risk_update,
            command_mmr,
            obligation_mmr,
            intent_mmr,
        })
    }

    /// Apply a prepared batch commitment after the caller's durability boundary.
    pub fn apply_prepared_commit(
        &mut self,
        prepared: PreparedBatchCommitment,
    ) -> anyhow::Result<BatchCommitment> {
        if prepared.commitment.version != self.version {
            anyhow::bail!(
                "prepared commitment version {} does not match pipeline version {}",
                prepared.commitment.version,
                self.version
            );
        }

        if let Some(update) = prepared.state_update {
            self.state_store.write_node_batch(&update.node_batch)?;
        }
        if let Some(update) = prepared.risk_update {
            self.risk_store.write_node_batch(&update.node_batch)?;
        }
        self.command_mmr
            .apply_prepared_append(prepared.command_mmr)?;
        self.obligation_mmr
            .apply_prepared_append(prepared.obligation_mmr)?;
        self.intent_mmr.apply_prepared_append(prepared.intent_mmr)?;

        self.version += 1;
        Ok(prepared.commitment)
    }

    /// Commit a batch of state changes and return the new roots.
    ///
    /// This is the main entry point for callers that do not need to interpose
    /// another durability boundary between root computation and store mutation.
    pub fn commit(&mut self, delta: &StateDelta) -> anyhow::Result<BatchCommitment> {
        let prepared = self.prepare_commit(delta)?;
        self.apply_prepared_commit(prepared)
    }

    /// Append a command hash to the command MMR.
    pub fn append_command(&mut self, command_hash: [u8; 32]) -> anyhow::Result<()> {
        self.command_mmr.append(command_hash)
    }

    /// Append an obligation to the obligation MMR.
    pub fn append_obligation(&mut self, obligation_hash: [u8; 32]) -> anyhow::Result<()> {
        self.obligation_mmr.append(obligation_hash)
    }

    /// Append a user intent to the intent MMR.
    pub fn append_intent(&mut self, intent_hash: [u8; 32]) -> anyhow::Result<()> {
        self.intent_mmr.append(intent_hash)
    }

    fn latest_root(&self, store: &S, version: Version) -> anyhow::Result<RootHash> {
        if version == 0 {
            return Ok(RootHash([0u8; 32]));
        }
        let tree = new_jmt(store);
        match tree.get_root_hash_option(version - 1)? {
            Some(root) => Ok(root),
            None => Ok(RootHash([0u8; 32])),
        }
    }

    fn prepare_state_delta(
        &self,
        delta: &StateDelta,
        version: Version,
    ) -> anyhow::Result<(RootHash, usize, Option<TreeUpdateBatch>)> {
        let mut entries: Vec<(KeyHash, Option<Vec<u8>>)> = Vec::new();

        for u in &delta.accounts {
            entries.push((StateKey::account(&u.address), Some(leaf_to_bytes(&u.leaf))));
        }

        for u in &delta.option_positions {
            let key = StateKey::option_position(&u.address, &u.symbol);
            entries.push((key, u.leaf.as_ref().map(leaf_to_bytes)));
        }

        for u in &delta.perp_positions {
            let key = StateKey::perp_position(&u.address, &u.coin);
            entries.push((key, u.leaf.as_ref().map(leaf_to_bytes)));
        }

        for u in &delta.orders {
            let key = StateKey::order(&u.address, u.order_id);
            entries.push((key, u.leaf.as_ref().map(leaf_to_bytes)));
        }

        for u in &delta.instruments {
            entries.push((
                StateKey::instrument(&u.symbol),
                Some(leaf_to_bytes(&u.leaf)),
            ));
        }

        for u in &delta.oracles {
            entries.push((
                StateKey::oracle(&u.underlying),
                Some(leaf_to_bytes(&u.leaf)),
            ));
        }

        for u in &delta.mmp_configs {
            entries.push((
                StateKey::mmp_config(&u.address, &u.currency),
                Some(leaf_to_bytes(&u.leaf)),
            ));
        }

        if let Some(ref g) = delta.global {
            entries.push((StateKey::global(), Some(leaf_to_bytes(g))));
        }

        let count = entries.len();
        if count == 0 {
            return Ok((self.latest_root(&self.state_store, version)?, 0, None));
        }

        let tree = new_jmt(&self.state_store);
        let (root, batch) = tree.put_value_set(entries, version)?;

        Ok((root, count, Some(batch)))
    }

    fn prepare_risk_delta(
        &self,
        risk_updates: &[RiskUpdate],
        version: Version,
    ) -> anyhow::Result<(RootHash, usize, Option<TreeUpdateBatch>)> {
        if risk_updates.is_empty() {
            return Ok((self.latest_root(&self.risk_store, version)?, 0, None));
        }

        let entries: Vec<(KeyHash, Option<Vec<u8>>)> = risk_updates
            .iter()
            .map(|u| (StateKey::risk(&u.address), Some(leaf_to_bytes(&u.leaf))))
            .collect();

        let count = entries.len();
        let tree = new_jmt(&self.risk_store);
        let (root, batch) = tree.put_value_set(entries, version)?;

        Ok((root, count, Some(batch)))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::store::FastMemoryStore;

    fn test_pipeline() -> CommitmentPipeline<FastMemoryStore> {
        CommitmentPipeline::new(FastMemoryStore::new(), FastMemoryStore::new(), 100)
    }

    #[test]
    fn empty_commit_produces_deterministic_root() {
        let mut pipeline = test_pipeline();
        let c1 = pipeline.commit(&StateDelta::default()).unwrap();
        assert_eq!(c1.version, 0);
        assert_eq!(c1.state_leaves_updated, 0);
        assert_eq!(c1.risk_leaves_updated, 0);
    }

    #[test]
    fn single_account_commit() {
        let mut pipeline = test_pipeline();
        let delta = StateDelta {
            accounts: vec![AccountUpdate {
                address: [0x42; 20],
                leaf: AccountLeaf {
                    cash: 1_000_000 * PRICE_SCALE,
                    nonce: 0,
                    margin_mode: 0,
                    liquidation_state: 0,
                },
            }],
            ..Default::default()
        };

        let c = pipeline.commit(&delta).unwrap();
        assert_eq!(c.version, 0);
        assert_eq!(c.state_leaves_updated, 1);
        assert_ne!(c.state_root.0, [0u8; 32]);
        assert_ne!(c.batch_root, [0u8; 32]);
    }

    #[test]
    fn prepare_commit_does_not_mutate_until_apply() {
        let mut pipeline = test_pipeline();
        let delta = StateDelta {
            commands: vec![[0xAA; 32]],
            accounts: vec![AccountUpdate {
                address: [0x42; 20],
                leaf: AccountLeaf {
                    cash: 1_000_000 * PRICE_SCALE,
                    nonce: 0,
                    margin_mode: 0,
                    liquidation_state: 0,
                },
            }],
            ..Default::default()
        };

        let prepared = pipeline.prepare_commit(&delta).unwrap();
        assert_eq!(prepared.commitment.version, 0);
        assert_eq!(pipeline.version(), 0);
        assert_eq!(pipeline.command_mmr.size(), 0);
        assert!(
            new_jmt(&pipeline.state_store)
                .get_root_hash_option(0)
                .unwrap()
                .is_none(),
            "prepare must not write the state root before apply"
        );

        let applied = pipeline.apply_prepared_commit(prepared).unwrap();
        assert_eq!(applied.version, 0);
        assert_eq!(pipeline.version(), 1);
        assert_eq!(pipeline.command_mmr.size(), 1);
        assert_eq!(
            new_jmt(&pipeline.state_store)
                .get_root_hash_option(0)
                .unwrap(),
            Some(applied.state_root)
        );
    }

    #[test]
    fn version_increments() {
        let mut pipeline = test_pipeline();
        let delta = StateDelta {
            accounts: vec![AccountUpdate {
                address: [0x01; 20],
                leaf: AccountLeaf {
                    cash: 100,
                    nonce: 0,
                    margin_mode: 0,
                    liquidation_state: 0,
                },
            }],
            ..Default::default()
        };

        let c0 = pipeline.commit(&delta).unwrap();
        assert_eq!(c0.version, 0);
        assert_eq!(pipeline.version(), 1);

        let c1 = pipeline.commit(&delta).unwrap();
        assert_eq!(c1.version, 1);
        assert_eq!(pipeline.version(), 2);
    }

    #[test]
    fn state_root_changes_on_balance_update() {
        let mut pipeline = test_pipeline();
        let addr = [0x42; 20];

        let d1 = StateDelta {
            accounts: vec![AccountUpdate {
                address: addr,
                leaf: AccountLeaf {
                    cash: 1000,
                    nonce: 0,
                    margin_mode: 0,
                    liquidation_state: 0,
                },
            }],
            ..Default::default()
        };
        let c1 = pipeline.commit(&d1).unwrap();

        let d2 = StateDelta {
            accounts: vec![AccountUpdate {
                address: addr,
                leaf: AccountLeaf {
                    cash: 2000,
                    nonce: 1,
                    margin_mode: 0,
                    liquidation_state: 0,
                },
            }],
            ..Default::default()
        };
        let c2 = pipeline.commit(&d2).unwrap();

        assert_ne!(c1.state_root, c2.state_root);
        assert_ne!(c1.batch_root, c2.batch_root);
    }

    #[test]
    fn risk_root_changes_independently() {
        let mut pipeline = test_pipeline();
        let addr = [0x01; 20];

        let d1 = StateDelta {
            accounts: vec![AccountUpdate {
                address: addr,
                leaf: AccountLeaf {
                    cash: 1000,
                    nonce: 0,
                    margin_mode: 0,
                    liquidation_state: 0,
                },
            }],
            risk: vec![RiskUpdate {
                address: addr,
                leaf: RiskLeaf {
                    equity: 1000,
                    initial_margin: 500,
                    maintenance_margin: 250,
                    scanning_risk: 400,
                    health_nonce: 0,
                },
            }],
            ..Default::default()
        };
        let c1 = pipeline.commit(&d1).unwrap();

        let d2 = StateDelta {
            risk: vec![RiskUpdate {
                address: addr,
                leaf: RiskLeaf {
                    equity: 800,
                    initial_margin: 500,
                    maintenance_margin: 250,
                    scanning_risk: 400,
                    health_nonce: 1,
                },
            }],
            ..Default::default()
        };
        let c2 = pipeline.commit(&d2).unwrap();

        assert_eq!(c1.state_root, c2.state_root);
        assert_ne!(c1.risk_root, c2.risk_root);
        assert_ne!(c1.batch_root, c2.batch_root);
    }

    #[test]
    fn command_mmr_affects_batch_root() {
        let mut pipeline = test_pipeline();

        let delta = StateDelta {
            accounts: vec![AccountUpdate {
                address: [0x01; 20],
                leaf: AccountLeaf {
                    cash: 100,
                    nonce: 0,
                    margin_mode: 0,
                    liquidation_state: 0,
                },
            }],
            ..Default::default()
        };

        let c1 = pipeline.commit(&delta).unwrap();
        pipeline.append_command([0xAA; 32]).unwrap();
        let c2 = pipeline.commit(&delta).unwrap();

        assert_ne!(c1.batch_root, c2.batch_root);
    }

    #[test]
    fn full_trading_scenario() {
        let mut pipeline = test_pipeline();

        let alice = [0x01; 20];
        let bob = [0x02; 20];
        let symbol = "BTC-20261231-100000-C";

        // Batch 0: account setup + oracle
        let d0 = StateDelta {
            accounts: vec![
                AccountUpdate {
                    address: alice,
                    leaf: AccountLeaf {
                        cash: 100_000 * PRICE_SCALE,
                        nonce: 0,
                        margin_mode: 1,
                        liquidation_state: 0,
                    },
                },
                AccountUpdate {
                    address: bob,
                    leaf: AccountLeaf {
                        cash: 500_000 * PRICE_SCALE,
                        nonce: 0,
                        margin_mode: 1,
                        liquidation_state: 0,
                    },
                },
            ],
            instruments: vec![InstrumentUpdate {
                symbol: symbol.to_string(),
                leaf: InstrumentLeaf {
                    expired: false,
                    trading_mode: 2,
                    expiry: 1798761600,
                    strike: 100_000 * PRICE_SCALE,
                    option_type: 0,
                },
            }],
            oracles: vec![OracleUpdate {
                underlying: "BTC".to_string(),
                leaf: OracleLeaf {
                    spot_price: 95_000 * PRICE_SCALE,
                    iv_surface_hash: [0xAA; 32],
                },
            }],
            global: Some(GlobalLeaf {
                next_order_id: 1,
                next_trade_id: 1,
                command_chain_root: [0; 32],
                command_chain_seq: 0,
            }),
            risk: vec![
                RiskUpdate {
                    address: alice,
                    leaf: RiskLeaf {
                        equity: 100_000 * PRICE_SCALE,
                        initial_margin: 0,
                        maintenance_margin: 0,
                        scanning_risk: 0,
                        health_nonce: 0,
                    },
                },
                RiskUpdate {
                    address: bob,
                    leaf: RiskLeaf {
                        equity: 500_000 * PRICE_SCALE,
                        initial_margin: 0,
                        maintenance_margin: 0,
                        scanning_risk: 0,
                        health_nonce: 0,
                    },
                },
            ],
            ..Default::default()
        };
        pipeline.append_command([0x01; 32]).unwrap();
        let c0 = pipeline.commit(&d0).unwrap();
        assert_eq!(c0.state_leaves_updated, 5);
        assert_eq!(c0.risk_leaves_updated, 2);

        // Batch 1: trade fills, positions created
        let d1 = StateDelta {
            accounts: vec![
                AccountUpdate {
                    address: alice,
                    leaf: AccountLeaf {
                        cash: 95_000 * PRICE_SCALE,
                        nonce: 1,
                        margin_mode: 1,
                        liquidation_state: 0,
                    },
                },
                AccountUpdate {
                    address: bob,
                    leaf: AccountLeaf {
                        cash: 505_000 * PRICE_SCALE,
                        nonce: 1,
                        margin_mode: 1,
                        liquidation_state: 0,
                    },
                },
            ],
            option_positions: vec![
                OptionPositionUpdate {
                    address: alice,
                    symbol: symbol.to_string(),
                    leaf: Some(OptionPositionLeaf {
                        quantity: 10 * PRICE_SCALE,
                        entry_price: 5_000 * PRICE_SCALE,
                    }),
                },
                OptionPositionUpdate {
                    address: bob,
                    symbol: symbol.to_string(),
                    leaf: Some(OptionPositionLeaf {
                        quantity: -10 * PRICE_SCALE,
                        entry_price: 5_000 * PRICE_SCALE,
                    }),
                },
            ],
            global: Some(GlobalLeaf {
                next_order_id: 3,
                next_trade_id: 2,
                command_chain_root: [0x11; 32],
                command_chain_seq: 2,
            }),
            risk: vec![
                RiskUpdate {
                    address: alice,
                    leaf: RiskLeaf {
                        equity: 95_000 * PRICE_SCALE,
                        initial_margin: 20_000 * PRICE_SCALE,
                        maintenance_margin: 10_000 * PRICE_SCALE,
                        scanning_risk: 15_000 * PRICE_SCALE,
                        health_nonce: 1,
                    },
                },
                RiskUpdate {
                    address: bob,
                    leaf: RiskLeaf {
                        equity: 505_000 * PRICE_SCALE,
                        initial_margin: 50_000 * PRICE_SCALE,
                        maintenance_margin: 25_000 * PRICE_SCALE,
                        scanning_risk: 40_000 * PRICE_SCALE,
                        health_nonce: 1,
                    },
                },
            ],
            ..Default::default()
        };
        pipeline.append_command([0x02; 32]).unwrap();
        pipeline.append_command([0x03; 32]).unwrap();
        let c1 = pipeline.commit(&d1).unwrap();
        assert_ne!(c0.batch_root, c1.batch_root);
        assert_eq!(c1.state_leaves_updated, 5);

        // Batch 2: price move, risk update only
        let d2 = StateDelta {
            oracles: vec![OracleUpdate {
                underlying: "BTC".to_string(),
                leaf: OracleLeaf {
                    spot_price: 100_000 * PRICE_SCALE,
                    iv_surface_hash: [0xBB; 32],
                },
            }],
            risk: vec![
                RiskUpdate {
                    address: alice,
                    leaf: RiskLeaf {
                        equity: 110_000 * PRICE_SCALE,
                        initial_margin: 22_000 * PRICE_SCALE,
                        maintenance_margin: 11_000 * PRICE_SCALE,
                        scanning_risk: 16_000 * PRICE_SCALE,
                        health_nonce: 2,
                    },
                },
                RiskUpdate {
                    address: bob,
                    leaf: RiskLeaf {
                        equity: 490_000 * PRICE_SCALE,
                        initial_margin: 55_000 * PRICE_SCALE,
                        maintenance_margin: 27_500 * PRICE_SCALE,
                        scanning_risk: 44_000 * PRICE_SCALE,
                        health_nonce: 2,
                    },
                },
            ],
            ..Default::default()
        };
        let c2 = pipeline.commit(&d2).unwrap();
        assert_ne!(c1.batch_root, c2.batch_root);
        assert_eq!(pipeline.version(), 3);
    }

    #[test]
    fn position_delete_changes_root() {
        let mut pipeline = test_pipeline();
        let addr = [0x01; 20];

        let d1 = StateDelta {
            option_positions: vec![OptionPositionUpdate {
                address: addr,
                symbol: "BTC-C".to_string(),
                leaf: Some(OptionPositionLeaf {
                    quantity: 100,
                    entry_price: 5000,
                }),
            }],
            ..Default::default()
        };
        let c1 = pipeline.commit(&d1).unwrap();

        let d2 = StateDelta {
            option_positions: vec![OptionPositionUpdate {
                address: addr,
                symbol: "BTC-C".to_string(),
                leaf: None,
            }],
            ..Default::default()
        };
        let c2 = pipeline.commit(&d2).unwrap();

        assert_ne!(c1.state_root, c2.state_root);
    }
}