get_current_block_color RPC utility (kaspanet#528)

* getCurrentBlockColor algorithm and RPC functions

* Add a small comment over RPC test

* Move get_current_block_color to consensus and apply standard sorting

* Apply msuttons suggestions except the block check

* Remove not  needed return and format

* Variable name consistency

* Check for block existence on get_current_block_color

* Add extra DAG order checks to ensure about children

* includes:
1. stylistic changes using ?
2. `is_dag_ancestor_of(a, b)` is different than `!is_dag_ancestor_of(b, a)` -- they are not negations of each other, bcs there's also the anticone

* 1. bug fix: hash -> child
2. make store calls only where they are actually used (within the if)

* style: 1. use struct unfloding syntax, 2. use a name such as decedent which reflects the relation to `hash`

* important note

* Fix Omega compatibility issues

* Remove Borsh derivations

* Fix gRPC message codes

* Fix gRPC getCurrentBlockColorResponse

* improve tests

---------

Co-authored-by: Michael Sutton <[email protected]>

cli/src/modules/rpc.rs

@@ -249,6 +249,15 @@ impl Rpc {
  let result = rpc.get_fee_estimate_experimental_call(None, GetFeeEstimateExperimentalRequest { verbose }).await?;
  self.println(&ctx, result);
  }
+ RpcApiOps::GetCurrentBlockColor => {
+ if argv.is_empty() {
+ return Err(Error::custom("Missing block hash argument"));
+ }
+ let hash = argv.remove(0);
+ let hash = RpcHash::from_hex(hash.as_str())?;
+ let result = rpc.get_current_block_color_call(None, GetCurrentBlockColorRequest { hash }).await?;
+ self.println(&ctx, result);
+ }
  _ => {
  tprintln!(ctx, "rpc method exists but is not supported by the cli: '{op_str}'\r\n");
  return Ok(());

components/consensusmanager/src/session.rs

@@ -247,6 +247,10 @@ impl ConsensusSessionOwned {
  self.clone().spawn_blocking(|c| c.get_sink_timestamp()).await
  }
 
+ pub async fn async_get_current_block_color(&self, hash: Hash) -> Option<bool> {
+ self.clone().spawn_blocking(move |c| c.get_current_block_color(hash)).await
+ }
+
  /// source refers to the earliest block from which the current node has full header & block data 
  pub async fn async_get_source(&self) -> Hash {
  self.clone().spawn_blocking(|c| c.get_source()).await

consensus/core/src/api/mod.rs

@@ -133,6 +133,10 @@ pub trait ConsensusApi: Send + Sync {
  unimplemented!()
  }
 
+ fn get_current_block_color(&self, hash: Hash) -> Option<bool> {
+ unimplemented!()
+ }
+
  fn get_virtual_state_approx_id(&self) -> VirtualStateApproxId {
  unimplemented!()
  }

consensus/src/consensus/mod.rs

@@ -36,7 +36,10 @@ use crate::{
  virtual_processor::{errors::PruningImportResult, VirtualStateProcessor},
  ProcessingCounters,
  },
- processes::window::{WindowManager, WindowType},
+ processes::{
+ ghostdag::ordering::SortableBlock,
+ window::{WindowManager, WindowType},
+ },
 };
 use kaspa_consensus_core::{
  acceptance_data::AcceptanceData,
@@ -64,7 +67,7 @@ use kaspa_consensus_core::{
  pruning::{PruningPointProof, PruningPointTrustedData, PruningPointsList},
  trusted::{ExternalGhostdagData, TrustedBlock},
  tx::{MutableTransaction, Transaction, TransactionOutpoint, UtxoEntry},
- BlockHashSet, BlueWorkType, ChainPath,
+ BlockHashSet, BlueWorkType, ChainPath, HashMapCustomHasher,
 };
 use kaspa_consensus_notify::root::ConsensusNotificationRoot;
 
@@ -80,6 +83,8 @@ use kaspa_muhash::MuHash;
 use kaspa_txscript::caches::TxScriptCacheCounters;
 
 use std::{
+ cmp::Reverse,
+ collections::BinaryHeap,
  future::Future,
  iter::once,
  ops::Deref,
@@ -505,6 +510,64 @@ impl ConsensusApi for Consensus {
  self.headers_store.get_timestamp(self.get_sink()).unwrap()
  }
 
+ fn get_current_block_color(&self, hash: Hash) -> Option<bool> {
+ let _guard = self.pruning_lock.blocking_read();
+
+ // Verify the block exists and can be assumed to have relations and reachability data
+ self.validate_block_exists(hash).ok()?;
+
+ // Verify that the block is in future(source), where Ghostdag data is complete
+ self.services.reachability_service.is_dag_ancestor_of(self.get_source(), hash).then_some(())?;
+
+ let sink = self.get_sink();
+
+ // Optimization: verify that the block is in past(sink), otherwise the search will fail anyway
+ // (means the block was not merged yet by a virtual chain block)
+ self.services.reachability_service.is_dag_ancestor_of(hash, sink).then_some(())?;
+
+ let mut heap: BinaryHeap<Reverse<SortableBlock>> = BinaryHeap::new();
+ let mut visited = BlockHashSet::new();
+
+ let initial_children = self.get_block_children(hash).unwrap();
+
+ for child in initial_children {
+ if visited.insert(child) {
+ let blue_work = self.ghostdag_primary_store.get_blue_work(child).unwrap();
+ heap.push(Reverse(SortableBlock::new(child, blue_work)));
+ }
+ }
+
+ while let Some(Reverse(SortableBlock { hash: decedent, .. })) = heap.pop() {
+ if self.services.reachability_service.is_chain_ancestor_of(decedent, sink) {
+ let decedent_data = self.get_ghostdag_data(decedent).unwrap();
+
+ if decedent_data.mergeset_blues.contains(&hash) {
+ return Some(true);
+ } else if decedent_data.mergeset_reds.contains(&hash) {
+ return Some(false);
+ }
+
+ // Note: because we are doing a topological BFS up (from `hash` towards virtual), the first chain block
+ // found must also be our merging block, so hash will be either in blues or in reds, rendering this line
+ // unreachable.
+ kaspa_core::warn!("DAG topology inconsistency: {decedent} is expected to be a merging block of {hash}");
+ // TODO: we should consider the option of returning Result<Option<bool>> from this method
+ return None;
+ }
+
+ let children = self.get_block_children(decedent).unwrap();
+
+ for child in children {
+ if visited.insert(child) {
+ let blue_work = self.ghostdag_primary_store.get_blue_work(child).unwrap();
+ heap.push(Reverse(SortableBlock::new(child, blue_work)));
+ }
+ }
+ }
+
+ None
+ }
+
  fn get_virtual_state_approx_id(&self) -> VirtualStateApproxId {
  self.lkg_virtual_state.load().to_virtual_state_approx_id()
  }

rpc/core/src/api/ops.rs

@@ -130,6 +130,8 @@ pub enum RpcApiOps {
  GetFeeEstimate = 147,
  /// Fee estimation (experimental)
  GetFeeEstimateExperimental = 148,
+ /// Block color determination by iterating DAG.
+ GetCurrentBlockColor = 149,
 }
 
 impl RpcApiOps {

rpc/core/src/api/rpc.rs

@@ -457,6 +457,16 @@ pub trait RpcApi: Sync + Send + AnySync {
  request: GetFeeEstimateExperimentalRequest,
  ) -> RpcResult<GetFeeEstimateExperimentalResponse>;
 
+ ///
+ async fn get_current_block_color(&self, hash: RpcHash) -> RpcResult<GetCurrentBlockColorResponse> {
+ Ok(self.get_current_block_color_call(None, GetCurrentBlockColorRequest { hash }).await?)
+ }
+ async fn get_current_block_color_call(
+ &self,
+ connection: Option<&DynRpcConnection>,
+ request: GetCurrentBlockColorRequest,
+ ) -> RpcResult<GetCurrentBlockColorResponse>;
+
  // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  // Notification API
 

rpc/core/src/error.rs

@@ -77,6 +77,9 @@ pub enum RpcError {
  #[error("IP {0} is not registered as banned.")]
  IpIsNotBanned(IpAddress),
 
+ #[error("Block {0} doesn't have any merger block.")]
+ MergerNotFound(RpcHash),
+
  #[error("Block was not submitted: {0}")]
  SubmitBlockError(SubmitBlockRejectReason),
 

rpc/core/src/model/message.rs

@@ -2613,6 +2613,54 @@ impl Deserializer for GetFeeEstimateExperimentalResponse {
  }
 }
 
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GetCurrentBlockColorRequest {
+ pub hash: RpcHash,
+}
+
+impl Serializer for GetCurrentBlockColorRequest {
+ fn serialize<W: std::io::Write>(&self, writer: &mut W) -> std::io::Result<()> {
+ store!(u16, &1, writer)?;
+ store!(RpcHash, &self.hash, writer)?;
+
+ Ok(())
+ }
+}
+
+impl Deserializer for GetCurrentBlockColorRequest {
+ fn deserialize<R: std::io::Read>(reader: &mut R) -> std::io::Result<Self> {
+ let _version = load!(u16, reader)?;
+ let hash = load!(RpcHash, reader)?;
+
+ Ok(Self { hash })
+ }
+}
+
+#[derive(Clone, Debug, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct GetCurrentBlockColorResponse {
+ pub blue: bool,
+}
+
+impl Serializer for GetCurrentBlockColorResponse {
+ fn serialize<W: std::io::Write>(&self, writer: &mut W) -> std::io::Result<()> {
+ store!(u16, &1, writer)?;
+ store!(bool, &self.blue, writer)?;
+
+ Ok(())
+ }
+}
+
+impl Deserializer for GetCurrentBlockColorResponse {
+ fn deserialize<R: std::io::Read>(reader: &mut R) -> std::io::Result<Self> {
+ let _version = load!(u16, reader)?;
+ let blue = load!(bool, reader)?;
+
+ Ok(Self { blue })
+ }
+}
+
 // ----------------------------------------------------------------------------
 // Subscriptions & notifications
 // ----------------------------------------------------------------------------

rpc/core/src/wasm/message.rs

@@ -856,6 +856,44 @@ try_from! ( args: GetBlockTemplateResponse, IGetBlockTemplateResponse, {
 
 // ---
 
+declare! {
+ IGetCurrentBlockColorRequest,
+ r#"
+ /**
+ * 
+ * 
+ * @category Node RPC
+ */
+ export interface IGetCurrentBlockColorRequest {
+ hash: HexString;
+ }
+ "#,
+}
+
+try_from! ( args: IGetCurrentBlockColorRequest, GetCurrentBlockColorRequest, {
+ Ok(from_value(args.into())?)
+});
+
+declare! {
+ IGetCurrentBlockColorResponse,
+ r#"
+ /**
+ * 
+ * 
+ * @category Node RPC
+ */
+ export interface IGetCurrentBlockColorResponse {
+ blue: boolean;
+ }
+ "#,
+}
+
+try_from! ( args: GetCurrentBlockColorResponse, IGetCurrentBlockColorResponse, {
+ Ok(to_value(&args)?.into())
+});
+
+// ---
+
 declare! {
  IGetDaaScoreTimestampEstimateRequest,
  r#"

rpc/grpc/client/src/lib.rs

@@ -276,6 +276,7 @@ impl RpcApi for GrpcClient {
  route!(get_daa_score_timestamp_estimate_call, GetDaaScoreTimestampEstimate);
  route!(get_fee_estimate_call, GetFeeEstimate);
  route!(get_fee_estimate_experimental_call, GetFeeEstimateExperimental);
+ route!(get_current_block_color_call, GetCurrentBlockColor);
 
  // ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
  // Notification API

rpc/grpc/core/proto/messages.proto

@@ -64,6 +64,7 @@ message KaspadRequest {
  GetSystemInfoRequestMessage getSystemInfoRequest = 1104;
  GetFeeEstimateRequestMessage getFeeEstimateRequest = 1106;
  GetFeeEstimateExperimentalRequestMessage getFeeEstimateExperimentalRequest = 1108;
+ GetCurrentBlockColorRequestMessage getCurrentBlockColorRequest = 1110;
  }
 }
 
@@ -128,6 +129,7 @@ message KaspadResponse {
  GetSystemInfoResponseMessage getSystemInfoResponse= 1105;
  GetFeeEstimateResponseMessage getFeeEstimateResponse = 1107;
  GetFeeEstimateExperimentalResponseMessage getFeeEstimateExperimentalResponse = 1109;
+ GetCurrentBlockColorResponseMessage getCurrentBlockColorResponse = 1111;
  }
 }
 

rpc/grpc/core/proto/rpc.proto

@@ -959,3 +959,13 @@ message GetFeeEstimateExperimentalResponseMessage {
 
  RPCError error = 1000;
 }
+
+message GetCurrentBlockColorRequestMessage {
+ string hash = 1;
+}
+
+message GetCurrentBlockColorResponseMessage {
+ bool blue = 1;
+
+ RPCError error = 1000;
+}

rpc/grpc/core/src/convert/kaspad.rs

@@ -62,6 +62,7 @@ pub mod kaspad_request_convert {
  impl_into_kaspad_request!(GetDaaScoreTimestampEstimate);
  impl_into_kaspad_request!(GetFeeEstimate);
  impl_into_kaspad_request!(GetFeeEstimateExperimental);
+ impl_into_kaspad_request!(GetCurrentBlockColor);
 
  impl_into_kaspad_request!(NotifyBlockAdded);
  impl_into_kaspad_request!(NotifyNewBlockTemplate);
@@ -198,6 +199,7 @@ pub mod kaspad_response_convert {
  impl_into_kaspad_response!(GetDaaScoreTimestampEstimate);
  impl_into_kaspad_response!(GetFeeEstimate);
  impl_into_kaspad_response!(GetFeeEstimateExperimental);
+ impl_into_kaspad_response!(GetCurrentBlockColor);
 
  impl_into_kaspad_notify_response!(NotifyBlockAdded);
  impl_into_kaspad_notify_response!(NotifyNewBlockTemplate);

rpc/grpc/core/src/convert/message.rs

@@ -421,6 +421,15 @@ from!(item: RpcResult<&kaspa_rpc_core::GetFeeEstimateExperimentalResponse>, prot
  }
 });
 
+from!(item: &kaspa_rpc_core::GetCurrentBlockColorRequest, protowire::GetCurrentBlockColorRequestMessage, {
+ Self {
+ hash: item.hash.to_string()
+ }
+});
+from!(item: RpcResult<&kaspa_rpc_core::GetCurrentBlockColorResponse>, protowire::GetCurrentBlockColorResponseMessage, {
+ Self { blue: item.blue, error: None }
+});
+
 from!(&kaspa_rpc_core::PingRequest, protowire::PingRequestMessage);
 from!(RpcResult<&kaspa_rpc_core::PingResponse>, protowire::PingResponseMessage);
 
@@ -896,6 +905,17 @@ try_from!(item: &protowire::GetFeeEstimateExperimentalResponseMessage, RpcResult
  }
 });
 
+try_from!(item: &protowire::GetCurrentBlockColorRequestMessage, kaspa_rpc_core::GetCurrentBlockColorRequest, {
+ Self {
+ hash: RpcHash::from_str(&item.hash)?
+ }
+});
+try_from!(item: &protowire::GetCurrentBlockColorResponseMessage, RpcResult<kaspa_rpc_core::GetCurrentBlockColorResponse>, {
+ Self {
+ blue: item.blue
+ }
+});
+
 try_from!(&protowire::PingRequestMessage, kaspa_rpc_core::PingRequest);
 try_from!(&protowire::PingResponseMessage, RpcResult<kaspa_rpc_core::PingResponse>);
 

rpc/grpc/core/src/ops.rs

@@ -86,6 +86,7 @@ pub enum KaspadPayloadOps {
  GetDaaScoreTimestampEstimate,
  GetFeeEstimate,
  GetFeeEstimateExperimental,
+ GetCurrentBlockColor,
 
  // Subscription commands for starting/stopping notifications
  NotifyBlockAdded,

rpc/grpc/server/src/request_handler/factory.rs

@@ -80,6 +80,7 @@ impl Factory {
  GetDaaScoreTimestampEstimate,
  GetFeeEstimate,
  GetFeeEstimateExperimental,
+ GetCurrentBlockColor,
  NotifyBlockAdded,
  NotifyNewBlockTemplate,
  NotifyFinalityConflict,

rpc/grpc/server/src/tests/rpc_core_mock.rs

@@ -230,6 +230,14 @@ impl RpcApi for RpcCoreMock {
  Err(RpcError::NotImplemented)
  }
 
+ async fn get_current_block_color_call(
+ &self,
+ _connection: Option<&DynRpcConnection>,
+ _request: GetCurrentBlockColorRequest,
+ ) -> RpcResult<GetCurrentBlockColorResponse> {
+ Err(RpcError::NotImplemented)
+ }
+
  async fn get_block_count_call(
  &self,
  _connection: Option<&DynRpcConnection>,