Repository: dbpunk-labs/db3 (language: rust)

Issue:
LightStore: add block subscribe and block fetch api
## Motivations

<!--
If your feature request is related to a problem, please describe it.
-->
Add block subscribe for light storage node and support API to fetch block/mutation based on block height
- Would you like to implement this feature? [y/n]

## Solution

<!-- Describe the solution you'd like. -->

## Alternatives

<!-- Describe any alternative solutions or features you've considered. -->

## Additional context

<!-- Add any other context or screenshots about the feature request here. -->



Discussion / hints:

Fix patch:
diff --git a/src/node/src/storage_node_light_impl.rs b/src/node/src/storage_node_light_impl.rs
index dc577756..10b25b35 100644
--- a/src/node/src/storage_node_light_impl.rs
+++ b/src/node/src/storage_node_light_impl.rs
@@ -24,15 +24,16 @@ use db3_error::Result;
 use db3_proto::db3_mutation_v2_proto::{
     mutation::body_wrapper::Body, MutationAction, MutationRollupStatus,
 };
+use db3_proto::db3_storage_proto::block_response;
 use db3_proto::db3_storage_proto::event_message::Event as EventV2;
 use db3_proto::db3_storage_proto::{
-    storage_node_server::StorageNode, ExtraItem, GetCollectionOfDatabaseRequest,
-    GetCollectionOfDatabaseResponse, GetDatabaseOfOwnerRequest, GetDatabaseOfOwnerResponse,
-    GetMutationBodyRequest, GetMutationBodyResponse, GetMutationHeaderRequest,
-    GetMutationHeaderResponse, GetNonceRequest, GetNonceResponse, ScanGcRecordRequest,
-    ScanGcRecordResponse, ScanMutationHeaderRequest, ScanMutationHeaderResponse,
-    ScanRollupRecordRequest, ScanRollupRecordResponse, SendMutationRequest, SendMutationResponse,
-    SubscribeRequest,
+    storage_node_server::StorageNode, BlockRequest, BlockResponse, ExtraItem,
+    GetCollectionOfDatabaseRequest, GetCollectionOfDatabaseResponse, GetDatabaseOfOwnerRequest,
+    GetDatabaseOfOwnerResponse, GetMutationBodyRequest, GetMutationBodyResponse,
+    GetMutationHeaderRequest, GetMutationHeaderResponse, GetNonceRequest, GetNonceResponse,
+    ScanGcRecordRequest, ScanGcRecordResponse, ScanMutationHeaderRequest,
+    ScanMutationHeaderResponse, ScanRollupRecordRequest, ScanRollupRecordResponse,
+    SendMutationRequest, SendMutationResponse, SubscribeRequest,
 };
 
 use db3_proto::db3_storage_proto::{
@@ -55,7 +56,7 @@ use tokio::task;
 use tokio::time::{sleep, Duration as TokioDuration};
 use tokio_stream::wrappers::ReceiverStream;
 use tonic::{Request, Response, Status};
-use tracing::{debug, info, warn};
+use tracing::{info, warn};
 pub struct StorageNodeV2Config {
     pub store_config: MutationStoreConfig,
     pub state_config: StateStoreConfig,
@@ -296,6 +297,25 @@ impl StorageNode for StorageNodeV2Impl {
         Ok(Response::new(ReceiverStream::new(msg_receiver)))
     }
 
+    async fn get_block(
+        &self,
+        request: Request<BlockRequest>,
+    ) -> std::result::Result<Response<BlockResponse>, Status> {
+        let r = request.into_inner();
+        let mutation_header_bodys = self
+            .storage
+            .get_range_mutations(r.block_start, r.block_end)
+            .map_err(|e| Status::internal(format!("{e}")))?;
+        let mutations = mutation_header_bodys
+            .iter()
+            .map(|(h, b)| block_response::MutationWrapper {
+                header: Some(h.to_owned()),
+                body: Some(b.to_owned()),
+            })
+            .collect();
+        Ok(Response::new(BlockResponse { mutations }))
+    }
+
     async fn get_collection_of_database(
         &self,
         request: Request<GetCollectionOfDatabaseRequest>,
diff --git a/src/proto/proto/db3_storage.proto b/src/proto/proto/db3_storage.proto
index 82a48b97..3f98ed14 100644
--- a/src/proto/proto/db3_storage.proto
+++ b/src/proto/proto/db3_storage.proto
@@ -140,7 +140,17 @@ message ScanGcRecordRequest {
 message ScanGcRecordResponse {
   repeated db3_rollup_proto.GcRecord records = 1;
 }
-
+message BlockRequest {
+  uint64 block_start = 1;
+  uint64 block_end = 2;
+}
+message BlockResponse {
+  message MutationWrapper {
+    db3_mutation_v2_proto.MutationHeader header = 1;
+    db3_mutation_v2_proto.MutationBody body = 2;
+  }
+  repeated MutationWrapper mutations = 3;
+}
 service StorageNode {
   rpc SendMutation(SendMutationRequest) returns (SendMutationResponse) {}
   rpc GetNonce(GetNonceRequest) returns (GetNonceResponse) {}
@@ -152,4 +162,6 @@ service StorageNode {
   rpc GetCollectionOfDatabase(GetCollectionOfDatabaseRequest) returns (GetCollectionOfDatabaseResponse) {}
   rpc ScanGcRecord(ScanGcRecordRequest) returns (ScanGcRecordResponse) {}
   rpc Subscribe(SubscribeRequest) returns (stream EventMessage) {}
+  // method for get block
+  rpc GetBlock(BlockRequest) returns (BlockResponse) {}
 }
diff --git a/src/sdk/src/store_sdk_v2.rs b/src/sdk/src/store_sdk_v2.rs
index 44f50f76..0c2ef43c 100644
--- a/src/sdk/src/store_sdk_v2.rs
+++ b/src/sdk/src/store_sdk_v2.rs
@@ -21,6 +21,7 @@ use db3_proto::db3_storage_proto::{
     storage_node_client::StorageNodeClient as StorageNodeV2Client, SubscribeRequest,
 };
 use db3_proto::db3_storage_proto::{
+    BlockRequest as BlockRequestV2, BlockResponse as BlockResponseV2,
     EventMessage as EventMessageV2, EventType as EventTypeV2, Subscription as SubscriptionV2,
 };
 
@@ -62,6 +63,18 @@ impl StoreSDKV2 {
         let mut client = self.client.as_ref().clone();
         client.subscribe(req).await
     }
+
+    pub async fn get_block_by_height(
+        &mut self,
+        height: u64,
+    ) -> Result<tonic::Response<BlockResponseV2>, Status> {
+        let req = BlockRequestV2 {
+            block_start: height,
+            block_end: height + 1,
+        };
+        let mut client = self.client.as_ref().clone();
+        client.get_block(req).await
+    }
 }
 
 #[cfg(test)]
@@ -73,7 +86,6 @@ mod tests {
     use tonic::transport::Endpoint;
 
     async fn subscribe_event_message_flow(
-        use_typed_format: bool,
         client: Arc<StorageNodeV2Client<tonic::transport::Channel>>,
         counter: i64,
     ) {
@@ -96,6 +108,18 @@ mod tests {
         }
     }
 
+    async fn get_block_by_height_flow(
+        client: Arc<StorageNodeV2Client<tonic::transport::Channel>>,
+        counter: i64,
+        height: u64,
+    ) {
+        let (_, signer) = sdk_test::gen_secp256k1_signer(counter);
+        let mut sdk = StoreSDKV2::new(client, signer);
+        let res = sdk.get_block_by_height(height).await;
+        println!("res {:?}", res);
+        assert!(res.is_ok(), "{:?}", res);
+    }
+
     #[tokio::test]
     async fn subscribe_event_message_ut() {
         let ep = "http://127.0.0.1:26619";
@@ -103,6 +127,14 @@ mod tests {
         let channel = rpc_endpoint.connect_lazy();
         let client = Arc::new(StorageNodeV2Client::new(channel));
 
-        subscribe_event_message_flow(false, client.clone(), 300).await;
+        subscribe_event_message_flow(client.clone(), 300).await;
+    }
+    #[tokio::test]
+    async fn get_block_by_height_ut() {
+        let ep = "http://127.0.0.1:26619";
+        let rpc_endpoint = Endpoint::new(ep.to_string()).unwrap();
+        let channel = rpc_endpoint.connect_lazy();
+        let client = Arc::new(StorageNodeV2Client::new(channel));
+        get_block_by_height_flow(client.clone(), 301, 1).await;
     }
 }


Test patch: