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 <msutton@cs.huji.ac.il>

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>,