Fast sync (WIP) (#155)

* Fast sync (work in progress)

* Cargo.lock

* Add missing hashes file

* clippy warnings

* Stub of database tool to create the fast sync hashes

* Command line arg for target height, error handling

* Cargo.lock

* fmt and unused imports

* fmt

* Add license information to consensus/fast-sync/Cargo.toml

Co-authored-by: Boog900 <boog900@tutanota.com>

* Order imports in consensus/fast-sync/src/create.rs

Co-authored-by: Boog900 <boog900@tutanota.com>

* beautify hex generation function & fmt

* Reorder imports consensus/fast-sync/src/fast_sync.rs

Cargo.lock

@@ -50,6 +50,12 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "anstyle"
+version = "1.0.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "038dfcf04a5feb68e9c60b21c9625a54c2c0616e79b72b0fd87075a056ae1d1b"
+
 [[package]]
 name = "async-buffer"
 version = "0.1.0"
@@ -71,6 +77,28 @@ dependencies = [
  "pin-project-lite",
 ]
 
+[[package]]
+name = "async-stream"
+version = "0.3.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd56dd203fef61ac097dd65721a419ddccb106b2d2b70ba60a6b529f03961a51"
+dependencies = [
+ "async-stream-impl",
+ "futures-core",
+ "pin-project-lite",
+]
+
+[[package]]
+name = "async-stream-impl"
+version = "0.3.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "16e62a023e7c117e27523144c5d2459f4397fcc3cab0085af8e2224f643a0193"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.60",
+]
+
 [[package]]
 name = "async-trait"
 version = "0.1.80"
@@ -287,6 +315,44 @@ dependencies = [
  "windows-targets 0.52.5",
 ]
 
+[[package]]
+name = "clap"
+version = "4.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a9689a29b593160de5bc4aacab7b5d54fb52231de70122626c178e6a368994c7"
+dependencies = [
+ "clap_builder",
+ "clap_derive",
+]
+
+[[package]]
+name = "clap_builder"
+version = "4.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2e5387378c84f6faa26890ebf9f0a92989f8873d4d380467bcd0d8d8620424df"
+dependencies = [
+ "anstyle",
+ "clap_lex",
+]
+
+[[package]]
+name = "clap_derive"
+version = "4.5.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c780290ccf4fb26629baa7a1081e68ced113f1d3ec302fa5948f1c381ebf06c6"
+dependencies = [
+ "heck 0.5.0",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.60",
+]
+
+[[package]]
+name = "clap_lex"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4b82cf0babdbd58558212896d1a4272303a57bdb245c2bf1147185fb45640e70"
+
 [[package]]
 name = "concurrent-queue"
 version = "2.5.0"
@@ -493,6 +559,22 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "cuprate-fast-sync"
+version = "0.1.0"
+dependencies = [
+ "clap",
+ "cuprate-blockchain",
+ "cuprate-types",
+ "hex",
+ "hex-literal",
+ "rayon",
+ "sha3",
+ "tokio",
+ "tokio-test",
+ "tower",
+]
+
 [[package]]
 name = "cuprate-helper"
 version = "0.1.0"
@@ -983,6 +1065,12 @@ version = "0.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "95505c38b4572b2d910cecb0281560f54b440a19336cbbcb27bf6ce6adc6f5a8"
 
+[[package]]
+name = "heck"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea"
+
 [[package]]
 name = "heed"
 version = "0.20.2"
