diff --git a/applications/tari_indexer/src/event_scanner.rs b/applications/tari_indexer/src/event_scanner.rs index 88cca187a..b75290b66 100644 --- a/applications/tari_indexer/src/event_scanner.rs +++ b/applications/tari_indexer/src/event_scanner.rs @@ -26,18 +26,16 @@ use anyhow::anyhow; use futures::StreamExt; use log::*; use tari_bor::decode; -use tari_common::configuration::Network; -use tari_consensus::consensus_constants::ConsensusConstants; -use tari_crypto::{ristretto::RistrettoPublicKey, tari_utilities::message_format::MessageFormat}; -use tari_dan_common_types::{committee::Committee, Epoch, NumPreshards, PeerAddress, ShardGroup}; +use tari_crypto::tari_utilities::message_format::MessageFormat; +use tari_dan_common_types::{committee::Committee, Epoch, PeerAddress, ShardGroup}; use tari_dan_p2p::proto::rpc::{GetTransactionResultRequest, PayloadResultStatus, SyncBlocksRequest}; -use tari_dan_storage::consensus_models::{Block, BlockError, BlockId, Decision, TransactionRecord}; +use tari_dan_storage::consensus_models::{Block, BlockId, Decision, TransactionRecord}; use tari_engine_types::{ commit_result::{ExecuteResult, TransactionResult}, events::Event, substate::{Substate, SubstateId, SubstateValue}, }; -use tari_epoch_manager::EpochManagerReader; +use tari_epoch_manager::{base_layer::EpochManagerHandle, EpochManagerReader}; use tari_template_lib::models::{EntityId, TemplateAddress}; use tari_transaction::{Transaction, TransactionId}; use tari_validator_node_rpc::client::{TariValidatorNodeRpcClientFactory, ValidatorNodeClientFactory}; @@ -96,33 +94,24 @@ struct TransactionMetadata { } pub struct EventScanner { - network: Network, - sidechain_id: Option, - epoch_manager: Box>, + epoch_manager: EpochManagerHandle, client_factory: TariValidatorNodeRpcClientFactory, substate_store: SqliteSubstateStore, event_filters: Vec, - consensus_constants: ConsensusConstants, } impl EventScanner { pub fn new( - network: Network, - sidechain_id: Option, - epoch_manager: Box>, + epoch_manager: EpochManagerHandle, client_factory: TariValidatorNodeRpcClientFactory, substate_store: SqliteSubstateStore, event_filters: Vec, - consensus_constants: ConsensusConstants, ) -> Self { Self { - network, - sidechain_id, epoch_manager, client_factory, substate_store, event_filters, - consensus_constants, } } @@ -151,7 +140,7 @@ impl EventScanner { }, None => { // by default we start scanning since the current epoch - // TODO: it would be nice a new parameter in the indexer to spcify a custom starting epoch + // TODO: it would be nice a new parameter in the indexer to specify a custom starting epoch event_count += self.scan_events_of_epoch(newest_epoch).await?; }, } @@ -454,12 +443,6 @@ impl EventScanner { .collect() } - fn build_genesis_block_id(&self, num_preshards: NumPreshards) -> Result { - // TODO: this should return the actual genesis for the shard group and epoch - let start_block = Block::zero_block(self.network, num_preshards, self.sidechain_id.clone())?; - Ok(*start_block.id()) - } - async fn get_oldest_scanned_epoch(&self) -> Result, anyhow::Error> { self.substate_store .with_read_tx(|tx| tx.get_oldest_scanned_epoch()) @@ -473,23 +456,18 @@ impl EventScanner { committee: &mut Committee, epoch: Epoch, ) -> Result, anyhow::Error> { - // We start scanning from the last scanned block for this commitee + // We start scanning from the last scanned block for this committee let start_block_id = self .substate_store .with_read_tx(|tx| tx.get_last_scanned_block_id(epoch, shard_group))?; - let start_block_id = match start_block_id { - Some(block_id) => block_id, - None => self.build_genesis_block_id(self.consensus_constants.num_preshards)?, - }; - committee.shuffle(); let mut last_block_id = start_block_id; info!( target: LOG_TARGET, - "Scanning new blocks since {} from (epoch={}, shard={})", - last_block_id, + "Scanning new blocks from (start_id={}, epoch={}, shard={})", + last_block_id.map(|id| id.to_string()).unwrap_or_else(|| "None".to_string()), epoch, shard_group ); @@ -502,7 +480,7 @@ impl EventScanner { epoch, shard_group ); - let resp = self.get_blocks_from_vn(member, start_block_id, Some(epoch)).await; + let resp = self.get_blocks_from_vn(member, last_block_id, epoch).await; match resp { Ok(blocks) => { @@ -520,9 +498,9 @@ impl EventScanner { let last_block = blocks.iter().max_by_key(|b| (b.epoch(), b.height())); if let Some(block) = last_block { - last_block_id = *block.id(); + last_block_id = Some(*block.id()); // Store the latest scanned block id in the database for future scans - self.save_scanned_block_id(epoch, shard_group, last_block_id)?; + self.save_scanned_block_id(epoch, shard_group, *block.id())?; } return Ok(blocks); }, @@ -578,8 +556,8 @@ impl EventScanner { async fn get_blocks_from_vn( &self, vn_addr: &PeerAddress, - start_block_id: BlockId, - up_to_epoch: Option, + start_block_id: Option, + up_to_epoch: Epoch, ) -> Result, anyhow::Error> { let mut blocks = vec![]; @@ -588,8 +566,8 @@ impl EventScanner { let mut stream = client .sync_blocks(SyncBlocksRequest { - start_block_id: start_block_id.as_bytes().to_vec(), - up_to_epoch: up_to_epoch.map(|epoch| epoch.into()), + start_block_id: start_block_id.map(|id| id.as_bytes().to_vec()).unwrap_or_default(), + epoch: Some(up_to_epoch.into()), }) .await?; while let Some(resp) = stream.next().await { diff --git a/applications/tari_indexer/src/lib.rs b/applications/tari_indexer/src/lib.rs index 290994c01..eb6c36cb5 100644 --- a/applications/tari_indexer/src/lib.rs +++ b/applications/tari_indexer/src/lib.rs @@ -172,16 +172,12 @@ pub async fn run_indexer(config: ApplicationConfig, mut shutdown_signal: Shutdow .map(TryInto::try_into) .collect::>() .map_err(|e| ExitError::new(ExitCode::ConfigError, format!("Invalid event filters: {}", e)))?; - let consensus_constants = ConsensusConstants::from(config.network); - let event_scanner = Arc::new(EventScanner::new( - config.network, - config.indexer.sidechain_id, - Box::new(services.epoch_manager.clone()), + let event_scanner = EventScanner::new( + services.epoch_manager.clone(), services.validator_node_client_factory.clone(), services.substate_store.clone(), event_filters, - consensus_constants, - )); + ); // Run the GraphQL API let graphql_address = config.indexer.graphql_address; diff --git a/applications/tari_indexer/src/substate_storage_sqlite/sqlite_substate_store_factory.rs b/applications/tari_indexer/src/substate_storage_sqlite/sqlite_substate_store_factory.rs index 2d94dd5e7..00c2c9ac4 100644 --- a/applications/tari_indexer/src/substate_storage_sqlite/sqlite_substate_store_factory.rs +++ b/applications/tari_indexer/src/substate_storage_sqlite/sqlite_substate_store_factory.rs @@ -847,7 +847,7 @@ impl SubstateStoreWriteTransaction for SqliteSubstateStoreWriteTransaction<'_> { use crate::substate_storage_sqlite::schema::scanned_block_ids; diesel::delete(scanned_block_ids::table) - .filter(scanned_block_ids::epoch.lt(epoch.0 as i64)) + .filter(scanned_block_ids::epoch.lt(epoch.as_u64() as i64)) .execute(&mut *self.connection()) .map_err(|e| StorageError::QueryError { reason: format!("delete_scanned_epochs_older_than: {}", e), diff --git a/applications/tari_swarm_daemon/webui/package-lock.json b/applications/tari_swarm_daemon/webui/package-lock.json index cc31416e8..955d57e7e 100644 --- a/applications/tari_swarm_daemon/webui/package-lock.json +++ b/applications/tari_swarm_daemon/webui/package-lock.json @@ -2650,9 +2650,9 @@ } }, "node_modules/rollup": { - "version": "3.29.4", - "resolved": "https://registry.npmjs.org/rollup/-/rollup-3.29.4.tgz", - "integrity": "sha512-oWzmBZwvYrU0iJHtDmhsm662rC15FRXmcjCk1xD771dFDx5jJ02ufAQQTn0etB2emNk4J9EZg/yWKpsn9BWGRw==", + "version": "3.29.5", + "resolved": "https://registry.npmjs.org/rollup/-/rollup-3.29.5.tgz", + "integrity": "sha512-GVsDdsbJzzy4S/v3dqWPJ7EfvZJfCHiDqe80IyrF59LYuP+e6U1LJoUqeuqRbwAWoMNoXivMNeNAOf5E22VA1w==", "dev": true, "bin": { "rollup": "dist/bin/rollup" diff --git a/applications/tari_validator_node/src/bootstrap.rs b/applications/tari_validator_node/src/bootstrap.rs index 2bb16a049..957e19da2 100644 --- a/applications/tari_validator_node/src/bootstrap.rs +++ b/applications/tari_validator_node/src/bootstrap.rs @@ -34,6 +34,7 @@ use tari_common::{ configuration::Network, exit_codes::{ExitCode, ExitError}, }; +use tari_common_types::types::FixedHash; use tari_consensus::consensus_constants::ConsensusConstants; #[cfg(not(feature = "metrics"))] use tari_consensus::traits::hooks::NoopHooks; @@ -602,8 +603,9 @@ where network, Epoch(0), ShardGroup::all_shards(num_preshards), + FixedHash::default(), sidechain_id.clone(), - )?; + ); let substate_id = substate_id.into(); let id = VersionedSubstateId::new(substate_id, 0); SubstateRecord { diff --git a/applications/tari_validator_node/src/p2p/rpc/block_sync_task.rs b/applications/tari_validator_node/src/p2p/rpc/block_sync_task.rs index 18171f008..fd43199dd 100644 --- a/applications/tari_validator_node/src/p2p/rpc/block_sync_task.rs +++ b/applications/tari_validator_node/src/p2p/rpc/block_sync_task.rs @@ -29,7 +29,7 @@ type BlockBuffer = Vec; pub struct BlockSyncTask { store: TStateStore, - start_block: Block, + start_block_id: BlockId, up_to_epoch: Option, sender: mpsc::Sender>, } @@ -37,13 +37,13 @@ pub struct BlockSyncTask { impl BlockSyncTask { pub fn new( store: TStateStore, - start_block: Block, + start_block_id: BlockId, up_to_epoch: Option, sender: mpsc::Sender>, ) -> Self { Self { store, - start_block, + start_block_id, up_to_epoch, sender, } @@ -51,7 +51,7 @@ impl BlockSyncTask { pub async fn run(mut self) -> Result<(), ()> { let mut buffer = Vec::with_capacity(BLOCK_BUFFER_SIZE); - let mut current_block_id = *self.start_block.id(); + let mut current_block_id = self.start_block_id; let mut counter = 0; loop { match self.fetch_next_batch(&mut buffer, ¤t_block_id) { diff --git a/applications/tari_validator_node/src/p2p/rpc/service_impl.rs b/applications/tari_validator_node/src/p2p/rpc/service_impl.rs index 07f2e1426..7094bb478 100644 --- a/applications/tari_validator_node/src/p2p/rpc/service_impl.rs +++ b/applications/tari_validator_node/src/p2p/rpc/service_impl.rs @@ -24,7 +24,7 @@ use std::convert::{TryFrom, TryInto}; use log::*; use tari_bor::{decode_exact, encode}; -use tari_dan_common_types::{optional::Optional, shard::Shard, Epoch, PeerAddress, SubstateAddress}; +use tari_dan_common_types::{optional::Optional, shard::Shard, Epoch, NodeHeight, PeerAddress, SubstateAddress}; use tari_dan_p2p::{ proto, proto::rpc::{ @@ -45,16 +45,7 @@ use tari_dan_p2p::{ }, }; use tari_dan_storage::{ - consensus_models::{ - Block, - BlockId, - EpochCheckpoint, - HighQc, - LockedBlock, - StateTransitionId, - SubstateRecord, - TransactionRecord, - }, + consensus_models::{Block, BlockId, EpochCheckpoint, HighQc, StateTransitionId, SubstateRecord, TransactionRecord}, StateStore, }; use tari_engine_types::virtual_substate::VirtualSubstateId; @@ -279,41 +270,53 @@ impl ValidatorNodeRpcService for ValidatorNodeRpcServiceImpl { .await .map_err(RpcStatus::log_internal_error(LOG_TARGET))?; - let (sender, receiver) = mpsc::channel(10); - - let start_block_id = BlockId::try_from(req.start_block_id) + let start_block_id = Some(req.start_block_id) + .filter(|i| !i.is_empty()) + .map(BlockId::try_from) + .transpose() .map_err(|e| RpcStatus::bad_request(format!("Invalid encoded block id: {}", e)))?; - // Check if we have the blocks - let start_block = store - .with_read_tx(|tx| Block::get(tx, &start_block_id).optional()) - .map_err(RpcStatus::log_internal_error(LOG_TARGET))? - .ok_or_else(|| RpcStatus::not_found(format!("start_block_id {start_block_id} not found")))?; - // Check that the start block is not after the locked block - let locked_block = store - .with_read_tx(|tx| LockedBlock::get(tx, current_epoch).optional()) - .map_err(RpcStatus::log_internal_error(LOG_TARGET))? - .ok_or_else(|| RpcStatus::not_found("No locked block"))?; - let epoch_is_after = start_block.epoch() > locked_block.epoch(); - let height_is_after = - (start_block.epoch() == locked_block.epoch) && (start_block.height() > locked_block.height()); - - if epoch_is_after || height_is_after { - return Err(RpcStatus::not_found(format!( - "start_block_id {} is after locked block {}", - start_block_id, locked_block - ))); - } + let start_block_id = { + let tx = store + .create_read_tx() + .map_err(RpcStatus::log_internal_error(LOG_TARGET))?; - task::spawn( - BlockSyncTask::new( - self.shard_state_store.clone(), - start_block, - req.up_to_epoch.map(|epoch| epoch.into()), - sender, - ) - .run(), - ); + match start_block_id { + Some(id) => { + if !Block::record_exists(&tx, &id).map_err(RpcStatus::log_internal_error(LOG_TARGET))? { + return Err(RpcStatus::not_found(format!("start_block_id {id} not found",))); + } + id + }, + None => { + let epoch = req + .epoch + .map(Epoch::from) + .map(|end| end.min(current_epoch)) + .unwrap_or(current_epoch); + + let mut block_ids = Block::get_ids_by_epoch_and_height(&tx, epoch, NodeHeight::zero()) + .map_err(RpcStatus::log_internal_error(LOG_TARGET))?; + + let Some(block_id) = block_ids.pop() else { + return Err(RpcStatus::not_found( + "Block not found with epoch={epoch},height={height}", + )); + }; + if !block_ids.is_empty() { + return Err(RpcStatus::conflict(format!( + "Multiple applicable blocks for epoch={} and height=0", + current_epoch + ))); + } + + block_id + }, + } + }; + + let (sender, receiver) = mpsc::channel(10); + task::spawn(BlockSyncTask::new(self.shard_state_store.clone(), start_block_id, None, sender).run()); Ok(Streaming::new(receiver)) } diff --git a/applications/tari_validator_node_web_ui/package-lock.json b/applications/tari_validator_node_web_ui/package-lock.json index 3131d21b1..cb3d48fbe 100644 --- a/applications/tari_validator_node_web_ui/package-lock.json +++ b/applications/tari_validator_node_web_ui/package-lock.json @@ -1860,9 +1860,9 @@ } }, "node_modules/rollup": { - "version": "3.29.4", - "resolved": "https://registry.npmjs.org/rollup/-/rollup-3.29.4.tgz", - "integrity": "sha512-oWzmBZwvYrU0iJHtDmhsm662rC15FRXmcjCk1xD771dFDx5jJ02ufAQQTn0etB2emNk4J9EZg/yWKpsn9BWGRw==", + "version": "3.29.5", + "resolved": "https://registry.npmjs.org/rollup/-/rollup-3.29.5.tgz", + "integrity": "sha512-GVsDdsbJzzy4S/v3dqWPJ7EfvZJfCHiDqe80IyrF59LYuP+e6U1LJoUqeuqRbwAWoMNoXivMNeNAOf5E22VA1w==", "dev": true, "bin": { "rollup": "dist/bin/rollup" diff --git a/applications/tari_validator_node_web_ui/src/routes/Blocks/BlockDetails.tsx b/applications/tari_validator_node_web_ui/src/routes/Blocks/BlockDetails.tsx index 7356583c9..09283970e 100644 --- a/applications/tari_validator_node_web_ui/src/routes/Blocks/BlockDetails.tsx +++ b/applications/tari_validator_node_web_ui/src/routes/Blocks/BlockDetails.tsx @@ -21,7 +21,7 @@ // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. import { useState, useEffect } from "react"; -import { useParams } from "react-router-dom"; +import { Link, useParams } from "react-router-dom"; import { Accordion, AccordionDetails, AccordionSummary } from "../../Components/Accordion"; import { Grid, Table, TableContainer, TableBody, TableRow, TableCell, Button, Fade, Alert } from "@mui/material"; import Typography from "@mui/material/Typography"; @@ -185,23 +185,23 @@ export default function BlockDetails() { Epoch - {block!.epoch} + {block!.header.epoch} Height - {block!.height} + {block!.header.height} Parent block - {block!.parent} + {block!.header.parent} Total Fees -
- {block!.total_leader_fee} +
+ {block!.header.total_leader_fee}
@@ -213,7 +213,7 @@ export default function BlockDetails() { Proposed by - {block!.proposed_by} + {block!.header.proposed_by} Block time diff --git a/applications/tari_validator_node_web_ui/src/routes/Blocks/Transactions.tsx b/applications/tari_validator_node_web_ui/src/routes/Blocks/Transactions.tsx index 91324b61c..f8c62432f 100644 --- a/applications/tari_validator_node_web_ui/src/routes/Blocks/Transactions.tsx +++ b/applications/tari_validator_node_web_ui/src/routes/Blocks/Transactions.tsx @@ -21,44 +21,45 @@ // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. import React from "react"; -import {Table, TableBody, TableCell, TableContainer, TableHead, TableRow} from "@mui/material"; +import { Table, TableBody, TableCell, TableContainer, TableHead, TableRow } from "@mui/material"; import StatusChip from "../../Components/StatusChip"; -import type {TransactionAtom} from "@tari-project/typescript-bindings"; +import type { TransactionAtom } from "@tari-project/typescript-bindings"; +import { Link } from "react-router-dom"; -function Transaction({transaction}: { transaction: TransactionAtom }) { - const decision = typeof transaction.decision === 'object' ? "Abort" : "Commit"; - return ( - - - {transaction.id} - - - - - {transaction.leader_fee?.fee} - {transaction.transaction_fee} - - ); +function Transaction({ transaction }: { transaction: TransactionAtom }) { + const decision = typeof transaction.decision === "object" ? "Abort" : "Commit"; + return ( + + + {transaction.id} + + + + + {transaction.leader_fee?.fee} + {transaction.transaction_fee} + + ); } -export default function Transactions({transactions}: { transactions: TransactionAtom[] }) { - return ( - - - - - Transaction ID - Decision - Leader fee - Transaction fee - - - - {transactions.map((tx: TransactionAtom) => ( - - ))} - -
- - ); +export default function Transactions({ transactions }: { transactions: TransactionAtom[] }) { + return ( + + + + + Transaction ID + Decision + Leader fee + Transaction fee + + + + {transactions.map((tx: TransactionAtom) => ( + + ))} + +
+ + ); } diff --git a/applications/tari_validator_node_web_ui/src/routes/Committees/CommitteesPieChart.tsx b/applications/tari_validator_node_web_ui/src/routes/Committees/CommitteesPieChart.tsx index 8cb8c6490..49f51269e 100644 --- a/applications/tari_validator_node_web_ui/src/routes/Committees/CommitteesPieChart.tsx +++ b/applications/tari_validator_node_web_ui/src/routes/Committees/CommitteesPieChart.tsx @@ -21,6 +21,7 @@ // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. import { useState, useEffect } from "react"; +import { Link } from "react-router-dom"; import EChartsReact from "echarts-for-react"; import "../../theme/echarts.css"; import type { @@ -73,7 +74,7 @@ const MyChartComponent = ({ chartData }: MyChartComponentProps) => { .map((item: any) => `
  • Address: ${item.address}
    • `) .slice(0, 5) .join(" ")}
    - View All Members
    `; + View All Members
    `; }; const option = { diff --git a/applications/tari_validator_node_web_ui/src/routes/Committees/CommitteesRadial.tsx b/applications/tari_validator_node_web_ui/src/routes/Committees/CommitteesRadial.tsx index 21ecf7aab..604c09491 100644 --- a/applications/tari_validator_node_web_ui/src/routes/Committees/CommitteesRadial.tsx +++ b/applications/tari_validator_node_web_ui/src/routes/Committees/CommitteesRadial.tsx @@ -21,6 +21,7 @@ // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. import { useState, useEffect } from "react"; +import { Link } from "react-router-dom"; import { fromHexString } from "../VN/Components/helpers"; import EChartsReact from "echarts-for-react"; import { ICommitteeChart } from "../../utils/interfaces"; @@ -113,7 +114,7 @@ export default function CommitteesRadial({ committees }: { committees: GetNetwor ${end}
    ${validators.length} Members:
      ${memberList}
    - View Committee`; + View Committee`; } const option = { @@ -141,7 +142,7 @@ export default function CommitteesRadial({ committees }: { committees: GetNetwor enterable: true, trigger: "axis", formatter: tooltipFormatter, - position: function (point: any) { + position: function(point: any) { const left = point[0] + 10; const top = point[1] - 10; return [left, top]; diff --git a/applications/tari_validator_node_web_ui/src/routes/VN/Components/Blocks.tsx b/applications/tari_validator_node_web_ui/src/routes/VN/Components/Blocks.tsx index f9654aa1c..6196e0f62 100644 --- a/applications/tari_validator_node_web_ui/src/routes/VN/Components/Blocks.tsx +++ b/applications/tari_validator_node_web_ui/src/routes/VN/Components/Blocks.tsx @@ -338,31 +338,31 @@ function Blocks() { {blocks.map((block) => { return ( - + - - {block.id.slice(0, 8)} + + {block.header.id.slice(0, 8)} - {block.epoch} - {block.height} + {block.header.epoch} + {block.header.height} {block.commands.length} -
    - {block.total_leader_fee} +
    + {block.header.total_leader_fee}
    {block.block_time} secs {primitiveDateTimeToDate(block.stored_at || []).toLocaleString()} -
    - {block.proposed_by.slice(0, 8)} +
    + {block.header.proposed_by.slice(0, 8)}
    diff --git a/bindings/dist/index.d.ts b/bindings/dist/index.d.ts index 7633f3b5b..f3d1f8ad3 100644 --- a/bindings/dist/index.d.ts +++ b/bindings/dist/index.d.ts @@ -5,6 +5,7 @@ export * from "./types/Amount"; export * from "./types/ArgDef"; export * from "./types/Arg"; export * from "./types/AuthHook"; +export * from "./types/BlockHeader"; export * from "./types/Block"; export * from "./types/BucketId"; export * from "./types/Claims"; @@ -27,6 +28,7 @@ export * from "./types/Decision"; export * from "./types/ElgamalVerifiableBalance"; export * from "./types/EntityId"; export * from "./types/Epoch"; +export * from "./types/Era"; export * from "./types/Event"; export * from "./types/Evidence"; export * from "./types/ExecutedTransaction"; diff --git a/bindings/dist/index.js b/bindings/dist/index.js index 59b1441ce..c93a6eb9d 100644 --- a/bindings/dist/index.js +++ b/bindings/dist/index.js @@ -7,6 +7,7 @@ export * from "./types/Amount"; export * from "./types/ArgDef"; export * from "./types/Arg"; export * from "./types/AuthHook"; +export * from "./types/BlockHeader"; export * from "./types/Block"; export * from "./types/BucketId"; export * from "./types/Claims"; @@ -29,6 +30,7 @@ export * from "./types/Decision"; export * from "./types/ElgamalVerifiableBalance"; export * from "./types/EntityId"; export * from "./types/Epoch"; +export * from "./types/Era"; export * from "./types/Event"; export * from "./types/Evidence"; export * from "./types/ExecutedTransaction"; diff --git a/bindings/dist/types/Block.d.ts b/bindings/dist/types/Block.d.ts index 5ac2b4aca..066444f2e 100644 --- a/bindings/dist/types/Block.d.ts +++ b/bindings/dist/types/Block.d.ts @@ -1,34 +1,12 @@ +import type { BlockHeader } from "./BlockHeader"; import type { Command } from "./Command"; -import type { Epoch } from "./Epoch"; -import type { ExtraData } from "./ExtraData"; -import type { NodeHeight } from "./NodeHeight"; import type { QuorumCertificate } from "./QuorumCertificate"; -import type { Shard } from "./Shard"; -import type { ShardGroup } from "./ShardGroup"; export interface Block { - id: string; - network: string; - parent: string; + header: BlockHeader; justify: QuorumCertificate; - height: NodeHeight; - epoch: Epoch; - shard_group: ShardGroup; - proposed_by: string; - total_leader_fee: number; - merkle_root: string; commands: Array; - is_dummy: boolean; is_justified: boolean; is_committed: boolean; - foreign_indexes: Record; - stored_at: Array | null; - signature: { - public_nonce: string; - signature: string; - } | null; block_time: number | null; - timestamp: number; - base_layer_block_height: number; - base_layer_block_hash: string; - extra_data: ExtraData | null; + stored_at: Array | null; } diff --git a/bindings/dist/types/BlockHeader.d.ts b/bindings/dist/types/BlockHeader.d.ts new file mode 100644 index 000000000..7d402f136 --- /dev/null +++ b/bindings/dist/types/BlockHeader.d.ts @@ -0,0 +1,28 @@ +import type { Epoch } from "./Epoch"; +import type { ExtraData } from "./ExtraData"; +import type { NodeHeight } from "./NodeHeight"; +import type { Shard } from "./Shard"; +import type { ShardGroup } from "./ShardGroup"; +export interface BlockHeader { + id: string; + network: string; + parent: string; + justify_id: string; + height: NodeHeight; + epoch: Epoch; + shard_group: ShardGroup; + proposed_by: string; + total_leader_fee: number; + state_merkle_root: string; + command_merkle_root: string; + is_dummy: boolean; + foreign_indexes: Record; + signature: { + public_nonce: string; + signature: string; + } | null; + timestamp: number; + base_layer_block_height: number; + base_layer_block_hash: string; + extra_data: ExtraData; +} diff --git a/bindings/dist/types/BlockHeader.js b/bindings/dist/types/BlockHeader.js new file mode 100644 index 000000000..cb0ff5c3b --- /dev/null +++ b/bindings/dist/types/BlockHeader.js @@ -0,0 +1 @@ +export {}; diff --git a/bindings/dist/types/Era.d.ts b/bindings/dist/types/Era.d.ts new file mode 100644 index 000000000..4948845b1 --- /dev/null +++ b/bindings/dist/types/Era.d.ts @@ -0,0 +1 @@ +export type Era = number; diff --git a/bindings/dist/types/Era.js b/bindings/dist/types/Era.js new file mode 100644 index 000000000..e5b481d1e --- /dev/null +++ b/bindings/dist/types/Era.js @@ -0,0 +1,2 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +export {}; diff --git a/bindings/src/index.ts b/bindings/src/index.ts index f80be792e..48f04cc35 100644 --- a/bindings/src/index.ts +++ b/bindings/src/index.ts @@ -8,6 +8,7 @@ export * from "./types/Amount"; export * from "./types/ArgDef"; export * from "./types/Arg"; export * from "./types/AuthHook"; +export * from "./types/BlockHeader"; export * from "./types/Block"; export * from "./types/BucketId"; export * from "./types/Claims"; @@ -30,6 +31,7 @@ export * from "./types/Decision"; export * from "./types/ElgamalVerifiableBalance"; export * from "./types/EntityId"; export * from "./types/Epoch"; +export * from "./types/Era"; export * from "./types/Event"; export * from "./types/Evidence"; export * from "./types/ExecutedTransaction"; diff --git a/bindings/src/types/Block.ts b/bindings/src/types/Block.ts index ef612eb0a..28cbff00e 100644 --- a/bindings/src/types/Block.ts +++ b/bindings/src/types/Block.ts @@ -1,33 +1,14 @@ // This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { BlockHeader } from "./BlockHeader"; import type { Command } from "./Command"; -import type { Epoch } from "./Epoch"; -import type { ExtraData } from "./ExtraData"; -import type { NodeHeight } from "./NodeHeight"; import type { QuorumCertificate } from "./QuorumCertificate"; -import type { Shard } from "./Shard"; -import type { ShardGroup } from "./ShardGroup"; export interface Block { - id: string; - network: string; - parent: string; + header: BlockHeader; justify: QuorumCertificate; - height: NodeHeight; - epoch: Epoch; - shard_group: ShardGroup; - proposed_by: string; - total_leader_fee: number; - merkle_root: string; commands: Array; - is_dummy: boolean; is_justified: boolean; is_committed: boolean; - foreign_indexes: Record; - stored_at: Array | null; - signature: { public_nonce: string; signature: string } | null; block_time: number | null; - timestamp: number; - base_layer_block_height: number; - base_layer_block_hash: string; - extra_data: ExtraData | null; + stored_at: Array | null; } diff --git a/bindings/src/types/BlockHeader.ts b/bindings/src/types/BlockHeader.ts new file mode 100644 index 000000000..4c31168af --- /dev/null +++ b/bindings/src/types/BlockHeader.ts @@ -0,0 +1,27 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. +import type { Epoch } from "./Epoch"; +import type { ExtraData } from "./ExtraData"; +import type { NodeHeight } from "./NodeHeight"; +import type { Shard } from "./Shard"; +import type { ShardGroup } from "./ShardGroup"; + +export interface BlockHeader { + id: string; + network: string; + parent: string; + justify_id: string; + height: NodeHeight; + epoch: Epoch; + shard_group: ShardGroup; + proposed_by: string; + total_leader_fee: number; + state_merkle_root: string; + command_merkle_root: string; + is_dummy: boolean; + foreign_indexes: Record; + signature: { public_nonce: string; signature: string } | null; + timestamp: number; + base_layer_block_height: number; + base_layer_block_hash: string; + extra_data: ExtraData; +} diff --git a/bindings/src/types/Era.ts b/bindings/src/types/Era.ts new file mode 100644 index 000000000..0880ef0bb --- /dev/null +++ b/bindings/src/types/Era.ts @@ -0,0 +1,3 @@ +// This file was generated by [ts-rs](https://github.com/Aleph-Alpha/ts-rs). Do not edit this file manually. + +export type Era = number; diff --git a/dan_layer/common_types/src/era.rs b/dan_layer/common_types/src/era.rs new file mode 100644 index 000000000..1cccafaab --- /dev/null +++ b/dan_layer/common_types/src/era.rs @@ -0,0 +1,86 @@ +// Copyright 2022. The Tari Project +// +// Redistribution and use in source and binary forms, with or without modification, are permitted provided that the +// following conditions are met: +// +// 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following +// disclaimer. +// +// 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the +// following disclaimer in the documentation and/or other materials provided with the distribution. +// +// 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote +// products derived from this software without specific prior written permission. +// +// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, +// INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +// WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE +// USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +use std::fmt::Display; + +use newtype_ops::newtype_ops; +use serde::{Deserialize, Serialize}; +#[cfg(feature = "ts")] +use ts_rs::TS; + +#[derive(Copy, Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash, Deserialize, Serialize)] +#[cfg_attr(feature = "ts", derive(TS), ts(export, export_to = "../../bindings/src/types/"))] +pub struct Era(#[cfg_attr(feature = "ts", ts(type = "number"))] pub u64); + +impl Era { + pub const fn zero() -> Self { + Self(0) + } + + pub const fn as_u64(self) -> u64 { + self.0 + } + + pub fn is_zero(&self) -> bool { + self.0 == 0 + } + + pub fn to_le_bytes(self) -> [u8; 8] { + self.0.to_le_bytes() + } + + pub fn saturating_sub>(&self, other: T) -> Self { + Self(self.0.saturating_sub(other.into().0)) + } + + pub fn checked_sub(&self, other: Self) -> Option { + self.0.checked_sub(other.0).map(Self) + } +} + +impl From for Era { + fn from(e: u64) -> Self { + Self(e) + } +} + +impl PartialEq for Era { + fn eq(&self, other: &u64) -> bool { + self.0 == *other + } +} + +impl PartialEq for u64 { + fn eq(&self, other: &Era) -> bool { + *self == other.0 + } +} + +impl Display for Era { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "Era({})", self.0) + } +} + +newtype_ops! { [Era] {add sub mul div} {:=} Self Self } +newtype_ops! { [Era] {add sub mul div} {:=} &Self &Self } +newtype_ops! { [Era] {add sub mul div} {:=} Self &Self } diff --git a/dan_layer/common_types/src/extra_data.rs b/dan_layer/common_types/src/extra_data.rs index 21b924bf0..19cc36e8a 100644 --- a/dan_layer/common_types/src/extra_data.rs +++ b/dan_layer/common_types/src/extra_data.rs @@ -23,11 +23,10 @@ use std::collections::BTreeMap; use serde::{Deserialize, Serialize}; -use tari_crypto::{ristretto::RistrettoPublicKey, tari_utilities::ByteArray}; #[cfg(feature = "ts")] use ts_rs::TS; -use crate::{MaxSizeBytes, MaxSizeBytesError}; +use crate::MaxSizeBytes; const MAX_DATA_SIZE: usize = 256; type ExtraFieldValue = MaxSizeBytes; @@ -47,18 +46,11 @@ impl ExtraData { Self(BTreeMap::new()) } - pub fn insert>(&mut self, key: ExtraFieldKey, value: V) -> &mut Self { - let value = value.into(); + pub fn insert(&mut self, key: ExtraFieldKey, value: ExtraFieldValue) -> &mut Self { self.0.insert(key, value); self } - pub fn insert_sidechain_id(&mut self, sidechain_id: RistrettoPublicKey) -> Result<&mut Self, MaxSizeBytesError> { - self.0 - .insert(ExtraFieldKey::SidechainId, sidechain_id.as_bytes().to_vec().try_into()?); - Ok(self) - } - pub fn get(&self, key: &ExtraFieldKey) -> Option<&ExtraFieldValue> { self.0.get(key) } diff --git a/dan_layer/common_types/src/hashing.rs b/dan_layer/common_types/src/hashing.rs index 98b22e9ce..6a78dbd19 100644 --- a/dan_layer/common_types/src/hashing.rs +++ b/dan_layer/common_types/src/hashing.rs @@ -36,6 +36,10 @@ pub fn block_hasher() -> TariHasher { dan_hasher("Block") } +pub fn command_hasher() -> TariHasher { + dan_hasher("Command") +} + pub fn quorum_certificate_hasher() -> TariHasher { dan_hasher("QuorumCertificate") } diff --git a/dan_layer/common_types/src/lib.rs b/dan_layer/common_types/src/lib.rs index 2437144a6..e99c066a8 100644 --- a/dan_layer/common_types/src/lib.rs +++ b/dan_layer/common_types/src/lib.rs @@ -9,6 +9,8 @@ pub mod crypto; mod epoch; pub use epoch::Epoch; +mod era; +pub use era::*; mod extra_data; pub use extra_data::{ExtraData, ExtraFieldKey}; @@ -50,4 +52,5 @@ mod versioned_substate_id; pub use versioned_substate_id::*; mod lock_intent; + pub use lock_intent::*; diff --git a/dan_layer/consensus/src/block_validations.rs b/dan_layer/consensus/src/block_validations.rs index 4f430e9f1..8d31c4ed9 100644 --- a/dan_layer/consensus/src/block_validations.rs +++ b/dan_layer/consensus/src/block_validations.rs @@ -30,6 +30,9 @@ pub fn check_proposal( // check_base_layer_block_hash::(block, epoch_manager, config).await?; check_network(block, config.network)?; check_sidechain_id(block, config)?; + if block.is_dummy() { + check_dummy(block)?; + } check_hash_and_height(block)?; check_proposed_by_leader(leader_strategy, committee_for_block, block)?; check_signature(block)?; @@ -37,6 +40,20 @@ pub fn check_proposal( Ok(()) } +pub fn check_dummy(candidate_block: &Block) -> Result<(), ProposalValidationError> { + if candidate_block.signature().is_some() { + return Err(ProposalValidationError::DummyBlockWithSignature { + block_id: *candidate_block.id(), + }); + } + if !candidate_block.commands().is_empty() { + return Err(ProposalValidationError::DummyBlockWithCommands { + block_id: *candidate_block.id(), + }); + } + Ok(()) +} + pub fn check_network(candidate_block: &Block, network: Network) -> Result<(), ProposalValidationError> { if candidate_block.network() != network { return Err(ProposalValidationError::InvalidNetwork { @@ -106,9 +123,9 @@ pub fn check_hash_and_height(candidate_block: &Block) -> Result<(), ProposalVali let calculated_hash = candidate_block.calculate_hash().into(); if calculated_hash != *candidate_block.id() { - return Err(ProposalValidationError::NodeHashMismatch { + return Err(ProposalValidationError::BlockIdMismatch { proposed_by: candidate_block.proposed_by().to_string(), - hash: *candidate_block.id(), + block_id: *candidate_block.id(), calculated_hash, }); } @@ -219,12 +236,7 @@ pub fn check_sidechain_id(candidate_block: &Block, config: &HotstuffConfig) -> R // If we are using a sidechain id in the network, we need to check it matches the candidate block one if let Some(expected_sidechain_id) = &config.sidechain_id { // Extract the sidechain id from the candidate block - let extra_data = candidate_block.extra_data().ok_or::( - ProposalValidationError::MissingSidechainId { - block_id: *candidate_block.id(), - } - .into(), - )?; + let extra_data = candidate_block.extra_data(); let sidechain_id_bytes = extra_data.get(&ExtraFieldKey::SidechainId).ok_or::( ProposalValidationError::InvalidSidechainId { block_id: *candidate_block.id(), diff --git a/dan_layer/consensus/src/consensus_constants.rs b/dan_layer/consensus/src/consensus_constants.rs index f671d6047..4eb090db7 100644 --- a/dan_layer/consensus/src/consensus_constants.rs +++ b/dan_layer/consensus/src/consensus_constants.rs @@ -23,7 +23,7 @@ use std::time::Duration; use tari_common::configuration::Network; -use tari_dan_common_types::NumPreshards; +use tari_dan_common_types::{Epoch, NumPreshards}; #[derive(Clone, Debug)] pub struct ConsensusConstants { @@ -33,16 +33,19 @@ pub struct ConsensusConstants { pub max_base_layer_blocks_behind: u64, pub num_preshards: NumPreshards, pub pacemaker_block_time: Duration, - /// The number of missed proposals before a SuspendNode proposal is sent + /// The number of missed proposals before a SuspendNode command is sent. pub missed_proposal_suspend_threshold: u64, - /// The maximum number of missed proposals to count. If a peer is offline, gets suspended and comes online, their - /// missed proposal count is decremented for each block that they participate in. Once this reaches zero, the node - /// is considered online and will be reinstated. This cap essentially gives the maximum number of rounds until they - /// will be reinstated once they resume participation in consensus. - pub missed_proposal_count_cap: u64, + /// The number of missed proposals before a EvictNode command is sent. + pub missed_proposal_evict_threshold: u64, + /// The number of rounds a node must participate in to be eligible for a ResumeNode command. If a peer is offline, + /// gets suspended and comes online, their missed proposal count is decremented for each block that they + /// participate (vote) in. Once this reaches zero, the node is considered online and will be reinstated. + pub missed_proposal_recovery_threshold: u64, + /// The maximum number of commands that a block may contain. pub max_block_size: usize, /// The value that fees are divided by to determine the amount of fees to burn. 0 means no fees are burned. pub fee_exhaust_divisor: u64, + pub epochs_per_era: Epoch, } impl ConsensusConstants { @@ -55,9 +58,11 @@ impl ConsensusConstants { num_preshards: NumPreshards::P256, pacemaker_block_time: Duration::from_secs(10), missed_proposal_suspend_threshold: 5, - missed_proposal_count_cap: 5, + missed_proposal_evict_threshold: 5, + missed_proposal_recovery_threshold: 5, max_block_size: 500, fee_exhaust_divisor: 20, // 5% + epochs_per_era: Epoch(10), } } } diff --git a/dan_layer/consensus/src/hotstuff/common.rs b/dan_layer/consensus/src/hotstuff/common.rs index 49154615b..855c40fe9 100644 --- a/dan_layer/consensus/src/hotstuff/common.rs +++ b/dan_layer/consensus/src/hotstuff/common.rs @@ -21,6 +21,7 @@ use tari_dan_common_types::{ use tari_dan_storage::{ consensus_models::{ Block, + BlockHeader, BlockId, EpochCheckpoint, LeafBlock, @@ -149,7 +150,7 @@ pub fn calculate_dummy_blocks_from_justify( break; } let (_, leader) = leader_strategy.get_leader(local_committee, current_height); - - let dummy_block = Block::dummy_block( + let dummy_header = BlockHeader::dummy_block( network, parent_block_id, leader.clone(), current_height, - qc.clone(), + *qc.id(), epoch, shard_group, parent_merkle_root, @@ -215,6 +215,7 @@ fn with_dummy_blocks( parent_base_layer_block_height, parent_base_layer_block_hash, ); + let dummy_block = Block::new(dummy_header, qc.clone(), Default::default()); debug!( target: LOG_TARGET, "🍼 new dummy block: {}", diff --git a/dan_layer/consensus/src/hotstuff/current_view.rs b/dan_layer/consensus/src/hotstuff/current_view.rs index efc441a50..666ccba55 100644 --- a/dan_layer/consensus/src/hotstuff/current_view.rs +++ b/dan_layer/consensus/src/hotstuff/current_view.rs @@ -50,6 +50,12 @@ impl CurrentView { is_updated } + // /// Sets the epoch to epoch + 1 + // pub(crate) fn next_epoch(&self) { + // let epoch = self.epoch.fetch_add(1, atomic::Ordering::SeqCst); + // info!(target: LOG_TARGET, "🧿 PACEMAKER SET EPOCH: {}", epoch + 1); + // } + /// Resets the height and epoch. Prefer update. pub(crate) fn reset(&self, epoch: Epoch, height: NodeHeight) { self.epoch.store(epoch.as_u64(), atomic::Ordering::SeqCst); diff --git a/dan_layer/consensus/src/hotstuff/error.rs b/dan_layer/consensus/src/hotstuff/error.rs index dcb4ef417..e940bec83 100644 --- a/dan_layer/consensus/src/hotstuff/error.rs +++ b/dan_layer/consensus/src/hotstuff/error.rs @@ -118,10 +118,10 @@ impl From for HotStuffError { pub enum ProposalValidationError { #[error("Storage error: {0}")] StorageError(#[from] StorageError), - #[error("Node proposed by {proposed_by} with hash {hash} does not match calculated hash {calculated_hash}")] - NodeHashMismatch { + #[error("Node proposed by {proposed_by} with ID {block_id} does not match calculated hash {calculated_hash}")] + BlockIdMismatch { proposed_by: String, - hash: BlockId, + block_id: BlockId, calculated_hash: BlockId, }, #[error("Node proposed by {proposed_by} with hash {hash} did not satisfy the safeNode predicate")] @@ -136,6 +136,12 @@ pub enum ProposalValidationError { }, #[error("Node proposed by {proposed_by} with hash {hash} is the genesis block")] ProposingGenesisBlock { proposed_by: String, hash: BlockId }, + #[error("Parent {parent_id} not found in block {block_id} proposed by {proposed_by}")] + ParentNotFound { + proposed_by: String, + parent_id: BlockId, + block_id: BlockId, + }, #[error("Justified block {justify_block} for proposed block {block_description} by {proposed_by} not found")] JustifyBlockNotFound { proposed_by: String, @@ -251,4 +257,8 @@ pub enum ProposalValidationError { block_epoch: Epoch, current_epoch: Epoch, }, + #[error("Dummy block {block_id} includes a signature")] + DummyBlockWithSignature { block_id: BlockId }, + #[error("Dummy block {block_id} includes commands")] + DummyBlockWithCommands { block_id: BlockId }, } diff --git a/dan_layer/consensus/src/hotstuff/mod.rs b/dan_layer/consensus/src/hotstuff/mod.rs index 34e0d49e7..2faec1870 100644 --- a/dan_layer/consensus/src/hotstuff/mod.rs +++ b/dan_layer/consensus/src/hotstuff/mod.rs @@ -6,6 +6,7 @@ mod current_view; mod error; mod event; mod on_beat; +mod on_catch_up_sync_request; mod on_force_beat; mod on_inbound_message; mod on_leader_timeout; @@ -18,7 +19,6 @@ mod on_receive_new_transaction; mod on_receive_new_view; mod on_receive_request_missing_transactions; mod on_receive_vote; -mod on_sync_request; // mod on_sync_response; mod block_change_set; mod foreign_proposal_processor; diff --git a/dan_layer/consensus/src/hotstuff/on_sync_request.rs b/dan_layer/consensus/src/hotstuff/on_catch_up_sync_request.rs similarity index 98% rename from dan_layer/consensus/src/hotstuff/on_sync_request.rs rename to dan_layer/consensus/src/hotstuff/on_catch_up_sync_request.rs index b274db9c2..be30e05e0 100644 --- a/dan_layer/consensus/src/hotstuff/on_sync_request.rs +++ b/dan_layer/consensus/src/hotstuff/on_catch_up_sync_request.rs @@ -90,9 +90,10 @@ impl OnSyncRequest { tx, leaf_block.epoch(), local_committee_info.shard_group(), - msg.high_qc.block_id(), - leaf_block.block_id(), + msg.high_qc.block_height(), + leaf_block.height(), true, + 1000, )?; Ok::<_, HotStuffError>(blocks) diff --git a/dan_layer/consensus/src/hotstuff/on_inbound_message.rs b/dan_layer/consensus/src/hotstuff/on_inbound_message.rs index cf9e15252..c59ba1b73 100644 --- a/dan_layer/consensus/src/hotstuff/on_inbound_message.rs +++ b/dan_layer/consensus/src/hotstuff/on_inbound_message.rs @@ -152,6 +152,9 @@ fn msg_epoch_and_height(msg: &HotstuffMessage) -> Option { HotstuffMessage::Proposal(msg) => Some((msg.block.epoch(), msg.block.height())), // Votes for block v occur in view v + 1 HotstuffMessage::Vote(msg) => Some((msg.epoch, msg.unverified_block_height.saturating_add(NodeHeight(1)))), + // We will buffer NEWVIEW messages until the appropriate height is set. This essentially prevents us from being + // forced to the next height without locally deciding to do so. + HotstuffMessage::NewView(msg) => Some((msg.high_qc.epoch(), msg.new_height.saturating_add(NodeHeight(1)))), _ => None, } } diff --git a/dan_layer/consensus/src/hotstuff/on_propose.rs b/dan_layer/consensus/src/hotstuff/on_propose.rs index 0811f1705..e39114a58 100644 --- a/dan_layer/consensus/src/hotstuff/on_propose.rs +++ b/dan_layer/consensus/src/hotstuff/on_propose.rs @@ -15,6 +15,7 @@ use tari_dan_common_types::{ optional::Optional, shard::Shard, Epoch, + ExtraData, NodeHeight, ToSubstateAddress, VersionedSubstateId, @@ -641,7 +642,7 @@ where TConsensusSpec: ConsensusSpec // Ensure that foreign indexes are canonically ordered foreign_indexes.sort_keys(); - let mut next_block = Block::new( + let mut next_block = Block::create( self.config.network, *parent_block.block_id(), high_qc_certificate, @@ -657,8 +658,8 @@ where TConsensusSpec: ConsensusSpec EpochTime::now().as_u64(), base_layer_block_height, base_layer_block_hash, - None, - ); + ExtraData::new(), + )?; let signature = self.signing_service.sign(next_block.id()); next_block.set_signature(signature); diff --git a/dan_layer/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs b/dan_layer/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs index 9e17eb5c0..c43261255 100644 --- a/dan_layer/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs +++ b/dan_layer/consensus/src/hotstuff/on_ready_to_vote_on_local_block.rs @@ -566,15 +566,15 @@ where TConsensusSpec: ConsensusSpec // Calculate for local shards only .filter(|ch| block.shard_group().contains(&ch.shard())), )?; - if expected_merkle_root != *block.merkle_root() { + if expected_merkle_root != *block.state_merkle_root() { warn!( target: LOG_TARGET, "❌ Merkle root disagreement for block {}. Leader proposed {}, we calculated {}", block, - block.merkle_root(), + block.state_merkle_root(), expected_merkle_root ); - proposed_block_change_set.no_vote(NoVoteReason::MerkleRootMismatch); + proposed_block_change_set.no_vote(NoVoteReason::StateMerkleRootMismatch); return Ok(()); } @@ -1752,7 +1752,10 @@ where TConsensusSpec: ConsensusSpec local_committee_info: &CommitteeInfo, ) -> Result, HotStuffError> { if block.is_dummy() { - block.increment_leader_failure_count(tx, self.config.consensus_constants.missed_proposal_count_cap)?; + block.increment_leader_failure_count( + tx, + self.config.consensus_constants.missed_proposal_recovery_threshold, + )?; // Nothing to do here for empty dummy blocks. Just mark the block as committed. block.commit_diff(tx, BlockDiff::empty(*block.id()))?; @@ -1803,7 +1806,7 @@ where TConsensusSpec: ConsensusSpec ); } - let total_transaction_fee = block.total_transaction_fee(); + let total_transaction_fee = block.calculate_total_transaction_fee(); if total_transaction_fee > 0 { info!( target: LOG_TARGET, diff --git a/dan_layer/consensus/src/hotstuff/on_receive_foreign_proposal.rs b/dan_layer/consensus/src/hotstuff/on_receive_foreign_proposal.rs index ddd5cd6ee..8ecde5b9a 100644 --- a/dan_layer/consensus/src/hotstuff/on_receive_foreign_proposal.rs +++ b/dan_layer/consensus/src/hotstuff/on_receive_foreign_proposal.rs @@ -158,9 +158,9 @@ where TConsensusSpec: ConsensusSpec let calculated_hash = candidate_block.calculate_hash().into(); if calculated_hash != *candidate_block.id() { - return Err(ProposalValidationError::NodeHashMismatch { + return Err(ProposalValidationError::BlockIdMismatch { proposed_by: candidate_block.proposed_by().to_string(), - hash: *candidate_block.id(), + block_id: *candidate_block.id(), calculated_hash, }); } diff --git a/dan_layer/consensus/src/hotstuff/on_receive_local_proposal.rs b/dan_layer/consensus/src/hotstuff/on_receive_local_proposal.rs index 5f91f5781..193850be7 100644 --- a/dan_layer/consensus/src/hotstuff/on_receive_local_proposal.rs +++ b/dan_layer/consensus/src/hotstuff/on_receive_local_proposal.rs @@ -15,7 +15,6 @@ use tari_dan_common_types::{ use tari_dan_storage::{ consensus_models::{ Block, - ForeignProposal, HighQc, LastSentVote, QuorumCertificate, @@ -174,6 +173,19 @@ impl OnReceiveLocalProposalHandler= REQUEST_FOREIGN_PROPOSAL_TIMEOUT + // - Request foreign proposal from remote shard group [END] + // + // Given a transaction that is awaiting a foreign proposal for MISSING_FOREIGN_PROPOSAL_TIMEOUT (e.g. 100) + // blocks + // - Load pending transactions that are awaiting foreign proposal for >= MISSING_FOREIGN_PROPOSAL_TIMEOUT + // - Set abort and ready = true + // self.update_foreign_proposal_transactions(tx, valid_block.block())?; + for foreign_proposal in foreign_proposals { if foreign_proposal.exists(&**tx)? { // This is expected behaviour, we may receive the same foreign proposal multiple times @@ -334,8 +346,9 @@ impl OnReceiveLocalProposalHandler OnReceiveLocalProposalHandler Result, HotStuffError> { let result = self.validate_local_proposed_block(tx, current_epoch, block, local_committee, local_committee_info); - // .and_then(|valid_block| { - // self.update_foreign_proposal_transactions(tx, valid_block.block())?; - // Ok(valid_block) - // }); match result { Ok(validated) => Ok(Some(validated)), @@ -696,6 +705,15 @@ impl OnReceiveLocalProposalHandler( Ok(()) } -fn cleanup_epoch(tx: &mut TTx, epoch: Epoch) -> Result<(), HotStuffError> { +fn cleanup_epoch(tx: &mut TTx, _epoch: Epoch) -> Result<(), HotStuffError> { Vote::delete_all(tx)?; - ForeignProposal::delete_in_epoch(tx, epoch)?; + // ForeignProposal::delete_in_epoch(tx, epoch)?; Ok(()) } diff --git a/dan_layer/consensus/src/hotstuff/substate_store/sharded_state_tree.rs b/dan_layer/consensus/src/hotstuff/substate_store/sharded_state_tree.rs index c451aeff9..42c7cb74f 100644 --- a/dan_layer/consensus/src/hotstuff/substate_store/sharded_state_tree.rs +++ b/dan_layer/consensus/src/hotstuff/substate_store/sharded_state_tree.rs @@ -139,7 +139,8 @@ impl ShardedStateTree<&TTx> { }, }; } - root_tree.put_root_hash_changes(None, 1, hashes) + let (hash, _) = root_tree.compute_update_batch(None, 1, hashes)?; + Ok(hash) } fn get_state_root_for_shard(&self, shard: Shard) -> Result { diff --git a/dan_layer/consensus/src/hotstuff/worker.rs b/dan_layer/consensus/src/hotstuff/worker.rs index e1ff6fa29..fec9201d3 100644 --- a/dan_layer/consensus/src/hotstuff/worker.rs +++ b/dan_layer/consensus/src/hotstuff/worker.rs @@ -9,6 +9,7 @@ use std::{ use log::*; use tari_dan_common_types::{ committee::{Committee, CommitteeInfo}, + optional::Optional, Epoch, NodeHeight, ShardGroup, @@ -18,6 +19,7 @@ use tari_dan_storage::{ Block, BlockDiff, BurntUtxo, + EpochCheckpoint, ForeignProposal, HighQc, LeafBlock, @@ -42,6 +44,7 @@ use crate::{ error::HotStuffError, event::HotstuffEvent, on_catch_up_sync::OnCatchUpSync, + on_catch_up_sync_request::OnSyncRequest, on_inbound_message::OnInboundMessage, on_message_validate::{MessageValidationResult, OnMessageValidate}, on_next_sync_view::OnNextSyncViewHandler, @@ -51,7 +54,6 @@ use crate::{ on_receive_new_view::OnReceiveNewViewHandler, on_receive_request_missing_transactions::OnReceiveRequestMissingTransactions, on_receive_vote::OnReceiveVoteHandler, - on_sync_request::OnSyncRequest, pacemaker::PaceMaker, pacemaker_handle::PaceMakerHandle, transaction_manager::ConsensusTransactionManager, @@ -218,7 +220,7 @@ impl HotstuffWorker { let current_epoch = self.epoch_manager.current_epoch().await?; let local_committee_info = self.epoch_manager.get_local_committee_info(current_epoch).await?; - self.create_zero_block_if_required(current_epoch, local_committee_info.shard_group())?; + self.create_genesis_block_if_required(current_epoch, local_committee_info.shard_group())?; // Resume pacemaker from the last epoch/height let (current_height, high_qc) = self.state_store.with_read_tx(|tx| { @@ -749,7 +751,7 @@ impl HotstuffWorker { block.shard_group(), *block.id(), &high_qc, - *block.merkle_root(), + *block.state_merkle_root(), &self.leader_strategy, local_committee, block.timestamp(), @@ -873,24 +875,21 @@ impl HotstuffWorker { } } - fn create_zero_block_if_required(&self, epoch: Epoch, shard_group: ShardGroup) -> Result<(), HotStuffError> { + fn create_genesis_block_if_required(&self, epoch: Epoch, shard_group: ShardGroup) -> Result<(), HotStuffError> { self.state_store.with_write_tx(|tx| { + let previous_epoch = epoch.saturating_sub(Epoch(1)); + let checkpoint = EpochCheckpoint::get(&**tx, previous_epoch).optional()?; + let state_merkle_root = checkpoint + .map(|cp| cp.compute_state_merkle_root()) + .transpose()? + .unwrap_or_default(); // The parent for genesis blocks refer to this zero block - let mut zero_block = Block::zero_block( - self.config.network, - self.config.consensus_constants.num_preshards, - self.config.sidechain_id.clone(), - )?; + let mut zero_block = Block::zero_block(self.config.network, self.config.consensus_constants.num_preshards); if !zero_block.exists(&**tx)? { debug!(target: LOG_TARGET, "Creating zero block"); zero_block.justify().insert(tx)?; zero_block.insert(tx)?; zero_block.set_as_justified(tx)?; - zero_block.as_locked_block().set(tx)?; - // zero_block.as_leaf_block().set(tx)?; - zero_block.as_last_executed().set(tx)?; - zero_block.as_last_voted().set(tx)?; - zero_block.justify().as_high_qc().set(tx)?; zero_block.commit_diff(tx, BlockDiff::empty(*zero_block.id()))?; } @@ -898,11 +897,12 @@ impl HotstuffWorker { self.config.network, epoch, shard_group, + state_merkle_root, self.config.sidechain_id.clone(), - )?; + ); if !genesis.exists(&**tx)? { info!(target: LOG_TARGET, "✨Creating genesis block {genesis}"); - genesis.justify().insert(tx)?; + genesis.justify().save(tx)?; genesis.insert(tx)?; genesis.set_as_justified(tx)?; genesis.as_locked_block().set(tx)?; diff --git a/dan_layer/consensus_tests/src/support/harness.rs b/dan_layer/consensus_tests/src/support/harness.rs index f4533cff5..b10c647a7 100644 --- a/dan_layer/consensus_tests/src/support/harness.rs +++ b/dan_layer/consensus_tests/src/support/harness.rs @@ -533,10 +533,12 @@ impl TestBuilder { num_preshards: TEST_NUM_PRESHARDS, pacemaker_block_time: Duration::from_secs(10), missed_proposal_suspend_threshold: 5, - missed_proposal_count_cap: 5, + missed_proposal_evict_threshold: 10, + missed_proposal_recovery_threshold: 5, max_block_size: 500, fee_exhaust_divisor: 20, max_vns_per_epoch_activated: 5, + epochs_per_era: Epoch(10), }, }, } diff --git a/dan_layer/engine/src/runtime/tracker.rs b/dan_layer/engine/src/runtime/tracker.rs index 11676d60e..96933d0a8 100644 --- a/dan_layer/engine/src/runtime/tracker.rs +++ b/dan_layer/engine/src/runtime/tracker.rs @@ -218,7 +218,7 @@ impl StateTracker { self.write_with(|state| { // If substates used in args are in scope for the current frame, we can bring then into scope for the new // frame - debug!( + trace!( target: LOG_TARGET, "CALL FRAME before:\n{}", state.current_call_scope()?, @@ -226,8 +226,7 @@ impl StateTracker { state.check_all_substates_in_scope(push_frame.arg_scope())?; let new_frame = push_frame.into_new_call_frame(); - debug!(target: LOG_TARGET, - "NEW CALL FRAME:\n{}", new_frame.scope()); + trace!(target: LOG_TARGET, "NEW CALL FRAME:\n{}", new_frame.scope()); state.push_frame(new_frame, max_call_depth) }) diff --git a/dan_layer/p2p/proto/consensus.proto b/dan_layer/p2p/proto/consensus.proto index 572606304..89ca702c8 100644 --- a/dan_layer/p2p/proto/consensus.proto +++ b/dan_layer/p2p/proto/consensus.proto @@ -81,24 +81,28 @@ message VoteMessage { tari.dan.common.SignatureAndPublicKey signature = 5; } -message Block { +message BlockHeader { bytes parent_id = 1; int32 network = 2; - QuorumCertificate justify = 3; - uint64 height = 4; - uint64 epoch = 5; - uint32 shard_group = 6; - bytes proposed_by = 7; - bytes merkle_root = 8; - repeated Command commands = 9; - uint64 total_leader_fee = 10; - bytes foreign_indexes = 11; - tari.dan.common.Signature signature = 12; - uint64 timestamp = 13; - uint64 base_layer_block_height = 14; - bytes base_layer_block_hash = 15; - bool is_dummy = 16; - ExtraData extra_data = 17; + uint64 height = 3; + uint64 epoch = 4; + uint32 shard_group = 5; + bytes proposed_by = 6; + bytes state_merkle_root = 7; + uint64 total_leader_fee = 8; + bytes foreign_indexes = 9; + tari.dan.common.Signature signature = 10; + uint64 timestamp = 11; + uint64 base_layer_block_height = 12; + bytes base_layer_block_hash = 13; + bool is_dummy = 14; + ExtraData extra_data = 15; +} + +message Block { + BlockHeader header = 1; + QuorumCertificate justify = 2; + repeated Command commands = 3; } message ExtraData { diff --git a/dan_layer/p2p/proto/rpc.proto b/dan_layer/p2p/proto/rpc.proto index 342dc64d0..5cdb7afda 100644 --- a/dan_layer/p2p/proto/rpc.proto +++ b/dan_layer/p2p/proto/rpc.proto @@ -192,8 +192,11 @@ message SubstateDestroyedProof { } message SyncBlocksRequest { + // Optional (empty for None). Must be provided if the epoch is not provided bytes start_block_id = 1; - tari.dan.common.Epoch up_to_epoch = 2; + // Optional - If start_block_id is provided, this is ignored. Must be provided if start_block_id is not provided. + // In which case, start block is implicitly the first block of the epoch. + tari.dan.common.Epoch epoch = 2; } message SyncBlocksResponse { diff --git a/dan_layer/p2p/src/conversions/consensus.rs b/dan_layer/p2p/src/conversions/consensus.rs index 598ac0b87..1f5adbe3b 100644 --- a/dan_layer/p2p/src/conversions/consensus.rs +++ b/dan_layer/p2p/src/conversions/consensus.rs @@ -20,7 +20,10 @@ // WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. -use std::convert::{TryFrom, TryInto}; +use std::{ + collections::BTreeSet, + convert::{TryFrom, TryInto}, +}; use anyhow::anyhow; use tari_bor::{decode_exact, encode}; @@ -48,27 +51,30 @@ use tari_dan_common_types::{ ValidatorMetadata, VersionedSubstateId, }; -use tari_dan_storage::consensus_models::{ - AbortReason, - BlockId, - Command, - Decision, - Evidence, - ForeignProposal, - ForeignProposalAtom, - HighQc, - LeaderFee, - MintConfidentialOutputAtom, - QcId, - QuorumCertificate, - QuorumDecision, - ResumeNodeAtom, - SubstateDestroyed, - SubstatePledge, - SubstatePledges, - SubstateRecord, - SuspendNodeAtom, - TransactionAtom, +use tari_dan_storage::{ + consensus_models, + consensus_models::{ + AbortReason, + BlockId, + Command, + Decision, + Evidence, + ForeignProposal, + ForeignProposalAtom, + HighQc, + LeaderFee, + MintConfidentialOutputAtom, + QcId, + QuorumCertificate, + QuorumDecision, + ResumeNodeAtom, + SubstateDestroyed, + SubstatePledge, + SubstatePledges, + SubstateRecord, + SuspendNodeAtom, + TransactionAtom, + }, }; use tari_engine_types::substate::{SubstateId, SubstateValue}; use tari_transaction::TransactionId; @@ -447,10 +453,9 @@ impl TryFrom for MissingTransacti }) } } -//---------------------------------- Block --------------------------------------------// -impl From<&tari_dan_storage::consensus_models::Block> for proto::consensus::Block { - fn from(value: &tari_dan_storage::consensus_models::Block) -> Self { +impl From<&consensus_models::BlockHeader> for proto::consensus::BlockHeader { + fn from(value: &consensus_models::BlockHeader) -> Self { Self { network: value.network().as_byte().into(), height: value.height().as_u64(), @@ -458,79 +463,109 @@ impl From<&tari_dan_storage::consensus_models::Block> for proto::consensus::Bloc shard_group: value.shard_group().encode_as_u32(), parent_id: value.parent().as_bytes().to_vec(), proposed_by: ByteArray::as_bytes(value.proposed_by()).to_vec(), - merkle_root: value.merkle_root().as_slice().to_vec(), - justify: Some(value.justify().into()), + state_merkle_root: value.state_merkle_root().as_slice().to_vec(), total_leader_fee: value.total_leader_fee(), - commands: value.commands().iter().map(Into::into).collect(), foreign_indexes: encode(value.foreign_indexes()).unwrap(), signature: value.signature().map(Into::into), timestamp: value.timestamp(), base_layer_block_height: value.base_layer_block_height(), base_layer_block_hash: value.base_layer_block_hash().as_bytes().to_vec(), is_dummy: value.is_dummy(), - extra_data: value.extra_data().map(Into::into), + extra_data: Some(value.extra_data().into()), + } + } +} + +fn try_convert_proto_block_header( + value: proto::consensus::BlockHeader, + justify_id: QcId, + commands: &BTreeSet, +) -> Result { + let network = u8::try_from(value.network) + .map_err(|_| anyhow!("Block conversion: Invalid network byte {}", value.network))? + .try_into()?; + + let shard_group = ShardGroup::decode_from_u32(value.shard_group) + .ok_or_else(|| anyhow!("Block shard_group ({}) is not a valid", value.shard_group))?; + + let proposed_by = PublicKey::from_canonical_bytes(&value.proposed_by) + .map_err(|_| anyhow!("Block conversion: Invalid proposed_by"))?; + + let extra_data = value + .extra_data + .ok_or_else(|| anyhow!("ExtraData not provided"))? + .try_into()?; + + if value.is_dummy { + Ok(consensus_models::BlockHeader::dummy_block( + network, + value.parent_id.try_into()?, + proposed_by, + NodeHeight(value.height), + justify_id, + Epoch(value.epoch), + shard_group, + value.state_merkle_root.try_into()?, + value.timestamp, + value.base_layer_block_height, + value.base_layer_block_hash.try_into()?, + )) + } else { + // We calculate the BlockId and command MR locally from remote data. This means that they will + // always be valid, therefore do not need to be explicitly validated. + // If there were a mismatch (perhaps due modified data over the wire) the signature verification will fail. + Ok(consensus_models::BlockHeader::create( + network, + value.parent_id.try_into()?, + justify_id, + NodeHeight(value.height), + Epoch(value.epoch), + shard_group, + proposed_by, + value.state_merkle_root.try_into()?, + commands, + value.total_leader_fee, + decode_exact(&value.foreign_indexes)?, + value.signature.map(TryInto::try_into).transpose()?, + value.timestamp, + value.base_layer_block_height, + value.base_layer_block_hash.try_into()?, + extra_data, + )?) + } +} + +//---------------------------------- Block --------------------------------------------// + +impl From<&consensus_models::Block> for proto::consensus::Block { + fn from(value: &consensus_models::Block) -> Self { + Self { + header: Some(value.header().into()), + justify: Some(value.justify().into()), + commands: value.commands().iter().map(Into::into).collect(), } } } -impl TryFrom for tari_dan_storage::consensus_models::Block { +impl TryFrom for consensus_models::Block { type Error = anyhow::Error; fn try_from(value: proto::consensus::Block) -> Result { - let network = u8::try_from(value.network) - .map_err(|_| anyhow!("Block conversion: Invalid network byte {}", value.network))? - .try_into()?; - - let shard_group = ShardGroup::decode_from_u32(value.shard_group) - .ok_or_else(|| anyhow!("Block shard_group ({}) is not a valid", value.shard_group))?; + let commands = value + .commands + .into_iter() + .map(TryInto::try_into) + .collect::>()?; - let proposed_by = PublicKey::from_canonical_bytes(&value.proposed_by) - .map_err(|_| anyhow!("Block conversion: Invalid proposed_by"))?; let justify = value .justify - .ok_or_else(|| anyhow!("Block conversion: QC not provided"))? - .try_into()?; - - let extra_data = value.extra_data.map(TryInto::try_into).transpose()?; - - if value.is_dummy { - Ok(Self::dummy_block( - network, - value.parent_id.try_into()?, - proposed_by, - NodeHeight(value.height), - justify, - Epoch(value.epoch), - shard_group, - value.merkle_root.try_into()?, - value.timestamp, - value.base_layer_block_height, - value.base_layer_block_hash.try_into()?, - )) - } else { - Ok(Self::new( - network, - value.parent_id.try_into()?, - justify, - NodeHeight(value.height), - Epoch(value.epoch), - shard_group, - proposed_by, - value - .commands - .into_iter() - .map(TryInto::try_into) - .collect::>()?, - value.merkle_root.try_into()?, - value.total_leader_fee, - decode_exact(&value.foreign_indexes)?, - value.signature.map(TryInto::try_into).transpose()?, - value.timestamp, - value.base_layer_block_height, - value.base_layer_block_hash.try_into()?, - extra_data, - )) - } + .ok_or_else(|| anyhow!("Block conversion: QC not provided"))?; + let justify = consensus_models::QuorumCertificate::try_from(justify)?; + + let header = value.header.ok_or_else(|| anyhow!("BlockHeader not provided"))?; + let header = try_convert_proto_block_header(header, *justify.id(), &commands)?; + + Ok(Self::new(header, justify, commands)) } } diff --git a/dan_layer/rpc_state_sync/src/manager.rs b/dan_layer/rpc_state_sync/src/manager.rs index 7ff6020b9..d5d0cd051 100644 --- a/dan_layer/rpc_state_sync/src/manager.rs +++ b/dan_layer/rpc_state_sync/src/manager.rs @@ -42,15 +42,7 @@ use tari_dan_storage::{ use tari_engine_types::substate::hash_substate; use tari_epoch_manager::EpochManagerReader; use tari_rpc_framework::RpcError; -use tari_state_tree::{ - memory_store::MemoryTreeStore, - Hash, - RootStateTree, - SpreadPrefixStateTree, - SubstateTreeChange, - Version, - SPARSE_MERKLE_PLACEHOLDER_HASH, -}; +use tari_state_tree::{Hash, SpreadPrefixStateTree, SubstateTreeChange, Version, SPARSE_MERKLE_PLACEHOLDER_HASH}; use tari_validator_node_rpc::{ client::{TariValidatorNodeRpcClientFactory, ValidatorNodeClientFactory}, rpc_service::ValidatorNodeRpcClient, @@ -338,16 +330,20 @@ where TConsensusSpec: ConsensusSpec fn validate_checkpoint(&self, checkpoint: &EpochCheckpoint) -> Result<(), CommsRpcConsensusSyncError> { // TODO: validate checkpoint - // Check the merkle root matches the provided shard roots - let mut mem_store = MemoryTreeStore::new(); - let mut root_tree = RootStateTree::new(&mut mem_store); - let shard_group = checkpoint.block().shard_group(); - let hashes = shard_group.shard_iter().map(|shard| checkpoint.get_shard_root(shard)); - let calculated_root = root_tree.put_root_hash_changes(None, 1, hashes)?; - if calculated_root != *checkpoint.block().merkle_root() { + if !checkpoint.block().is_epoch_end() { + return Err(CommsRpcConsensusSyncError::InvalidResponse(anyhow!( + "Checkpoint block is not an Epoch End block" + ))); + } + + // Sanity check that the calculated merkle root matches the provided shard roots + // Note this allows us to use each of the provided shard MRs assuming we trust the provided block that has been + // signed by a BFT majority of registered VNs + let calculated_root = checkpoint.compute_state_merkle_root()?; + if calculated_root != *checkpoint.block().state_merkle_root() { return Err(CommsRpcConsensusSyncError::InvalidResponse(anyhow!( "Checkpoint merkle root mismatch. Expected {expected} but got {actual}", - expected = checkpoint.block().merkle_root(), + expected = checkpoint.block().state_merkle_root(), actual = calculated_root, ))); } @@ -441,6 +437,7 @@ where TConsensusSpec: ConsensusSpec + Send + Sync + 'static info!(target: LOG_TARGET, "🛜 Checkpoint: {checkpoint}"); self.validate_checkpoint(&checkpoint)?; + self.state_store.with_write_tx(|tx| checkpoint.save(tx))?; match self.start_state_sync(&mut client, shard, &checkpoint).await { Ok(current_version) => { @@ -454,7 +451,7 @@ where TConsensusSpec: ConsensusSpec + Send + Sync + 'static actual = state_root, ); last_error = Some(CommsRpcConsensusSyncError::StateRootMismatch { - expected: *checkpoint.block().merkle_root(), + expected: *checkpoint.block().state_merkle_root(), actual: state_root, }); // TODO: rollback state diff --git a/dan_layer/state_store_sqlite/migrations/2023-06-08-091819_create_state_store/up.sql b/dan_layer/state_store_sqlite/migrations/2023-06-08-091819_create_state_store/up.sql index 4d8a4dded..4ca94a0cc 100644 --- a/dan_layer/state_store_sqlite/migrations/2023-06-08-091819_create_state_store/up.sql +++ b/dan_layer/state_store_sqlite/migrations/2023-06-08-091819_create_state_store/up.sql @@ -18,7 +18,8 @@ create table blocks id integer not null primary key AUTOINCREMENT, block_id text not NULL, parent_block_id text not NULL REFERENCES blocks (block_id), - merkle_root text not NULL, + state_merkle_root text not NULL, + command_merkle_root text not NULL, network text not NULL, height bigint not NULL, epoch bigint not NULL, @@ -53,7 +54,8 @@ create table parked_blocks id integer not null primary key AUTOINCREMENT, block_id text not NULL, parent_block_id text not NULL, - merkle_root text not NULL, + state_merkle_root text not NULL, + command_merkle_root text not NULL, network text not NULL, height bigint not NULL, epoch bigint not NULL, @@ -382,7 +384,8 @@ CREATE TABLE foreign_proposals id integer not null primary key AUTOINCREMENT, block_id text not NULL, parent_block_id text not NULL, - merkle_root text not NULL, + state_merkle_root text not NULL, + command_merkle_root text not NULL, network text not NULL, height bigint not NULL, epoch bigint not NULL, @@ -506,12 +509,13 @@ CREATE INDEX state_transitions_epoch on state_transitions (epoch); CREATE TABLE validator_epoch_stats ( - id integer not NULL primary key AUTOINCREMENT, - epoch bigint not NULL, - public_key text not NULL, - participation_shares bigint not NULL DEFAULT '0', - missed_proposals bigint not NULL DEFAULT '0', - created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP + id integer not NULL primary key AUTOINCREMENT, + epoch bigint not NULL, + public_key text not NULL, + participation_shares bigint not NULL DEFAULT '0', + missed_proposals bigint not NULL DEFAULT '0', + missed_proposals_capped bigint not NULL DEFAULT '0', + created_at timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE UNIQUE INDEX participation_shares_uniq_idx_epoch_public_key on validator_epoch_stats (epoch, public_key); @@ -547,7 +551,8 @@ create table diagnostic_deleted_blocks id integer not null primary key AUTOINCREMENT, block_id text not NULL, parent_block_id text not NULL, - merkle_root text not NULL, + state_merkle_root text not NULL, + command_merkle_root text not NULL, network text not NULL, height bigint not NULL, epoch bigint not NULL, diff --git a/dan_layer/state_store_sqlite/src/reader.rs b/dan_layer/state_store_sqlite/src/reader.rs index b7052d76f..480905120 100644 --- a/dan_layer/state_store_sqlite/src/reader.rs +++ b/dan_layer/state_store_sqlite/src/reader.rs @@ -140,7 +140,7 @@ impl<'a, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'a> SqliteState // Blocks without commands may change pending transaction state because they justify a // block that proposes a change. So we cannot only use blocks that have commands. - let applicable_block_ids = self.get_block_ids_between(from_block_id, to_block_id)?; + let applicable_block_ids = self.get_block_ids_between(from_block_id, to_block_id, 1000)?; debug!( target: LOG_TARGET, @@ -254,6 +254,7 @@ impl<'a, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'a> SqliteState &self, start_block: &BlockId, end_block: &BlockId, + limit: u64, ) -> Result, SqliteStorageError> { debug!(target: LOG_TARGET, "get_block_ids_between: start: {start_block}, end: {end_block}"); let block_ids = sql_query( @@ -266,12 +267,13 @@ impl<'a, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'a> SqliteState block_id = tree.parent AND tree.bid != ? AND tree.parent != '0000000000000000000000000000000000000000000000000000000000000000' - LIMIT 1000 + LIMIT ? ) SELECT bid FROM tree"#, ) .bind::(serialize_hex(end_block)) .bind::(serialize_hex(start_block)) + .bind::(limit as i64) .load_iter::(self.connection()) .map_err(|e| SqliteStorageError::DieselError { operation: "get_block_ids_that_change_state_between", @@ -811,7 +813,7 @@ impl<'tx, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'tx> StateStor } let commit_block = self.get_commit_block()?; - let block_ids = self.get_block_ids_between(commit_block.block_id(), from_block_id)?; + let block_ids = self.get_block_ids_between(commit_block.block_id(), from_block_id, 1000)?; let tx_id = serialize_hex(tx_id); let execution = transaction_executions::table @@ -953,33 +955,24 @@ impl<'tx, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'tx> StateStor &self, epoch: Epoch, shard_group: ShardGroup, - start_block_id: &BlockId, - end_block_id: &BlockId, + start_block_height: NodeHeight, + end_block_height: NodeHeight, include_dummy_blocks: bool, + limit: u64, ) -> Result, StorageError> { use crate::schema::{blocks, quorum_certificates}; - if !self.blocks_exists(start_block_id)? { - return Err(StorageError::QueryError { - reason: format!("blocks_all_between: Start block {} does not exist", start_block_id), - }); - } - - if !self.blocks_exists(end_block_id)? { + if start_block_height > end_block_height { return Err(StorageError::QueryError { - reason: format!("blocks_all_between: End block {} does not exist", end_block_id), + reason: format!( + "Start block height {start_block_height} must be less than end block height {end_block_height}" + ), }); } - let block_ids = self.get_block_ids_between(start_block_id, end_block_id)?; - if block_ids.is_empty() { - return Ok(vec![]); - } - let mut query = blocks::table .left_join(quorum_certificates::table.on(blocks::qc_id.eq(quorum_certificates::qc_id))) .select((blocks::all_columns, quorum_certificates::all_columns.nullable())) - .filter(blocks::block_id.eq_any(block_ids)) .into_boxed(); if !include_dummy_blocks { @@ -989,7 +982,10 @@ impl<'tx, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'tx> StateStor let results = query .filter(blocks::epoch.eq(epoch.as_u64() as i64)) .filter(blocks::shard_group.eq(shard_group.encode_as_u32() as i32)) + .filter(blocks::height.ge(start_block_height.as_u64() as i64)) + .filter(blocks::height.le(end_block_height.as_u64() as i64)) .order_by(blocks::height.asc()) + .limit(limit as i64) .get_results::<(sql_models::Block, Option)>(self.connection()) .map_err(|e| SqliteStorageError::DieselError { operation: "blocks_all_after_height", @@ -2186,7 +2182,7 @@ impl<'tx, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'tx> StateStor let commit_block = self.get_commit_block()?; // Block may modify state with zero commands because the justify a block that changes state - let block_ids = self.get_block_ids_between(commit_block.block_id(), block_id)?; + let block_ids = self.get_block_ids_between(commit_block.block_id(), block_id, 1000)?; if block_ids.is_empty() { return Ok(HashMap::new()); @@ -2503,7 +2499,7 @@ impl<'tx, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'tx> StateStor } let commit_block = self.get_commit_block()?; - let block_ids = self.get_block_ids_between(commit_block.block_id(), block_id)?; + let block_ids = self.get_block_ids_between(commit_block.block_id(), block_id, 1000)?; let pks = validator_epoch_stats::table .select(validator_epoch_stats::public_key) @@ -2550,7 +2546,7 @@ impl<'tx, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'tx> StateStor let commit_block = self.get_commit_block()?; - let block_ids = self.get_block_ids_between(commit_block.block_id(), block_id)?; + let block_ids = self.get_block_ids_between(commit_block.block_id(), block_id, 1000)?; let pks = validator_epoch_stats::table .select(validator_epoch_stats::public_key) @@ -2566,7 +2562,7 @@ impl<'tx, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'tx> StateStor ), ), ) - .filter(validator_epoch_stats::missed_proposals.eq(0i64)) + .filter(validator_epoch_stats::missed_proposals_capped.eq(0i64)) .filter(validator_epoch_stats::epoch.eq(commit_block.epoch().as_u64() as i64)) .limit(limit as i64) .get_results::(self.connection()) @@ -2596,7 +2592,7 @@ impl<'tx, TAddr: NodeAddressable + Serialize + DeserializeOwned + 'tx> StateStor } let commit_block = self.get_commit_block()?; - let block_ids = self.get_block_ids_between(commit_block.block_id(), block_id)?; + let block_ids = self.get_block_ids_between(commit_block.block_id(), block_id, 1000)?; let count = suspended_nodes::table .count() diff --git a/dan_layer/state_store_sqlite/src/schema.rs b/dan_layer/state_store_sqlite/src/schema.rs index 7462cb20a..93febe35c 100644 --- a/dan_layer/state_store_sqlite/src/schema.rs +++ b/dan_layer/state_store_sqlite/src/schema.rs @@ -19,7 +19,8 @@ diesel::table! { id -> Integer, block_id -> Text, parent_block_id -> Text, - merkle_root -> Text, + state_merkle_root -> Text, + command_merkle_root -> Text, network -> Text, height -> BigInt, epoch -> BigInt, @@ -61,7 +62,8 @@ diesel::table! { id -> Integer, block_id -> Text, parent_block_id -> Text, - merkle_root -> Text, + state_merkle_root -> Text, + command_merkle_root -> Text, network -> Text, height -> BigInt, epoch -> BigInt, @@ -133,7 +135,8 @@ diesel::table! { id -> Integer, block_id -> Text, parent_block_id -> Text, - merkle_root -> Text, + state_merkle_root -> Text, + command_merkle_root -> Text, network -> Text, height -> BigInt, epoch -> BigInt, @@ -289,7 +292,8 @@ diesel::table! { id -> Integer, block_id -> Text, parent_block_id -> Text, - merkle_root -> Text, + state_merkle_root -> Text, + command_merkle_root -> Text, network -> Text, height -> BigInt, epoch -> BigInt, @@ -527,6 +531,7 @@ diesel::table! { public_key -> Text, participation_shares -> BigInt, missed_proposals -> BigInt, + missed_proposals_capped -> BigInt, created_at -> Timestamp, } } diff --git a/dan_layer/state_store_sqlite/src/sql_models/block.rs b/dan_layer/state_store_sqlite/src/sql_models/block.rs index 96fe9d3f7..4da046811 100644 --- a/dan_layer/state_store_sqlite/src/sql_models/block.rs +++ b/dan_layer/state_store_sqlite/src/sql_models/block.rs @@ -19,7 +19,8 @@ pub struct Block { pub id: i32, pub block_id: String, pub parent_block_id: String, - pub merkle_root: String, + pub state_merkle_root: String, + pub command_merkle_root: String, pub network: String, pub height: i64, pub epoch: i64, @@ -72,8 +73,9 @@ impl Block { details: format!("Block #{} proposed_by is malformed", self.id), } })?, + deserialize_hex_try_from(&self.state_merkle_root)?, deserialize_json(&self.commands)?, - deserialize_hex_try_from(&self.merkle_root)?, + deserialize_hex_try_from(&self.command_merkle_root)?, self.total_leader_fee as u64, self.is_dummy, self.is_justified, @@ -85,7 +87,10 @@ impl Block { self.timestamp as u64, self.base_layer_block_height as u64, deserialize_hex_try_from(&self.base_layer_block_hash)?, - self.extra_data.map(|val| deserialize_json(&val)).transpose()?, + self.extra_data + .map(|val| deserialize_json(&val)) + .transpose()? + .unwrap_or_default(), )) } } @@ -95,7 +100,8 @@ pub struct ParkedBlock { pub id: i32, pub block_id: String, pub parent_block_id: String, - pub merkle_root: String, + pub state_merkle_root: String, + pub command_merkle_root: String, pub network: String, pub height: i64, pub epoch: i64, @@ -144,8 +150,9 @@ impl TryFrom for (consensus_models::Block, Vec for (consensus_models::Block, Vec SqliteStateStoreWriteTransaction<'a, TAddr> { parked_blocks::block_id.eq(&block_id), parked_blocks::parent_block_id.eq(serialize_hex(block.parent())), parked_blocks::network.eq(block.network().to_string()), - parked_blocks::merkle_root.eq(block.merkle_root().to_string()), + parked_blocks::state_merkle_root.eq(block.state_merkle_root().to_string()), + parked_blocks::command_merkle_root.eq(block.command_merkle_root().to_string()), parked_blocks::height.eq(block.height().as_u64() as i64), parked_blocks::epoch.eq(block.epoch().as_u64() as i64), parked_blocks::shard_group.eq(block.shard_group().encode_as_u32() as i32), @@ -192,7 +193,7 @@ impl<'a, TAddr: NodeAddressable> SqliteStateStoreWriteTransaction<'a, TAddr> { parked_blocks::base_layer_block_height.eq(block.base_layer_block_height() as i64), parked_blocks::base_layer_block_hash.eq(serialize_hex(block.base_layer_block_hash())), parked_blocks::foreign_proposals.eq(serialize_json(foreign_proposals)?), - parked_blocks::extra_data.eq(block.extra_data().map(serialize_json).transpose()?), + parked_blocks::extra_data.eq(serialize_json(block.extra_data())?), ); diesel::insert_into(parked_blocks::table) @@ -228,7 +229,8 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for SqliteSta let insert = ( blocks::block_id.eq(serialize_hex(block.id())), blocks::parent_block_id.eq(serialize_hex(block.parent())), - blocks::merkle_root.eq(block.merkle_root().to_string()), + blocks::state_merkle_root.eq(block.state_merkle_root().to_string()), + blocks::command_merkle_root.eq(block.command_merkle_root().to_string()), blocks::network.eq(block.network().to_string()), blocks::height.eq(block.height().as_u64() as i64), blocks::epoch.eq(block.epoch().as_u64() as i64), @@ -246,7 +248,7 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for SqliteSta blocks::timestamp.eq(block.timestamp() as i64), blocks::base_layer_block_height.eq(block.base_layer_block_height() as i64), blocks::base_layer_block_hash.eq(serialize_hex(block.base_layer_block_hash())), - blocks::extra_data.eq(block.extra_data().map(serialize_json).transpose()?), + blocks::extra_data.eq(serialize_json(block.extra_data())?), ); diesel::insert_into(blocks::table) @@ -625,7 +627,8 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for SqliteSta let values = ( foreign_proposals::block_id.eq(serialize_hex(block.id())), foreign_proposals::parent_block_id.eq(serialize_hex(block.parent())), - foreign_proposals::merkle_root.eq(block.merkle_root().to_string()), + foreign_proposals::state_merkle_root.eq(block.state_merkle_root().to_string()), + foreign_proposals::command_merkle_root.eq(block.command_merkle_root().to_string()), foreign_proposals::network.eq(block.network().to_string()), foreign_proposals::height.eq(block.height().as_u64() as i64), foreign_proposals::epoch.eq(block.epoch().as_u64() as i64), @@ -643,7 +646,7 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for SqliteSta foreign_proposals::justify_qc_id.eq(serialize_hex(foreign_proposal.justify_qc().id())), foreign_proposals::block_pledge.eq(serialize_json(foreign_proposal.block_pledge())?), foreign_proposals::status.eq(ForeignProposalStatus::New.to_string()), - foreign_proposals::extra_data.eq(foreign_proposal.block().extra_data().map(serialize_json).transpose()?), + foreign_proposals::extra_data.eq(serialize_json(foreign_proposal.block().extra_data())?), ); diesel::insert_into(foreign_proposals::table) @@ -2184,10 +2187,11 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for SqliteSta .select(( validator_epoch_stats::participation_shares, validator_epoch_stats::missed_proposals, + validator_epoch_stats::missed_proposals_capped, )) .filter(validator_epoch_stats::epoch.eq(epoch)) .filter(validator_epoch_stats::public_key.eq(serialize_hex(update.public_key().as_bytes()))) - .first::<(i64, i64)>(self.connection()) + .first::<(i64, i64, i64)>(self.connection()) .optional() .map_err(|e| SqliteStorageError::DieselError { operation: "validator_epoch_stats_updates", @@ -2195,7 +2199,9 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for SqliteSta })?; match existing { - Some((participation_shares, missed_proposals)) => match update.missed_proposal_change() { + Some((participation_shares, missed_proposals, missed_proposals_capped)) => match update + .missed_proposal_change() + { Some(0) => { diesel::update(validator_epoch_stats::table) .filter(validator_epoch_stats::epoch.eq(epoch)) @@ -2204,6 +2210,7 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for SqliteSta validator_epoch_stats::participation_shares .eq(participation_shares + update.participation_shares_increment() as i64), validator_epoch_stats::missed_proposals.eq(0), + validator_epoch_stats::missed_proposals_capped.eq(0), )) .execute(self.connection()) .map_err(|e| SqliteStorageError::DieselError { @@ -2212,9 +2219,11 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for SqliteSta })?; }, Some(n) => { - let missed_proposal_count = update + // NOTE: n can be negative + let missed_proposal_count = cmp::max(missed_proposals + n, 0); + let capped_missed_proposal_count = update .max_total_missed_proposals() - .min(cmp::max(missed_proposals + n, 0)); + .min(cmp::max(missed_proposals_capped + n, 0)); diesel::update(validator_epoch_stats::table) .filter(validator_epoch_stats::epoch.eq(epoch)) .filter(validator_epoch_stats::public_key.eq(serialize_hex(update.public_key().as_bytes()))) @@ -2222,6 +2231,7 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for SqliteSta validator_epoch_stats::participation_shares .eq(participation_shares + update.participation_shares_increment() as i64), validator_epoch_stats::missed_proposals.eq(missed_proposal_count), + validator_epoch_stats::missed_proposals_capped.eq(capped_missed_proposal_count), )) .execute(self.connection()) .map_err(|e| SqliteStorageError::DieselError { @@ -2252,6 +2262,7 @@ impl<'tx, TAddr: NodeAddressable + 'tx> StateStoreWriteTransaction for SqliteSta validator_epoch_stats::public_key.eq(serialize_hex(update.public_key().as_bytes())), validator_epoch_stats::participation_shares.eq(update.participation_shares_increment() as i64), validator_epoch_stats::missed_proposals.eq(leader_failure_inc), + validator_epoch_stats::missed_proposals_capped.eq(leader_failure_inc), ); diesel::insert_into(validator_epoch_stats::table) diff --git a/dan_layer/state_store_sqlite/tests/tests.rs b/dan_layer/state_store_sqlite/tests/tests.rs index ae511a8e5..91e951f3f 100644 --- a/dan_layer/state_store_sqlite/tests/tests.rs +++ b/dan_layer/state_store_sqlite/tests/tests.rs @@ -31,7 +31,7 @@ fn create_tx_atom() -> TransactionAtom { } mod confirm_all_transitions { - use tari_dan_common_types::{NumPreshards, ShardGroup}; + use tari_dan_common_types::{ExtraData, NumPreshards, ShardGroup}; use super::*; @@ -47,9 +47,9 @@ mod confirm_all_transitions { let atom3 = create_tx_atom(); let network = Default::default(); - let zero_block = Block::zero_block(network, NumPreshards::P64, None).unwrap(); + let zero_block = Block::zero_block(network, NumPreshards::P64); zero_block.insert(&mut tx).unwrap(); - let block1 = Block::new( + let block1 = Block::create( network, *zero_block.id(), zero_block.justify().clone(), @@ -67,8 +67,9 @@ mod confirm_all_transitions { EpochTime::now().as_u64(), 0, FixedHash::zero(), - None, - ); + ExtraData::default(), + ) + .unwrap(); block1.insert(&mut tx).unwrap(); tx.transaction_pool_insert_new(atom1.id, atom1.decision, true).unwrap(); diff --git a/dan_layer/state_tree/src/jellyfish/tree.rs b/dan_layer/state_tree/src/jellyfish/tree.rs index 70ee10c45..ef386c958 100644 --- a/dan_layer/state_tree/src/jellyfish/tree.rs +++ b/dan_layer/state_tree/src/jellyfish/tree.rs @@ -728,3 +728,63 @@ pub struct StaleNodeIndex { /// record. pub node_key: NodeKey, } + +#[cfg(test)] +mod tests { + use super::*; + use crate::{jmt_node_hash, memory_store::MemoryTreeStore, StaleTreeNode, TreeStoreWriter}; + + fn leaf_key(seed: u64) -> LeafKey { + LeafKey::new(jmt_node_hash(&seed)) + } + + #[test] + fn check_merkle_proof() { + // Evaluating the functionality of the JMT Merkle proof. + let mut mem = MemoryTreeStore::new(); + let jmt = JellyfishMerkleTree::new(&mem); + + let values = [ + (leaf_key(1), Some((jmt_node_hash(&10), Some(1u64)))), + (leaf_key(2), Some((jmt_node_hash(&11), Some(2)))), + (leaf_key(3), Some((jmt_node_hash(&12), Some(3)))), + ]; + let (_, diff) = jmt.batch_put_value_set(values, None, None, 1).unwrap(); + for (k, v) in diff.node_batch { + mem.insert_node(k, v).unwrap(); + } + + for a in diff.stale_node_index_batch { + mem.record_stale_tree_node(StaleTreeNode::Node(a.node_key)).unwrap(); + } + mem.clear_stale_nodes(); + + let jmt = JellyfishMerkleTree::new(&mem); + + // This causes get_with_proof to fail with node NotFound. + let values = [ + (leaf_key(4), Some((jmt_node_hash(&13), Some(4u64)))), + (leaf_key(5), Some((jmt_node_hash(&14), Some(5)))), + (leaf_key(6), Some((jmt_node_hash(&15), Some(6)))), + ]; + let (_mr, diff) = jmt.batch_put_value_set(values, None, Some(1), 2).unwrap(); + + for (k, v) in diff.node_batch { + mem.insert_node(k, v).unwrap(); + } + for a in diff.stale_node_index_batch { + mem.record_stale_tree_node(StaleTreeNode::Node(a.node_key)).unwrap(); + } + mem.clear_stale_nodes(); + let jmt = JellyfishMerkleTree::new(&mem); + + let k = leaf_key(3); + let (_value, sparse) = jmt.get_with_proof(k.as_ref(), 2).unwrap(); + + let leaf = sparse.leaf().unwrap(); + assert_eq!(*leaf.key(), k); + assert_eq!(*leaf.value_hash(), jmt_node_hash(&12)); + // Unanswered: How do we verify the proof root matches a Merkle root? + // assert!(sparse.siblings().iter().any(|h| *h == mr)); + } +} diff --git a/dan_layer/state_tree/src/tree.rs b/dan_layer/state_tree/src/tree.rs index 3c89df7fc..91d0464ee 100644 --- a/dan_layer/state_tree/src/tree.rs +++ b/dan_layer/state_tree/src/tree.rs @@ -1,7 +1,7 @@ // Copyright 2024 The Tari Project // SPDX-License-Identifier: BSD-3-Clause -use std::marker::PhantomData; +use std::{iter::Peekable, marker::PhantomData}; use serde::{Deserialize, Serialize}; use tari_engine_types::substate::SubstateId; @@ -10,12 +10,14 @@ use crate::{ error::StateTreeError, jellyfish::{Hash, JellyfishMerkleTree, SparseMerkleProofExt, TreeStore, Version}, key_mapper::{DbKeyMapper, HashIdentityKeyMapper, SpreadPrefixKeyMapper}, + memory_store::MemoryTreeStore, Node, NodeKey, ProofValue, StaleTreeNode, TreeStoreReader, TreeUpdateBatch, + SPARSE_MERKLE_PLACEHOLDER_HASH, }; pub type SpreadPrefixStateTree<'a, S> = StateTree<'a, S, SpreadPrefixKeyMapper>; @@ -94,19 +96,13 @@ impl<'a, S: TreeStore, M: DbKeyMapper> StateTree<'a, S, M> } impl<'a, S: TreeStore<()>, M: DbKeyMapper> StateTree<'a, S, M> { - pub fn put_root_hash_changes>( + pub fn put_changes>( &mut self, current_version: Option, next_version: Version, changes: I, ) -> Result { - let jmt = JellyfishMerkleTree::<_, ()>::new(self.store); - - let changes = changes - .into_iter() - .map(|hash| (M::map_to_leaf_key(&hash), Some((hash, ())))); - - let (root_hash, update_result) = jmt.batch_put_value_set(changes, None, current_version, next_version)?; + let (root_hash, update_result) = self.compute_update_batch(current_version, next_version, changes)?; for (k, node) in update_result.node_batch { self.store.insert_node(k, node)?; @@ -119,6 +115,22 @@ impl<'a, S: TreeStore<()>, M: DbKeyMapper> StateTree<'a, S, M> { Ok(root_hash) } + + pub fn compute_update_batch>( + &mut self, + current_version: Option, + next_version: Version, + changes: I, + ) -> Result<(Hash, TreeUpdateBatch<()>), StateTreeError> { + let jmt = JellyfishMerkleTree::<_, ()>::new(self.store); + + let changes = changes + .into_iter() + .map(|hash| (M::map_to_leaf_key(&hash), Some((hash, ())))); + + let (root, update) = jmt.batch_put_value_set(changes, None, current_version, next_version)?; + Ok((root, update)) + } } /// Calculates the new root hash and tree updates for the given substate changes. @@ -185,3 +197,15 @@ impl

    From> for StateHashTreeDiff

    { } } } + +pub fn compute_merkle_root_for_hashes>( + mut hashes: Peekable, +) -> Result { + if hashes.peek().is_none() { + return Ok(SPARSE_MERKLE_PLACEHOLDER_HASH); + } + let mut mem_store = MemoryTreeStore::new(); + let mut root_tree = RootStateTree::new(&mut mem_store); + let (hash, _) = root_tree.compute_update_batch(None, 1, hashes)?; + Ok(hash) +} diff --git a/dan_layer/storage/src/consensus_models/block.rs b/dan_layer/storage/src/consensus_models/block.rs index 60e226357..185fa98ba 100644 --- a/dan_layer/storage/src/consensus_models/block.rs +++ b/dan_layer/storage/src/consensus_models/block.rs @@ -4,7 +4,6 @@ use std::{ collections::{BTreeSet, HashSet}, fmt::{Debug, Display, Formatter}, - hash::Hash, iter, ops::{Deref, RangeInclusive}, }; @@ -14,22 +13,21 @@ use log::*; use serde::{Deserialize, Serialize}; use tari_common::configuration::Network; use tari_common_types::types::{FixedHash, FixedHashSizeError, PublicKey}; -use tari_crypto::{ristretto::RistrettoPublicKey, tari_utilities::epoch_time::EpochTime}; +use tari_crypto::{ristretto::RistrettoPublicKey, tari_utilities::ByteArray}; use tari_dan_common_types::{ committee::CommitteeInfo, - hashing, optional::Optional, serde_with, shard::Shard, Epoch, ExtraData, - MaxSizeBytesError, - NodeAddressable, + ExtraFieldKey, NodeHeight, NumPreshards, ShardGroup, SubstateAddress, }; +use tari_state_tree::StateTreeError; use tari_transaction::TransactionId; use time::PrimitiveDateTime; #[cfg(feature = "ts")] @@ -56,6 +54,7 @@ use super::{ }; use crate::{ consensus_models::{ + block_header::{compute_command_merkle_root, BlockHeader}, Command, LastExecuted, LastProposed, @@ -76,62 +75,34 @@ const LOG_TARGET: &str = "tari::dan::storage::consensus_models::block"; #[derive(Debug, thiserror::Error)] pub enum BlockError { - #[error("Extra data size error: {0}")] - ExtraDataSizeError(#[from] MaxSizeBytesError), + #[error("Error computing command merkle hash: {0}")] + StateTreeError(#[from] StateTreeError), } #[derive(Debug, Clone, Serialize, Deserialize)] #[cfg_attr(feature = "ts", derive(TS), ts(export, export_to = "../../bindings/src/types/"))] pub struct Block { - // Header - #[cfg_attr(feature = "ts", ts(type = "string"))] - id: BlockId, - #[cfg_attr(feature = "ts", ts(type = "string"))] - network: Network, - #[cfg_attr(feature = "ts", ts(type = "string"))] - parent: BlockId, + header: BlockHeader, justify: QuorumCertificate, - height: NodeHeight, - epoch: Epoch, - shard_group: ShardGroup, - #[cfg_attr(feature = "ts", ts(type = "string"))] - proposed_by: PublicKey, - #[cfg_attr(feature = "ts", ts(type = "number"))] - total_leader_fee: u64, - // Body - #[cfg_attr(feature = "ts", ts(type = "string"))] - merkle_root: FixedHash, - /// Ordered commands that help ensure a deterministic block hash. + /// Commands in the block. These are in canonical order to ensure a deterministic block hash. commands: BTreeSet, - /// If the block is a dummy block. This is metadata and not sent over - /// the wire or part of the block hash. - is_dummy: bool, + // Metadata - not included in the block hash /// Flag that indicates that the block has been justified by a new high QC. is_justified: bool, /// Flag that indicates that the block has been committed. is_committed: bool, - /// Counter for each foreign shard for reliable broadcast. - foreign_indexes: IndexMap, + #[cfg_attr(feature = "ts", ts(type = "number | null"))] + block_time: Option, /// Timestamp when was this stored. #[cfg_attr(feature = "ts", ts(type = "Array| null"))] stored_at: Option, - /// Signature of block by the proposer. - #[cfg_attr(feature = "ts", ts(type = "{public_nonce : string, signature: string} | null"))] - signature: Option, - #[cfg_attr(feature = "ts", ts(type = "number | null"))] - block_time: Option, - #[cfg_attr(feature = "ts", ts(type = "number"))] - timestamp: u64, - #[cfg_attr(feature = "ts", ts(type = "number"))] - base_layer_block_height: u64, - #[cfg_attr(feature = "ts", ts(type = "string"))] - base_layer_block_hash: FixedHash, - extra_data: Option, } impl Block { + /// Creates a new block from the provided params. Returns an error if the command merkle root fails to construct. + /// This is infallible for empty commands. #[allow(clippy::too_many_arguments)] - pub fn new( + pub fn create( network: Network, parent: BlockId, justify: QuorumCertificate, @@ -140,41 +111,46 @@ impl Block { shard_group: ShardGroup, proposed_by: PublicKey, commands: BTreeSet, - merkle_root: FixedHash, + state_merkle_root: FixedHash, total_leader_fee: u64, sorted_foreign_indexes: IndexMap, signature: Option, timestamp: u64, base_layer_block_height: u64, base_layer_block_hash: FixedHash, - extra_data: Option, - ) -> Self { - let mut block = Self { - id: BlockId::zero(), + extra_data: ExtraData, + ) -> Result { + let header = BlockHeader::create( network, parent, - justify, + *justify.id(), height, epoch, shard_group, proposed_by, - merkle_root, - commands, + state_merkle_root, + &commands, total_leader_fee, - is_dummy: false, - is_justified: false, - is_committed: false, - foreign_indexes: sorted_foreign_indexes, - stored_at: None, + sorted_foreign_indexes, signature, - block_time: None, timestamp, base_layer_block_height, base_layer_block_hash, extra_data, - }; - block.id = block.calculate_hash().into(); - block + )?; + Ok(Self::new(header, justify, commands)) + } + + pub fn new(header: BlockHeader, justify: QuorumCertificate, commands: BTreeSet) -> Self { + Self { + header, + justify, + commands, + is_justified: false, + is_committed: false, + block_time: None, + stored_at: None, + } } #[allow(clippy::too_many_arguments)] @@ -187,8 +163,9 @@ impl Block { epoch: Epoch, shard_group: ShardGroup, proposed_by: PublicKey, + state_merkle_root: FixedHash, commands: BTreeSet, - merkle_root: FixedHash, + command_merkle_root: FixedHash, total_leader_fee: u64, is_dummy: bool, is_justified: bool, @@ -200,31 +177,36 @@ impl Block { timestamp: u64, base_layer_block_height: u64, base_layer_block_hash: FixedHash, - extra_data: Option, + extra_data: ExtraData, ) -> Self { - Self { + let header = BlockHeader::load( id, network, parent, - justify, + *justify.id(), height, epoch, shard_group, proposed_by, - merkle_root, - commands, + state_merkle_root, total_leader_fee, is_dummy, - is_justified, - is_committed, - foreign_indexes: sorted_foreign_indexes, - stored_at: Some(created_at), + sorted_foreign_indexes, signature, - block_time, timestamp, base_layer_block_height, base_layer_block_hash, extra_data, + command_merkle_root, + ); + Self { + header, + justify, + commands, + is_justified, + is_committed, + block_time, + stored_at: Some(created_at), } } @@ -232,9 +214,22 @@ impl Block { network: Network, epoch: Epoch, shard_group: ShardGroup, + state_merkle_root: FixedHash, sidechain_id: Option, - ) -> Result { - Ok(Self::new( + ) -> Self { + let mut extra_data = ExtraData::new(); + if let Some(sidechain_id) = sidechain_id { + extra_data.insert( + ExtraFieldKey::SidechainId, + sidechain_id + .as_bytes() + .to_vec() + .try_into() + .expect("RistrettoPublicKey is 32 bytes"), + ); + } + + Self::create( network, BlockId::zero(), QuorumCertificate::genesis(epoch, shard_group), @@ -243,139 +238,41 @@ impl Block { shard_group, PublicKey::default(), Default::default(), - // TODO: the merkle hash should be initialized to something committing to the previous state. - FixedHash::zero(), + state_merkle_root, 0, IndexMap::new(), None, 0, 0, FixedHash::zero(), - Self::extra_data_from_sidechain_id(sidechain_id)?, - )) + extra_data, + ) + .expect("Infallible with empty commands") } /// This is the parent block for all genesis blocks. Its block ID is always zero. - pub fn zero_block( - network: Network, - num_preshards: NumPreshards, - sidechain_id: Option, - ) -> Result { - Ok(Self { - network, - id: BlockId::zero(), - parent: BlockId::zero(), + pub fn zero_block(network: Network, num_preshards: NumPreshards) -> Self { + Self { + header: BlockHeader::zero_block(network, num_preshards), justify: QuorumCertificate::genesis(Epoch::zero(), ShardGroup::all_shards(num_preshards)), - height: NodeHeight::zero(), - epoch: Epoch::zero(), - shard_group: ShardGroup::all_shards(num_preshards), - proposed_by: PublicKey::default(), - merkle_root: FixedHash::zero(), commands: Default::default(), - total_leader_fee: 0, - is_dummy: false, is_justified: false, is_committed: true, - foreign_indexes: IndexMap::new(), - stored_at: None, - signature: None, - block_time: None, - timestamp: EpochTime::now().as_u64(), - base_layer_block_height: 0, - base_layer_block_hash: FixedHash::zero(), - extra_data: Self::extra_data_from_sidechain_id(sidechain_id)?, - }) - } - - pub fn dummy_block( - network: Network, - parent: BlockId, - proposed_by: PublicKey, - height: NodeHeight, - high_qc: QuorumCertificate, - epoch: Epoch, - shard_group: ShardGroup, - parent_merkle_root: FixedHash, - parent_timestamp: u64, - parent_base_layer_block_height: u64, - parent_base_layer_block_hash: FixedHash, - ) -> Self { - let mut block = Self { - id: BlockId::zero(), - network, - parent, - justify: high_qc, - height, - epoch, - shard_group, - proposed_by, - merkle_root: parent_merkle_root, - commands: BTreeSet::new(), - total_leader_fee: 0, - is_dummy: true, - is_justified: false, - is_committed: false, - foreign_indexes: IndexMap::new(), stored_at: None, - signature: None, block_time: None, - timestamp: parent_timestamp, - base_layer_block_height: parent_base_layer_block_height, - base_layer_block_hash: parent_base_layer_block_hash, - extra_data: None, - }; - block.id = block.calculate_hash().into(); - block.is_justified = false; - block + } } - fn extra_data_from_sidechain_id(sidechain_id: Option) -> Result, BlockError> { - let extra_data = sidechain_id - .map(|id| ExtraData::new().insert_sidechain_id(id).cloned()) - .transpose()?; - Ok(extra_data) + pub fn calculate_hash(&self) -> FixedHash { + self.header.calculate_hash() } - pub fn calculate_hash(&self) -> FixedHash { - // Hash is created from the hash of the "body" and - // then hashed with the parent, so that you can - // create a merkle proof of a chain of blocks - // ```pre - // root - // |\ - // | block1 - // |\ - // | block2 - // | - // blockbody - // ``` - - let inner_hash = hashing::block_hasher() - .chain(&self.network) - // This allows us to exclude the justify and still validate the block - .chain(self.justify.id()) - .chain(&self.height) - .chain(&self.total_leader_fee) - .chain(&self.epoch) - .chain(&self.shard_group) - .chain(&self.proposed_by) - .chain(&self.merkle_root) - .chain(&self.is_dummy) - .chain(&self.commands) - .chain(&self.foreign_indexes) - .chain(&self.timestamp) - .chain(&self.base_layer_block_height) - .chain(&self.base_layer_block_hash) - .chain(&self.extra_data) - .result(); - - hashing::block_hasher().chain(&self.parent).chain(&inner_hash).result() + pub fn header(&self) -> &BlockHeader { + &self.header } -} -impl Block { pub fn is_genesis(&self) -> bool { - self.height.is_zero() + self.header().is_genesis() } pub fn is_epoch_end(&self) -> bool { @@ -430,55 +327,35 @@ impl Block { } pub fn as_locked_block(&self) -> LockedBlock { - LockedBlock { - height: self.height, - block_id: self.id, - epoch: self.epoch, - } + self.header().as_locked_block() } pub fn as_last_executed(&self) -> LastExecuted { - LastExecuted { - height: self.height, - block_id: self.id, - epoch: self.epoch, - } + self.header().as_last_executed() } pub fn as_last_voted(&self) -> LastVoted { - LastVoted { - height: self.height, - block_id: self.id, - epoch: self.epoch, - } + self.header().as_last_voted() } pub fn as_leaf_block(&self) -> LeafBlock { - LeafBlock { - height: self.height, - block_id: self.id, - epoch: self.epoch, - } + self.header().as_leaf_block() } pub fn as_last_proposed(&self) -> LastProposed { - LastProposed { - height: self.height, - block_id: self.id, - epoch: self.epoch, - } + self.header().as_last_proposed() } pub fn id(&self) -> &BlockId { - &self.id + self.header.id() } pub fn network(&self) -> Network { - self.network + self.header.network() } pub fn parent(&self) -> &BlockId { - &self.parent + self.header.parent() } pub fn justify(&self) -> &QuorumCertificate { @@ -490,30 +367,26 @@ impl Block { } pub fn justifies_parent(&self) -> bool { - *self.justify.block_id() == self.parent + self.justify.block_id() == self.parent() } pub fn height(&self) -> NodeHeight { - self.height - } - - pub fn is_zero(&self) -> bool { - self.id.is_zero() + self.header.height() } pub fn epoch(&self) -> Epoch { - self.epoch + self.header.epoch() } pub fn shard_group(&self) -> ShardGroup { - self.shard_group + self.header.shard_group() } pub fn total_leader_fee(&self) -> u64 { - self.total_leader_fee + self.header.total_leader_fee() } - pub fn total_transaction_fee(&self) -> u64 { + pub fn calculate_total_transaction_fee(&self) -> u64 { self.commands .iter() .filter_map(|c| c.committing()) @@ -522,11 +395,19 @@ impl Block { } pub fn proposed_by(&self) -> &PublicKey { - &self.proposed_by + self.header.proposed_by() + } + + pub fn state_merkle_root(&self) -> &FixedHash { + self.header.state_merkle_root() } - pub fn merkle_root(&self) -> &FixedHash { - &self.merkle_root + pub fn command_merkle_root(&self) -> &FixedHash { + self.header.command_merkle_root() + } + + pub fn compute_command_merkle_root(&self) -> Result { + compute_command_merkle_root(&self.commands) } pub fn commands(&self) -> &BTreeSet { @@ -538,7 +419,7 @@ impl Block { } pub fn is_dummy(&self) -> bool { - self.is_dummy + self.header.is_dummy() } pub fn is_justified(&self) -> bool { @@ -549,12 +430,12 @@ impl Block { self.is_committed } - pub fn get_foreign_counter(&self, bucket: &Shard) -> Option { - self.foreign_indexes.get(bucket).copied() + pub fn get_foreign_counter(&self, shard: &Shard) -> Option { + self.header.get_foreign_counter(shard) } pub fn foreign_indexes(&self) -> &IndexMap { - &self.foreign_indexes + self.header.foreign_indexes() } pub fn block_time(&self) -> Option { @@ -562,31 +443,27 @@ impl Block { } pub fn timestamp(&self) -> u64 { - self.timestamp + self.header.timestamp() } pub fn signature(&self) -> Option<&ValidatorSchnorrSignature> { - self.signature.as_ref() + self.header.signature() } pub fn set_signature(&mut self, signature: ValidatorSchnorrSignature) { - self.signature = Some(signature); - } - - pub fn is_proposed_by_addr>(&self, address: &A) -> Option { - Some(A::try_from_public_key(&self.proposed_by)? == *address) + self.header.set_signature(signature); } pub fn base_layer_block_height(&self) -> u64 { - self.base_layer_block_height + self.header.base_layer_block_height() } pub fn base_layer_block_hash(&self) -> &FixedHash { - &self.base_layer_block_hash + self.header.base_layer_block_hash() } - pub fn extra_data(&self) -> Option<&ExtraData> { - self.extra_data.as_ref() + pub fn extra_data(&self) -> &ExtraData { + self.header.extra_data() } } @@ -595,16 +472,32 @@ impl Block { tx.blocks_get(id) } + pub fn get_ids_by_epoch_and_height( + tx: &TTx, + epoch: Epoch, + height: NodeHeight, + ) -> Result, StorageError> { + tx.blocks_get_all_ids_by_height(epoch, height) + } + /// Returns all blocks from and excluding the start block (lower height) to the end block (inclusive) pub fn get_all_blocks_between( tx: &TTx, epoch: Epoch, shard_group: ShardGroup, - start_block_id: &BlockId, - end_block_id: &BlockId, + start_block_height: NodeHeight, + end_block_height: NodeHeight, include_dummy_blocks: bool, + limit: u64, ) -> Result, StorageError> { - tx.blocks_get_all_between(epoch, shard_group, start_block_id, end_block_id, include_dummy_blocks) + tx.blocks_get_all_between( + epoch, + shard_group, + start_block_height, + end_block_height, + include_dummy_blocks, + limit, + ) } pub fn get_last_n_in_epoch( @@ -673,7 +566,7 @@ impl Block { TTx: StateStoreWriteTransaction + Deref, TTx::Target: StateStoreReadTransaction, { - let other_blocks = tx.blocks_get_all_ids_by_height(self.epoch(), self.height())?; + let other_blocks = Self::get_ids_by_epoch_and_height(&**tx, self.epoch(), self.height())?; for block_id in other_blocks { if block_id == *self.id() { continue; @@ -807,10 +700,10 @@ impl Block { } pub fn extends(&self, tx: &TTx, ancestor: &BlockId) -> Result { - if self.id == *ancestor { + if self.id() == ancestor { return Ok(false); } - if self.parent == *ancestor { + if self.parent() == ancestor { return Ok(true); } // First check the parent here, if it does not exist, then this block cannot extend anything. @@ -822,14 +715,14 @@ impl Block { } pub fn get_parent(&self, tx: &TTx) -> Result { - if self.id.is_zero() && self.parent.is_zero() { + if self.id().is_zero() && self.parent().is_zero() { return Err(StorageError::NotFound { item: "Block parent", - key: self.parent.to_string(), + key: self.parent().to_string(), }); } - Block::get(tx, &self.parent) + Block::get(tx, self.parent()) } pub fn get_parent_chain( @@ -841,7 +734,7 @@ impl Block { } pub fn get_votes(&self, tx: &TTx) -> Result, StorageError> { - Vote::get_for_block(tx, &self.id) + Vote::get_for_block(tx, self.id()) } pub fn get_child_block_ids(&self, tx: &TTx) -> Result, StorageError> { @@ -955,7 +848,7 @@ impl Block { return Ok(high_qc); } - let current_locked = LockedBlock::get(&**tx, self.epoch)?; + let current_locked = LockedBlock::get(&**tx, self.epoch())?; if prepared_node.height() > current_locked.height { on_locked_block_recurse( tx, @@ -1040,7 +933,7 @@ impl Block { { let mut counters = ForeignSendCounters::get_or_default(&**tx, self.justify().block_id())?; // Add counters for this block and carry over the counters from the justify block, if any - for shard in self.foreign_indexes.keys() { + for shard in self.foreign_indexes().keys() { counters.increment_counter(*shard); } if !counters.is_empty() { @@ -1061,7 +954,7 @@ impl Block { continue; } - let Some(evidence) = atom.evidence.get(&self.shard_group) else { + let Some(evidence) = atom.evidence.get(&self.shard_group()) else { // CASE: The output-only shard group has sequenced this transaction debug!( "get_block_pledge: Local evidence for atom {} is missing in block {}", diff --git a/dan_layer/storage/src/consensus_models/block_header.rs b/dan_layer/storage/src/consensus_models/block_header.rs new file mode 100644 index 000000000..62cf952d8 --- /dev/null +++ b/dan_layer/storage/src/consensus_models/block_header.rs @@ -0,0 +1,398 @@ +// Copyright 2024 The Tari Project +// SPDX-License-Identifier: BSD-3-Clause + +use std::{ + collections::BTreeSet, + fmt::{Debug, Display, Formatter}, +}; + +use indexmap::IndexMap; +use serde::{Deserialize, Serialize}; +use tari_common::configuration::Network; +use tari_common_types::types::{FixedHash, PublicKey}; +use tari_crypto::tari_utilities::epoch_time::EpochTime; +use tari_dan_common_types::{hashing, shard::Shard, Epoch, ExtraData, NodeHeight, NumPreshards, ShardGroup}; +use tari_state_tree::{compute_merkle_root_for_hashes, StateTreeError}; +#[cfg(feature = "ts")] +use ts_rs::TS; + +use super::{BlockError, BlockId, QcId, QuorumCertificate, ValidatorSchnorrSignature}; +use crate::consensus_models::{Command, LastExecuted, LastProposed, LastVoted, LeafBlock, LockedBlock}; + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[cfg_attr(feature = "ts", derive(TS), ts(export, export_to = "../../bindings/src/types/"))] +pub struct BlockHeader { + #[cfg_attr(feature = "ts", ts(type = "string"))] + id: BlockId, + #[cfg_attr(feature = "ts", ts(type = "string"))] + network: Network, + #[cfg_attr(feature = "ts", ts(type = "string"))] + parent: BlockId, + #[cfg_attr(feature = "ts", ts(type = "string"))] + justify_id: QcId, + height: NodeHeight, + epoch: Epoch, + shard_group: ShardGroup, + #[cfg_attr(feature = "ts", ts(type = "string"))] + proposed_by: PublicKey, + #[cfg_attr(feature = "ts", ts(type = "number"))] + total_leader_fee: u64, + #[cfg_attr(feature = "ts", ts(type = "string"))] + state_merkle_root: FixedHash, + #[cfg_attr(feature = "ts", ts(type = "string"))] + command_merkle_root: FixedHash, + /// If the block is a dummy block. This is metadata and not sent over + /// the wire or part of the block hash. + is_dummy: bool, + /// Counter for each foreign shard for reliable broadcast. + foreign_indexes: IndexMap, + /// Signature of block by the proposer. + #[cfg_attr(feature = "ts", ts(type = "{public_nonce : string, signature: string} | null"))] + signature: Option, + #[cfg_attr(feature = "ts", ts(type = "number"))] + timestamp: u64, + #[cfg_attr(feature = "ts", ts(type = "number"))] + base_layer_block_height: u64, + #[cfg_attr(feature = "ts", ts(type = "string"))] + base_layer_block_hash: FixedHash, + extra_data: ExtraData, +} + +impl BlockHeader { + #[allow(clippy::too_many_arguments)] + pub fn create( + network: Network, + parent: BlockId, + justify_id: QcId, + height: NodeHeight, + epoch: Epoch, + shard_group: ShardGroup, + proposed_by: PublicKey, + state_merkle_root: FixedHash, + commands: &BTreeSet, + total_leader_fee: u64, + sorted_foreign_indexes: IndexMap, + signature: Option, + timestamp: u64, + base_layer_block_height: u64, + base_layer_block_hash: FixedHash, + extra_data: ExtraData, + ) -> Result { + let command_merkle_root = compute_command_merkle_root(commands)?; + let mut header = BlockHeader { + id: BlockId::zero(), + network, + parent, + justify_id, + height, + epoch, + shard_group, + proposed_by, + state_merkle_root, + command_merkle_root, + total_leader_fee, + is_dummy: false, + foreign_indexes: sorted_foreign_indexes, + signature, + timestamp, + base_layer_block_height, + base_layer_block_hash, + extra_data, + }; + header.id = header.calculate_hash().into(); + + Ok(header) + } + + #[allow(clippy::too_many_arguments)] + pub fn load( + id: BlockId, + network: Network, + parent: BlockId, + justify_id: QcId, + height: NodeHeight, + epoch: Epoch, + shard_group: ShardGroup, + proposed_by: PublicKey, + state_merkle_root: FixedHash, + total_leader_fee: u64, + is_dummy: bool, + sorted_foreign_indexes: IndexMap, + signature: Option, + timestamp: u64, + base_layer_block_height: u64, + base_layer_block_hash: FixedHash, + extra_data: ExtraData, + command_merkle_root: FixedHash, + ) -> Self { + Self { + id, + network, + parent, + justify_id, + height, + epoch, + shard_group, + proposed_by, + state_merkle_root, + command_merkle_root, + total_leader_fee, + is_dummy, + foreign_indexes: sorted_foreign_indexes, + signature, + timestamp, + base_layer_block_height, + base_layer_block_hash, + extra_data, + } + } + + /// This is the parent block for all genesis blocks. Its block ID is always zero. + pub fn zero_block(network: Network, num_preshards: NumPreshards) -> Self { + Self { + network, + id: BlockId::zero(), + parent: BlockId::zero(), + justify_id: *QuorumCertificate::genesis(Epoch::zero(), ShardGroup::all_shards(num_preshards)).id(), + height: NodeHeight::zero(), + epoch: Epoch::zero(), + shard_group: ShardGroup::all_shards(num_preshards), + proposed_by: PublicKey::default(), + state_merkle_root: FixedHash::zero(), + command_merkle_root: FixedHash::zero(), + total_leader_fee: 0, + is_dummy: false, + foreign_indexes: IndexMap::new(), + signature: None, + timestamp: EpochTime::now().as_u64(), + base_layer_block_height: 0, + base_layer_block_hash: FixedHash::zero(), + extra_data: ExtraData::new(), + } + } + + pub fn dummy_block( + network: Network, + parent: BlockId, + proposed_by: PublicKey, + height: NodeHeight, + justify_id: QcId, + epoch: Epoch, + shard_group: ShardGroup, + parent_state_merkle_root: FixedHash, + parent_timestamp: u64, + parent_base_layer_block_height: u64, + parent_base_layer_block_hash: FixedHash, + ) -> Self { + let mut block = Self { + id: BlockId::zero(), + network, + parent, + justify_id, + height, + epoch, + shard_group, + proposed_by, + state_merkle_root: parent_state_merkle_root, + command_merkle_root: compute_command_merkle_root([].into_iter().peekable()) + .expect("compute_command_merkle_root is infallible for empty commands"), + total_leader_fee: 0, + is_dummy: true, + foreign_indexes: IndexMap::new(), + signature: None, + timestamp: parent_timestamp, + base_layer_block_height: parent_base_layer_block_height, + base_layer_block_hash: parent_base_layer_block_hash, + extra_data: ExtraData::new(), + }; + block.id = block.calculate_hash().into(); + block + } + + pub fn calculate_hash(&self) -> FixedHash { + // Hash is created from the hash of the "body" and + // then hashed with the parent, so that you can + // create a merkle proof of a chain of blocks + // ```pre + // root + // |\ + // | block1 + // |\ + // | block2 + // | + // blockbody + // ``` + + let inner_hash = hashing::block_hasher() + .chain(&self.network) + .chain(&self.justify_id) + .chain(&self.height) + .chain(&self.total_leader_fee) + .chain(&self.epoch) + .chain(&self.shard_group) + .chain(&self.proposed_by) + .chain(&self.state_merkle_root) + .chain(&self.is_dummy) + .chain(&self.command_merkle_root) + .chain(&self.foreign_indexes) + .chain(&self.timestamp) + .chain(&self.base_layer_block_height) + .chain(&self.base_layer_block_hash) + .chain(&self.extra_data) + .result(); + + hashing::block_hasher().chain(&self.parent).chain(&inner_hash).result() + } + + pub fn is_genesis(&self) -> bool { + self.height.is_zero() + } + + pub fn as_locked_block(&self) -> LockedBlock { + LockedBlock { + height: self.height, + block_id: self.id, + epoch: self.epoch, + } + } + + pub fn as_last_executed(&self) -> LastExecuted { + LastExecuted { + height: self.height, + block_id: self.id, + epoch: self.epoch, + } + } + + pub fn as_last_voted(&self) -> LastVoted { + LastVoted { + height: self.height, + block_id: self.id, + epoch: self.epoch, + } + } + + pub fn as_leaf_block(&self) -> LeafBlock { + LeafBlock { + height: self.height, + block_id: self.id, + epoch: self.epoch, + } + } + + pub fn as_last_proposed(&self) -> LastProposed { + LastProposed { + height: self.height, + block_id: self.id, + epoch: self.epoch, + } + } + + pub fn id(&self) -> &BlockId { + &self.id + } + + pub fn network(&self) -> Network { + self.network + } + + pub fn parent(&self) -> &BlockId { + &self.parent + } + + pub fn justify_id(&self) -> &QcId { + &self.justify_id + } + + pub fn height(&self) -> NodeHeight { + self.height + } + + pub fn epoch(&self) -> Epoch { + self.epoch + } + + pub fn shard_group(&self) -> ShardGroup { + self.shard_group + } + + pub fn total_leader_fee(&self) -> u64 { + self.total_leader_fee + } + + pub fn total_transaction_fee(&self) -> u64 { + self.total_leader_fee + } + + pub fn proposed_by(&self) -> &PublicKey { + &self.proposed_by + } + + pub fn state_merkle_root(&self) -> &FixedHash { + &self.state_merkle_root + } + + pub fn command_merkle_root(&self) -> &FixedHash { + &self.command_merkle_root + } + + pub fn is_dummy(&self) -> bool { + self.is_dummy + } + + pub fn get_foreign_counter(&self, bucket: &Shard) -> Option { + self.foreign_indexes.get(bucket).copied() + } + + pub fn foreign_indexes(&self) -> &IndexMap { + &self.foreign_indexes + } + + pub fn timestamp(&self) -> u64 { + self.timestamp + } + + pub fn signature(&self) -> Option<&ValidatorSchnorrSignature> { + self.signature.as_ref() + } + + pub fn set_signature(&mut self, signature: ValidatorSchnorrSignature) { + self.signature = Some(signature); + } + + pub fn base_layer_block_height(&self) -> u64 { + self.base_layer_block_height + } + + pub fn base_layer_block_hash(&self) -> &FixedHash { + &self.base_layer_block_hash + } + + pub fn extra_data(&self) -> &ExtraData { + &self.extra_data + } +} + +impl Display for BlockHeader { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + if self.is_dummy() { + write!(f, "Dummy")?; + } + write!( + f, + "[{}, {}, {}, {}->{}]", + self.height(), + self.epoch(), + self.shard_group(), + self.id(), + self.parent() + ) + } +} + +pub(crate) fn compute_command_merkle_root<'a, I: IntoIterator>( + commands: I, +) -> Result { + let hashes = commands.into_iter().map(|cmd| cmd.hash()).peekable(); + compute_merkle_root_for_hashes(hashes) +} diff --git a/dan_layer/storage/src/consensus_models/command.rs b/dan_layer/storage/src/consensus_models/command.rs index ec71257da..74f90a00d 100644 --- a/dan_layer/storage/src/consensus_models/command.rs +++ b/dan_layer/storage/src/consensus_models/command.rs @@ -7,8 +7,8 @@ use std::{ }; use serde::{Deserialize, Serialize}; -use tari_common_types::types::PublicKey; -use tari_dan_common_types::ShardGroup; +use tari_common_types::types::{FixedHash, PublicKey}; +use tari_dan_common_types::{hashing::command_hasher, ShardGroup}; use tari_engine_types::substate::SubstateId; use tari_transaction::TransactionId; @@ -114,6 +114,7 @@ pub enum Command { MintConfidentialOutput(MintConfidentialOutputAtom), SuspendNode(SuspendNodeAtom), ResumeNode(ResumeNodeAtom), + // EvictNode(EvictNodeAtom), EndEpoch, } @@ -122,6 +123,7 @@ pub enum Command { enum CommandOrdering<'a> { ResumeNode, SuspendNode, + // EvictNode, /// Foreign proposals should come first in the block so that they are processed before commands ForeignProposal(ShardGroup, &'a BlockId), MintConfidentialOutput(&'a SubstateId), @@ -169,6 +171,10 @@ impl Command { } } + pub fn hash(&self) -> FixedHash { + command_hasher().chain(self).result() + } + pub fn local_only(&self) -> Option<&TransactionAtom> { match self { Command::LocalOnly(tx) => Some(tx), @@ -330,13 +336,35 @@ pub struct ResumeNodeAtom { pub public_key: PublicKey, } -impl ResumeNodeAtom { +impl crate::consensus_models::ResumeNodeAtom { + pub fn delete_suspended_node(&self, tx: &mut TTx) -> Result<(), StorageError> { + tx.suspended_nodes_delete(&self.public_key) + } +} + +impl Display for crate::consensus_models::ResumeNodeAtom { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.public_key) + } +} +#[cfg_attr( + feature = "ts", + derive(ts_rs::TS), + ts(export, export_to = "../../bindings/src/types/") +)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +pub struct EvictNodeAtom { + #[cfg_attr(feature = "ts", ts(type = "string"))] + pub public_key: PublicKey, +} + +impl EvictNodeAtom { pub fn delete_suspended_node(&self, tx: &mut TTx) -> Result<(), StorageError> { tx.suspended_nodes_delete(&self.public_key) } } -impl Display for ResumeNodeAtom { +impl Display for EvictNodeAtom { fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { write!(f, "{}", self.public_key) } diff --git a/dan_layer/storage/src/consensus_models/epoch_checkpoint.rs b/dan_layer/storage/src/consensus_models/epoch_checkpoint.rs index ef886ebc5..5c5f5e887 100644 --- a/dan_layer/storage/src/consensus_models/epoch_checkpoint.rs +++ b/dan_layer/storage/src/consensus_models/epoch_checkpoint.rs @@ -5,7 +5,7 @@ use std::fmt::Display; use indexmap::IndexMap; use tari_dan_common_types::{shard::Shard, Epoch}; -use tari_state_tree::{Hash, SPARSE_MERKLE_PLACEHOLDER_HASH}; +use tari_state_tree::{compute_merkle_root_for_hashes, Hash, StateTreeError, SPARSE_MERKLE_PLACEHOLDER_HASH}; use crate::{ consensus_models::{Block, QuorumCertificate}, @@ -48,6 +48,15 @@ impl EpochCheckpoint { .copied() .unwrap_or(SPARSE_MERKLE_PLACEHOLDER_HASH) } + + pub fn compute_state_merkle_root(&self) -> Result { + let shard_group = self.block().shard_group(); + let hashes = shard_group + .shard_iter() + .map(|shard| self.get_shard_root(shard)) + .peekable(); + compute_merkle_root_for_hashes(hashes) + } } impl EpochCheckpoint { diff --git a/dan_layer/storage/src/consensus_models/mod.rs b/dan_layer/storage/src/consensus_models/mod.rs index 87177d3dd..5c0584fcb 100644 --- a/dan_layer/storage/src/consensus_models/mod.rs +++ b/dan_layer/storage/src/consensus_models/mod.rs @@ -3,6 +3,7 @@ mod block; mod block_diff; +mod block_header; mod block_pledges; mod burnt_utxo; mod command; @@ -43,6 +44,7 @@ mod vote_signature; pub use block::*; pub use block_diff::*; +pub use block_header::*; pub use block_pledges::*; pub use burnt_utxo::*; pub use command::*; diff --git a/dan_layer/storage/src/consensus_models/no_vote.rs b/dan_layer/storage/src/consensus_models/no_vote.rs index d70d7e10f..f2598649b 100644 --- a/dan_layer/storage/src/consensus_models/no_vote.rs +++ b/dan_layer/storage/src/consensus_models/no_vote.rs @@ -48,8 +48,10 @@ pub enum NoVoteReason { NotEndOfEpoch, #[error("The node is not at the end of the epoch and other commands are present")] EndOfEpochWithOtherCommands, - #[error("The Merkle root does not match")] - MerkleRootMismatch, + #[error("The state Merkle root does not match")] + StateMerkleRootMismatch, + #[error("The command Merkle root does not match")] + CommandMerkleRootMismatch, #[error("Not all foreign input pledges are present")] NotAllForeignInputPledges, #[error("Leader proposed to suspend a node that should not be suspended")] @@ -87,7 +89,8 @@ impl NoVoteReason { Self::NotEndOfEpoch => "NotEndOfEpoch", Self::EndOfEpochWithOtherCommands => "EndOfEpochWithOtherCommands", Self::TotalLeaderFeeDisagreement => "TotalLeaderFeeDisagreement", - Self::MerkleRootMismatch => "MerkleRootMismatch", + Self::StateMerkleRootMismatch => "StateMerkleRootMismatch", + Self::CommandMerkleRootMismatch => "CommandMerkleRootMismatch", Self::NotAllForeignInputPledges => "NotAllForeignInputPledges", Self::ShouldNotSuspendNode => "ShouldNotSuspendNode", Self::NodeAlreadySuspended => "NodeAlreadySuspended", diff --git a/dan_layer/storage/src/state_store/mod.rs b/dan_layer/storage/src/state_store/mod.rs index 831343b32..b10ca688e 100644 --- a/dan_layer/storage/src/state_store/mod.rs +++ b/dan_layer/storage/src/state_store/mod.rs @@ -174,9 +174,10 @@ pub trait StateStoreReadTransaction: Sized { &self, epoch: Epoch, shard_group: ShardGroup, - start_block_id: &BlockId, - end_block_id: &BlockId, + start_block_height: NodeHeight, + end_block_height: NodeHeight, include_dummy_blocks: bool, + limit: u64, ) -> Result, StorageError>; fn blocks_exists(&self, block_id: &BlockId) -> Result; fn blocks_is_ancestor(&self, descendant: &BlockId, ancestor: &BlockId) -> Result; diff --git a/integration_tests/tests/features/state_sync.feature b/integration_tests/tests/features/state_sync.feature index ab6875688..9a59d66d1 100644 --- a/integration_tests/tests/features/state_sync.feature +++ b/integration_tests/tests/features/state_sync.feature @@ -50,10 +50,9 @@ Feature: State Sync When I wait for validator VN has leaf block height of at least 1 at epoch 3 When I wait for validator VN2 has leaf block height of at least 1 at epoch 3 - # FIXME: These steps fail (because VN2 does not participate timeously in consensus) -# When I create an account UNUSED4 via the wallet daemon WALLET_D -# When I create an account UNUSED5 via the wallet daemon WALLET_D -# -# When I wait for validator VN has leaf block height of at least 5 at epoch 3 -# When I wait for validator VN2 has leaf block height of at least 5 at epoch 3 + When I create an account UNUSED4 via the wallet daemon WALLET_D + When I create an account UNUSED5 via the wallet daemon WALLET_D + + When I wait for validator VN has leaf block height of at least 5 at epoch 3 + When I wait for validator VN2 has leaf block height of at least 5 at epoch 3 diff --git a/networking/core/src/spawn.rs b/networking/core/src/spawn.rs index ade70db52..19a258614 100644 --- a/networking/core/src/spawn.rs +++ b/networking/core/src/spawn.rs @@ -109,10 +109,10 @@ impl MessagingMode { .topic .as_str() .split_once(TOPIC_DELIMITER) - .ok_or(GossipSendError::InvalidToken(msg.topic.clone().into_string()))?; + .ok_or_else(|| GossipSendError::InvalidToken(msg.topic.clone().into_string()))?; let tx_gossip_messages = tx_gossip_messages_by_topic .get(prefix) - .ok_or(GossipSendError::InvalidToken(msg.topic.clone().into_string()))?; + .ok_or_else(|| GossipSendError::InvalidToken(msg.topic.clone().into_string()))?; tx_gossip_messages.send((peer_id, msg))?; } Ok(()) diff --git a/networking/core/src/worker.rs b/networking/core/src/worker.rs index 601eca386..af0fc5b52 100644 --- a/networking/core/src/worker.rs +++ b/networking/core/src/worker.rs @@ -633,22 +633,16 @@ where }, } } else { - match self.messaging_mode.send_gossip_message(source, message) { - Ok(_) => { - self.swarm.behaviour_mut().gossipsub.report_message_validation_result( - &message_id, - &propagation_source, - gossipsub::MessageAcceptance::Accept, - )?; - }, - Err(e) => { - warn!(target: LOG_TARGET, "📢 Gossipsub message failed to be handled: {}", e); - self.swarm.behaviour_mut().gossipsub.report_message_validation_result( - &message_id, - &propagation_source, - gossipsub::MessageAcceptance::Reject, - )?; - }, + // We accept all messages as we cannot validate them in this service. + // We could allow users to report back the validation result e.g. if a proposal is valid, however a naive + // implementation would likely incur a substantial cost for many messages. + self.swarm.behaviour_mut().gossipsub.report_message_validation_result( + &message_id, + &propagation_source, + gossipsub::MessageAcceptance::Accept, + )?; + if let Err(e) = self.messaging_mode.send_gossip_message(source, message) { + warn!(target: LOG_TARGET, "📢 Gossipsub message failed to be handled: {}", e); } } Ok(())