use libp2p::{
request_response::{self, ProtocolSupport},
StreamProtocol,
};
use serde::{Deserialize, Serialize};
use std::time::Duration;
use tenzro_types::block::Block;
use tenzro_types::primitives::{BlockHeight, Hash};
pub const BLOCK_SYNC_PROTOCOL: &str = "/tenzro/block-sync/1.0.0";
pub const MAX_BLOCKS_PER_RANGE: u32 = 128;
pub const MAX_BLOCK_HASHES_PER_REQUEST: usize = 128;
pub const MAX_INFLIGHT_REQUESTS_PER_PEER: usize = 10;
pub const MAX_INBOUND_STREAMS_PER_PEER: usize = 5;
pub const REQUEST_TIMEOUT_HEADERS: Duration = Duration::from_secs(10);
pub const REQUEST_TIMEOUT_BODIES: Duration = Duration::from_secs(30);
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum BlockSyncRequest {
GetTipInfo,
GetBlockRange { start: BlockHeight, count: u32 },
GetBlockByHash { hash: Hash },
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum BlockSyncResponse {
TipInfo {
tip_height: BlockHeight,
tip_hash: Hash,
lowest_available: BlockHeight,
},
BlockRange { blocks: Vec<Block> },
BlockByHash { block: Option<Block> },
Error(BlockSyncError),
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, thiserror::Error)]
pub enum BlockSyncError {
#[error("requested range too large (max {max}, got {got})")]
RangeTooLarge { got: u32, max: u32 },
#[error("server busy — exceeded {limit} concurrent inbound streams from this peer")]
ServerBusy { limit: usize },
#[error("requested block below pruned tail (start={start}, lowest_available={lowest_available})")]
BelowPrunedTail {
start: BlockHeight,
lowest_available: BlockHeight,
},
#[error("server-side storage error: {0}")]
Storage(String),
}
pub type BlockSyncBehaviour =
request_response::cbor::Behaviour<BlockSyncRequest, BlockSyncResponse>;
pub fn new_behaviour() -> BlockSyncBehaviour {
let protocol = StreamProtocol::new(BLOCK_SYNC_PROTOCOL);
let cfg = request_response::Config::default()
.with_request_timeout(REQUEST_TIMEOUT_BODIES);
request_response::cbor::Behaviour::new(
std::iter::once((protocol, ProtocolSupport::Full)),
cfg,
)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn request_round_trip() {
let cases = vec![
BlockSyncRequest::GetTipInfo,
BlockSyncRequest::GetBlockRange {
start: BlockHeight::from(42u64),
count: 16,
},
BlockSyncRequest::GetBlockByHash {
hash: Hash::from_bytes(&[7u8; 32]).unwrap(),
},
];
for req in cases {
let bytes = bincode::serialize(&req).expect("encode");
let decoded: BlockSyncRequest =
bincode::deserialize(&bytes).expect("decode");
assert_eq!(req, decoded);
}
}
#[test]
fn response_round_trip() {
let cases = vec![
BlockSyncResponse::TipInfo {
tip_height: BlockHeight::from(1000u64),
tip_hash: Hash::from_bytes(&[1u8; 32]).unwrap(),
lowest_available: BlockHeight::from(0u64),
},
BlockSyncResponse::BlockRange { blocks: vec![] },
BlockSyncResponse::BlockByHash { block: None },
BlockSyncResponse::Error(BlockSyncError::RangeTooLarge {
got: 999,
max: MAX_BLOCKS_PER_RANGE,
}),
BlockSyncResponse::Error(BlockSyncError::ServerBusy {
limit: MAX_INBOUND_STREAMS_PER_PEER,
}),
];
for resp in cases {
let bytes = bincode::serialize(&resp).expect("encode");
let decoded: BlockSyncResponse =
bincode::deserialize(&bytes).expect("decode");
assert_eq!(resp, decoded);
}
}
#[test]
fn behaviour_constructs() {
let _ = new_behaviour();
}
}