@@ -2076,7 +2164,7 @@ version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f4a8caec23b7800fb97971a1c6ae365b6239aaeddfb934d6265f8505e795699d"
 dependencies = [
- "heck",
+ "heck 0.4.1",
  "proc-macro2",
  "quote",
  "syn 2.0.60",
@@ -2414,6 +2502,19 @@ dependencies = [
  "tokio-util",
 ]
 
+[[package]]
+name = "tokio-test"
+version = "0.4.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2468baabc3311435b55dd935f702f42cd1b8abb7e754fb7dfb16bd36aa88f9f7"
+dependencies = [
+ "async-stream",
+ "bytes",
+ "futures-core",
+ "tokio",
+ "tokio-stream",
+]
+
 [[package]]
 name = "tokio-util"
 version = "0.7.10"

Cargo.toml

@@ -3,6 +3,7 @@ resolver = "2"
 
 members = [
 	"consensus",
+	"consensus/fast-sync",
 	"consensus/rules",
 	"cryptonight",
 	"helper",

consensus/fast-sync/Cargo.toml

@@ -0,0 +1,23 @@
+[package]
+name = "cuprate-fast-sync"
+version = "0.1.0"
+edition = "2021"
+license = "MIT"
+
+[[bin]]
+name = "cuprate-fast-sync-create-hashes"
+path = "src/create.rs"
+
+[dependencies]
+clap = { workspace = true, features = ["derive", "std"] }
+cuprate-blockchain = { path = "../../storage/cuprate-blockchain" }
+cuprate-types = { path = "../../types" }
+hex.workspace = true
+hex-literal.workspace = true
+rayon.workspace = true
+sha3 = "0.10.8"
+tokio = { workspace = true, features = ["full"] }
+tower.workspace = true
+
+[dev-dependencies]
+tokio-test = "0.4.4"

consensus/fast-sync/src/create.rs

@@ -0,0 +1,87 @@
+use std::{fmt::Write, fs::write};
+
+use clap::Parser;
+use tower::{Service, ServiceExt};
+
+use cuprate_blockchain::{config::ConfigBuilder, service::DatabaseReadHandle, RuntimeError};
+use cuprate_types::blockchain::{BCReadRequest, BCResponse};
+
+use cuprate_fast_sync::{hash_of_hashes, BlockId, HashOfHashes};
+
+const BATCH_SIZE: u64 = 512;
+
+async fn read_batch(
+    handle: &mut DatabaseReadHandle,
+    height_from: u64,
+) -> Result<Vec<BlockId>, RuntimeError> {
+    let mut block_ids = Vec::<BlockId>::with_capacity(BATCH_SIZE as usize);
+
+    for height in height_from..(height_from + BATCH_SIZE) {
+        let request = BCReadRequest::BlockHash(height);
+        let response_channel = handle.ready().await?.call(request);
+        let response = response_channel.await?;
+
+        match response {
+            BCResponse::BlockHash(block_id) => block_ids.push(block_id),
+            _ => unreachable!(),
+        }
+    }
+
+    Ok(block_ids)
+}
+
+fn generate_hex(hashes: &[HashOfHashes]) -> String {
+    let mut s = String::new();
+
+    writeln!(&mut s, "[").unwrap();
+
+    for hash in hashes {
+        writeln!(&mut s, "\thex!(\"{}\"),", hex::encode(hash)).unwrap();
+    }
+
+    writeln!(&mut s, "]").unwrap();
+
+    s
+}
+
+#[derive(Parser)]
+#[command(version, about, long_about = None)]
+struct Args {
+    #[arg(short, long)]
+    height: u64,
+}
+
+#[tokio::main]
+async fn main() {
+    let args = Args::parse();
+    let height_target = args.height;
+
+    let config = ConfigBuilder::new().build();
+
+    let (mut read_handle, _) = cuprate_blockchain::service::init(config).unwrap();
+
+    let mut hashes_of_hashes = Vec::new();
+
+    let mut height = 0u64;
+
+    while height < height_target {
+        match read_batch(&mut read_handle, height).await {
+            Ok(block_ids) => {
+                let hash = hash_of_hashes(block_ids.as_slice());
+                hashes_of_hashes.push(hash);
+            }
+            Err(_) => {
+                println!("Failed to read next batch from database");
+                break;
+            }
+        }
+        height += BATCH_SIZE;
+    }
+
+    drop(read_handle);
+
+    let generated = generate_hex(&hashes_of_hashes);
+    write("src/data/hashes_of_hashes", generated).expect("Could not write file");
+
+    println!("Generated hashes up to block height {}", height);
+}

consensus/fast-sync/src/data/hashes_of_hashes

@@ -0,0 +1,12 @@
+[
+	hex!("1adffbaf832784406018009e07d3dc3a39da7edb6632523c119ed8acb32eb934"),
+	hex!("ae960265e3398d04f3cd4f949ed13c2689424887c71c1441a03d900a9d3a777f"),
+	hex!("938c72d267bbd3a17cdecbe02443d00012ee62d6e9f3524f5a914192110b1798"),
+	hex!("de0c82e51549b6514b42a591fd5440dddb5cc0118ec461459a99017bf06a0a0a"),
+	hex!("9a50f4586ec7e0fb58c6383048d3b334180235fd34bb714af20f1a3ebce4c911"),
+	hex!("5a3942f9bb318d65997bf57c40e045d62e7edbe35f3dae57499c2c5554896543"),
+	hex!("9dccee3b094cdd1b98e357c2c81bfcea798ea75efd94e67c6f5e86f428c5ec2c"),
+	hex!("620397540d44f21c3c57c20e9d47c6aaf0b1bf4302a4d43e75f2e33edd1a4032"),
+	hex!("ef6c612fb17bd70ac2ac69b2f85a421b138cc3a81daf622b077cb402dbf68377"),
+	hex!("6815ecb2bd73a3ba5f20558bfe1b714c30d6892b290e0d6f6cbf18237cedf75a"),
+]

consensus/fast-sync/src/fast_sync.rs

@@ -0,0 +1,216 @@
+use std::{
+    cmp,
+    future::Future,
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+#[allow(unused_imports)]
+use hex_literal::hex;
+use tower::Service;
+
+use crate::{hash_of_hashes, BlockId, HashOfHashes};
+#[cfg(not(test))]
+static HASHES_OF_HASHES: &[HashOfHashes] = &include!("./data/hashes_of_hashes");
+
+#[cfg(not(test))]
+const BATCH_SIZE: usize = 512;
+
+#[cfg(test)]
+static HASHES_OF_HASHES: &[HashOfHashes] = &[
+    hex!("3fdc9032c16d440f6c96be209c36d3d0e1aed61a2531490fe0ca475eb615c40a"),
+    hex!("0102030405060708010203040506070801020304050607080102030405060708"),
+    hex!("0102030405060708010203040506070801020304050607080102030405060708"),
+];
+
+#[cfg(test)]
+const BATCH_SIZE: usize = 4;
+
+#[inline]
+fn max_height() -> u64 {
+    (HASHES_OF_HASHES.len() * BATCH_SIZE) as u64
+}
+
+pub enum FastSyncRequest {
+    ValidateHashes {
+        start_height: u64,
+        block_ids: Vec<BlockId>,
+    },
+}
+
+#[derive(Debug, PartialEq)]
+pub struct ValidBlockId(BlockId);
+
+fn valid_block_ids(block_ids: &[BlockId]) -> Vec<ValidBlockId> {
+    block_ids.iter().map(|b| ValidBlockId(*b)).collect()
+}
+
+#[derive(Debug, PartialEq)]
+pub enum FastSyncResponse {
+    ValidateHashes {
+        validated_hashes: Vec<ValidBlockId>,
+        unknown_hashes: Vec<BlockId>,
+    },
+}
+
+#[derive(Debug, PartialEq)]
+pub enum FastSyncError {
+    InvalidStartHeight, // start_height not a multiple of BATCH_SIZE
+    Mismatch,           // hash does not match
+    NothingToDo,        // no complete batch to check
+    OutOfRange,         // start_height too high
+}
+
+#[allow(dead_code)]
+pub struct FastSyncService<C> {
+    context_svc: C,
+}
+
+impl<C> FastSyncService<C>
+where
+    C: Service<FastSyncRequest, Response = FastSyncResponse, Error = FastSyncError>
+        + Clone
+        + Send
+        + 'static,
+{
+    #[allow(dead_code)]
+    pub(crate) fn new(context_svc: C) -> FastSyncService<C> {
+        FastSyncService { context_svc }
+    }
+}
+
+impl<C> Service<FastSyncRequest> for FastSyncService<C>
+where
+    C: Service<FastSyncRequest, Response = FastSyncResponse, Error = FastSyncError>
+        + Clone
+        + Send
+        + 'static,
+    C::Future: Send + 'static,
+{
+    type Response = FastSyncResponse;
+    type Error = FastSyncError;
+    type Future =
+        Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send + 'static>>;
+
+    fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        Poll::Ready(Ok(()))
+    }
+
+    fn call(&mut self, req: FastSyncRequest) -> Self::Future {
+        Box::pin(async move {
+            match req {
+                FastSyncRequest::ValidateHashes {
+                    start_height,
+                    block_ids,
+                } => validate_hashes(start_height, &block_ids).await,
+            }
+        })
+    }
+}
+
+async fn validate_hashes(
+    start_height: u64,
+    block_ids: &[BlockId],
+) -> Result<FastSyncResponse, FastSyncError> {
+    if start_height as usize % BATCH_SIZE != 0 {
+        return Err(FastSyncError::InvalidStartHeight);
+    }
+
+    if start_height >= max_height() {
+        return Err(FastSyncError::OutOfRange);
+    }
+
+    let stop_height = start_height as usize + block_ids.len();
+
+    let batch_from = start_height as usize / BATCH_SIZE;
+    let batch_to = cmp::min(stop_height / BATCH_SIZE, HASHES_OF_HASHES.len());
+    let n_batches = batch_to - batch_from;
+
+    if n_batches == 0 {
+        return Err(FastSyncError::NothingToDo);
+    }
+
+    for i in 0..n_batches {
+        let batch = &block_ids[BATCH_SIZE * i..BATCH_SIZE * (i + 1)];
+        let actual = hash_of_hashes(batch);
+        let expected = HASHES_OF_HASHES[batch_from + i];
+
+        if expected != actual {
+            return Err(FastSyncError::Mismatch);
+        }
+    }
+
+    let validated_hashes = valid_block_ids(&block_ids[..n_batches * BATCH_SIZE]);
+    let unknown_hashes = block_ids[n_batches * BATCH_SIZE..].to_vec();
+
+    Ok(FastSyncResponse::ValidateHashes {
+        validated_hashes,
+        unknown_hashes,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio_test::block_on;
+
+    #[test]
+    fn test_validate_hashes_errors() {
+        let ids = [[1u8; 32], [2u8; 32], [3u8; 32], [4u8; 32], [5u8; 32]];
+        assert_eq!(
+            block_on(validate_hashes(3, &[])),
+            Err(FastSyncError::InvalidStartHeight)
+        );
+        assert_eq!(
+            block_on(validate_hashes(3, &ids)),
+            Err(FastSyncError::InvalidStartHeight)
+        );
+
+        assert_eq!(
+            block_on(validate_hashes(20, &[])),
+            Err(FastSyncError::OutOfRange)
+        );
+        assert_eq!(
+            block_on(validate_hashes(20, &ids)),
+            Err(FastSyncError::OutOfRange)
+        );
+
+        assert_eq!(
+            block_on(validate_hashes(4, &[])),
+            Err(FastSyncError::NothingToDo)
+        );
+        assert_eq!(
+            block_on(validate_hashes(4, &ids[..3])),
+            Err(FastSyncError::NothingToDo)
+        );
+    }
+
+    #[test]
+    fn test_validate_hashes_success() {
+        let ids = [[1u8; 32], [2u8; 32], [3u8; 32], [4u8; 32], [5u8; 32]];
+        let validated_hashes = valid_block_ids(&ids[0..4]);
+        let unknown_hashes = ids[4..].to_vec();
+        assert_eq!(
+            block_on(validate_hashes(0, &ids)),
+            Ok(FastSyncResponse::ValidateHashes {
+                validated_hashes,
+                unknown_hashes
+            })
+        );
+    }
+
+    #[test]
+    fn test_validate_hashes_mismatch() {
+        let ids = [
+            [1u8; 32], [2u8; 32], [3u8; 32], [5u8; 32], [1u8; 32], [2u8; 32], [3u8; 32], [4u8; 32],
+        ];
+        assert_eq!(
+            block_on(validate_hashes(0, &ids)),
+            Err(FastSyncError::Mismatch)
+        );
+        assert_eq!(
+            block_on(validate_hashes(4, &ids)),
+            Err(FastSyncError::Mismatch)
+        );
+    }
+}

consensus/fast-sync/src/lib.rs

@@ -0,0 +1,4 @@
+pub mod fast_sync;
+pub mod util;
+
+pub use util::{hash_of_hashes, BlockId, HashOfHashes};

consensus/fast-sync/src/util.rs

@@ -0,0 +1,8 @@
+use sha3::{Digest, Keccak256};
+
+pub type BlockId = [u8; 32];
+pub type HashOfHashes = [u8; 32];
+
+pub fn hash_of_hashes(hashes: &[BlockId]) -> HashOfHashes {
+    Keccak256::digest(hashes.concat().as_slice()).into()
+}