refactor: storage (#32)

* rebase to 0g-storage-node log-entry-sync

* modified log-entry-sync

* feat: sync log-entry-sync

* update: dependency

* feat: update log entry sycn

* refactor: storage

* fix: flow store

* chore: update dependency

Cargo.toml

@@ -16,11 +16,11 @@ enr = { path = "version-meld/enr" }
 discv5 = { path = "version-meld/discv5" }
 
 [workspace.dependencies]
-append_merkle = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "da2cdec8a1ddfd52333a6c10141df31611d0e1f8" }
-merkle_light = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "da2cdec8a1ddfd52333a6c10141df31611d0e1f8" }
-merkle_tree = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "da2cdec8a1ddfd52333a6c10141df31611d0e1f8" }
-shared_types = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "da2cdec8a1ddfd52333a6c10141df31611d0e1f8" }
-task_executor = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "da2cdec8a1ddfd52333a6c10141df31611d0e1f8" }
-storage = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "da2cdec8a1ddfd52333a6c10141df31611d0e1f8", package = "storage" }
-contract-interface = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "da2cdec8a1ddfd52333a6c10141df31611d0e1f8" }
+append_merkle = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "96f846073f111adfa31df926be0e4a25e0b85f54" }
+merkle_light = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "96f846073f111adfa31df926be0e4a25e0b85f54" }
+merkle_tree = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "96f846073f111adfa31df926be0e4a25e0b85f54" }
+shared_types = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "96f846073f111adfa31df926be0e4a25e0b85f54" }
+task_executor = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "96f846073f111adfa31df926be0e4a25e0b85f54" }
+storage = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "96f846073f111adfa31df926be0e4a25e0b85f54", package = "storage" }
+contract-interface = { git = "https://github.com/0glabs/0g-storage-node.git",rev = "96f846073f111adfa31df926be0e4a25e0b85f54" }
 metrics = { git = "https://github.com/Conflux-Chain/conflux-rust.git", rev = "992ebc5483d937c8f6b883e266f8ed2a67a7fa9a" }

node/kv_types/src/lib.rs

@@ -1,9 +1,11 @@
 use ethereum_types::{H160, H256};
 
 use serde::{Deserialize, Serialize};
-use shared_types::Transaction;
+use shared_types::{DataRoot, Transaction, TxID};
+use ssz::Encode;
 use ssz_derive::{Decode as DeriveDecode, Encode as DeriveEncode};
 use std::collections::HashSet;
+use tiny_keccak::{Hasher, Keccak};
 
 use std::sync::Arc;
 
@@ -29,16 +31,52 @@ pub fn submission_topic_to_stream_ids(topic: Vec<u8>) -> Vec<H256> {
 
 #[derive(Clone, Debug, Eq, PartialEq, DeriveDecode, DeriveEncode, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
-pub struct KVMetadata {
+pub struct KVTransaction {
     pub stream_ids: Vec<H256>,
     pub sender: H160,
+    pub data_merkle_root: DataRoot,
+    /// `(subtree_depth, subtree_root)`
+    pub merkle_nodes: Vec<(usize, DataRoot)>,
+
+    pub start_entry_index: u64,
+    pub size: u64,
+    pub seq: u64,
 }
 
-#[derive(Clone, Debug, Eq, PartialEq, DeriveDecode, DeriveEncode, Deserialize, Serialize)]
-#[serde(rename_all = "camelCase")]
-pub struct KVTransaction {
-    pub metadata: KVMetadata,
-    pub transaction: Transaction,
+impl KVTransaction {
+    pub fn num_entries_of_node(depth: usize) -> usize {
+        1 << (depth - 1)
+    }
+
+    pub fn num_entries_of_list(merkle_nodes: &[(usize, DataRoot)]) -> usize {
+        merkle_nodes.iter().fold(0, |size, &(depth, _)| {
+            size + Transaction::num_entries_of_node(depth)
+        })
+    }
+
+    pub fn num_entries(&self) -> usize {
+        Self::num_entries_of_list(&self.merkle_nodes)
+    }
+
+    pub fn hash(&self) -> H256 {
+        let bytes = self.as_ssz_bytes();
+        let mut h = Keccak::v256();
+        let mut e = H256::zero();
+        h.update(&bytes);
+        h.finalize(e.as_mut());
+        e
+    }
+
+    pub fn id(&self) -> TxID {
+        TxID {
+            seq: self.seq,
+            hash: self.hash(),
+        }
+    }
+
+    pub fn start_entry_index(&self) -> u64 {
+        self.start_entry_index
+    }
 }
 
 #[derive(Debug)]

node/log_entry_sync/Cargo.toml

@@ -25,4 +25,6 @@ futures-core = "0.3.28"
 futures-util = "0.3.28"
 thiserror = "1.0.44"
 lazy_static = "1.4.0"
-metrics = { workspace = true }
+metrics = { workspace = true }
+reqwest = {version = "0.11", features = ["json"]}
+url = { version = "2.4", default-features = false }