Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

refactor: root chunk_id should be equal to app_hash #301

Merged
merged 6 commits into from
Jun 3, 2024
Merged

refactor: root chunk_id should be equal to app_hash #301

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
eecadff
adjustements
ogabrielides Jun 3, 2024
dfb4244
fmt
ogabrielides Jun 3, 2024
9853ce7
clippy
ogabrielides Jun 3, 2024
b6d2871
clippy
ogabrielides Jun 3, 2024
48e67f4
fmt
ogabrielides Jun 3, 2024
83ecbb0
suggestions
ogabrielides Jun 3, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
62 changes: 27 additions & 35 deletions grovedb/src/replication.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,20 +32,15 @@ struct SubtreeStateSyncInfo<'db> {
num_processed_chunks: usize,
}

impl<'a> SubtreeStateSyncInfo<'a> {
// Function to create an instance of SubtreeStateSyncInfo with default values
pub fn new() -> Self {
Self::default()
}
}

// Struct governing state sync
pub struct MultiStateSyncInfo<'db> {
// Map of current processing subtrees
// SubtreePrefix (Path digest) -> SubtreeStateSyncInfo
current_prefixes: BTreeMap<SubtreePrefix, SubtreeStateSyncInfo<'db>>,
// Set of processed prefixes (Path digests)
processed_prefixes: BTreeSet<SubtreePrefix>,
// Root app_hash
app_hash: [u8; 32],
// Version of state sync protocol,
version: u16,
}
Expand All @@ -55,6 +50,7 @@ impl<'db> Default for MultiStateSyncInfo<'db> {
Self {
current_prefixes: BTreeMap::new(),
processed_prefixes: BTreeSet::new(),
app_hash: [0; 32],
version: CURRENT_STATE_SYNC_VERSION,
}
}
Expand Down Expand Up @@ -115,6 +111,7 @@ pub fn util_path_to_string(path: &[Vec<u8>]) -> Vec<String> {
// Splits the given global chunk id into [SUBTREE_PREFIX:CHUNK_ID]
pub fn util_split_global_chunk_id(
global_chunk_id: &[u8],
app_hash: &[u8],
) -> Result<(crate::SubtreePrefix, Vec<u8>), Error> {
let chunk_prefix_length: usize = 32;
if global_chunk_id.len() < chunk_prefix_length {
Expand All @@ -123,6 +120,12 @@ pub fn util_split_global_chunk_id(
));
}

if global_chunk_id == app_hash {
let array_of_zeros: [u8; 32] = [0; 32];
let root_chunk_prefix_key: crate::SubtreePrefix = array_of_zeros;
return Ok((root_chunk_prefix_key, vec![]));
}

Comment on lines +123 to +128
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In rust you shouldn't have special numbers, meaning that [0;32] shouldn't have a special signicance, instead you should use enums or Options

let (chunk_prefix, chunk_id) = global_chunk_id.split_at(chunk_prefix_length);
let mut array = [0u8; 32];
array.copy_from_slice(chunk_prefix);
Expand Down Expand Up @@ -253,22 +256,13 @@ impl GroveDb {
));
}

let chunk_prefix_length: usize = 32;
if global_chunk_id.len() < chunk_prefix_length {
return Err(Error::CorruptedData(
"expected global chunk id of at least 32 length".to_string(),
));
}

let (chunk_prefix, chunk_id) = global_chunk_id.split_at(chunk_prefix_length);

let mut array = [0u8; 32];
array.copy_from_slice(chunk_prefix);
let chunk_prefix_key: crate::SubtreePrefix = array;
let root_app_hash = self.root_hash(tx).value?;
let (chunk_prefix, chunk_id) =
replication::util_split_global_chunk_id(global_chunk_id, &root_app_hash)?;

let subtrees_metadata = self.get_subtrees_metadata(tx)?;

match subtrees_metadata.data.get(&chunk_prefix_key) {
match subtrees_metadata.data.get(&chunk_prefix) {
Some(path_data) => {
let subtree = &path_data.0;
let subtree_path: Vec<&[u8]> = subtree.iter().map(|vec| vec.as_slice()).collect();
Expand All @@ -287,7 +281,7 @@ impl GroveDb {
let chunk_producer_res = ChunkProducer::new(&merk);
match chunk_producer_res {
Ok(mut chunk_producer) => {
let chunk_res = chunk_producer.chunk(chunk_id);
let chunk_res = chunk_producer.chunk(&chunk_id);
match chunk_res {
Ok((chunk, _)) => match util_encode_vec_ops(chunk) {
Ok(op_bytes) => Ok(op_bytes),
Expand Down Expand Up @@ -317,7 +311,7 @@ impl GroveDb {
let chunk_producer_res = ChunkProducer::new(&merk);
match chunk_producer_res {
Ok(mut chunk_producer) => {
let chunk_res = chunk_producer.chunk(chunk_id);
let chunk_res = chunk_producer.chunk(&chunk_id);
match chunk_res {
Ok((chunk, _)) => match util_encode_vec_ops(chunk) {
Ok(op_bytes) => Ok(op_bytes),
Expand Down Expand Up @@ -346,15 +340,14 @@ impl GroveDb {
// state_sync_info: Consumed StateSyncInfo
// app_hash: Snapshot's AppHash
// tx: Transaction for the state sync
// Returns the first set of global chunk ids that can be fetched from sources (+
// the StateSyncInfo transferring ownership back to the caller)
// Returns the StateSyncInfo transferring ownership back to the caller)
pub fn start_snapshot_syncing<'db>(
&'db self,
mut state_sync_info: MultiStateSyncInfo<'db>,
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

perhaps could be a &mut without that ownership return trick

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this PR doesn't contain changes of #300

app_hash: CryptoHash,
tx: &'db Transaction,
version: u16,
) -> Result<(Vec<Vec<u8>>, MultiStateSyncInfo), Error> {
) -> Result<MultiStateSyncInfo, Error> {
// For now, only CURRENT_STATE_SYNC_VERSION is supported
if version != CURRENT_STATE_SYNC_VERSION {
return Err(Error::CorruptedData(
Expand All @@ -367,8 +360,6 @@ impl GroveDb {
));
}

let mut res = vec![];

if !state_sync_info.current_prefixes.is_empty()
|| !state_sync_info.processed_prefixes.is_empty()
{
Expand All @@ -391,26 +382,27 @@ impl GroveDb {
state_sync_info
.current_prefixes
.insert(root_prefix, root_prefix_state_sync_info);

res.push(root_prefix.to_vec());
state_sync_info.app_hash = app_hash;
} else {
return Err(Error::InternalError("Unable to open merk for replication"));
}

Ok((res, state_sync_info))
Ok(state_sync_info)
}

// Apply a chunk (should be called by ABCI when ApplySnapshotChunk method is
// called) Params:
// state_sync_info: Consumed MultiStateSyncInfo
// chunk: (Global chunk id, Chunk proof operators encoded in bytes)
// global_chunk_id: Global chunk id
// chunk: Chunk proof operators encoded in bytes
// tx: Transaction for the state sync
// Returns the next set of global chunk ids that can be fetched from sources (+
// the MultiStateSyncInfo transferring ownership back to the caller)
pub fn apply_chunk<'db>(
&'db self,
mut state_sync_info: MultiStateSyncInfo<'db>,
chunk: (&[u8], Vec<u8>),
global_chunk_id: &[u8],
chunk: Vec<u8>,
tx: &'db Transaction,
version: u16,
) -> Result<(Vec<Vec<u8>>, MultiStateSyncInfo), Error> {
Expand All @@ -428,15 +420,15 @@ impl GroveDb {

let mut next_chunk_ids = vec![];

let (global_chunk_id, chunk_data) = chunk;
let (chunk_prefix, chunk_id) = replication::util_split_global_chunk_id(global_chunk_id)?;
let (chunk_prefix, chunk_id) =
replication::util_split_global_chunk_id(global_chunk_id, &state_sync_info.app_hash)?;

if state_sync_info.current_prefixes.is_empty() {
return Err(Error::InternalError("GroveDB is not in syncing mode"));
}
if let Some(subtree_state_sync) = state_sync_info.current_prefixes.remove(&chunk_prefix) {
if let Ok((res, mut new_subtree_state_sync)) =
self.apply_inner_chunk(subtree_state_sync, &chunk_id, chunk_data)
self.apply_inner_chunk(subtree_state_sync, &chunk_id, chunk)
{
if !res.is_empty() {
for local_chunk_id in res.iter() {
Expand Down
7 changes: 4 additions & 3 deletions tutorials/src/bin/replication.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -245,15 +245,16 @@ fn sync_db_demo(
target_tx: &Transaction,
) -> Result<(), grovedb::Error> {
let app_hash = source_db.root_hash(None).value.unwrap();
let (chunk_ids, mut state_sync_info) = target_db.start_snapshot_syncing(state_sync_info, app_hash, target_tx, CURRENT_STATE_SYNC_VERSION)?;
let mut state_sync_info = target_db.start_snapshot_syncing(state_sync_info, app_hash, target_tx, CURRENT_STATE_SYNC_VERSION)?;

let mut chunk_queue : VecDeque<Vec<u8>> = VecDeque::new();

chunk_queue.extend(chunk_ids);
// The very first chunk to fetch is always identified by the root app_hash
chunk_queue.push_back(app_hash.to_vec());

while let Some(chunk_id) = chunk_queue.pop_front() {
let ops = source_db.fetch_chunk(chunk_id.as_slice(), None, CURRENT_STATE_SYNC_VERSION)?;
let (more_chunks, new_state_sync_info) = target_db.apply_chunk(state_sync_info, (chunk_id.as_slice(), ops), target_tx, CURRENT_STATE_SYNC_VERSION)?;
let (more_chunks, new_state_sync_info) = target_db.apply_chunk(state_sync_info, chunk_id.as_slice(), ops, target_tx, CURRENT_STATE_SYNC_VERSION)?;
state_sync_info = new_state_sync_info;
chunk_queue.extend(more_chunks);
}
Expand Down
Loading