logos-blockchain/logos-blockchain
1
0
Fork 0
mirror of https://github.com/logos-blockchain/logos-blockchain.git synced 2026-08-01 12:03:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: bounded blocks range stream anomalies (#2757)

Browse Source
This commit is contained in:
Hansie Odendaal 2026-05-21 19:20:56 +02:00 committed by GitHub
parent 1ecc29c0e6
commit 8904f2e825
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 681 additions and 630 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
Cargo.lock generated
View File

@ -4592,9 +4592,14 @@ dependencies = [
"pprof",
"serde",
"serde_json",
"serde_urlencoded",
"serde_with",
"time",
"tokio",
"tracing",
"url",
"utoipa",
"validator",
]
[[package]]
@ -4826,6 +4831,7 @@ dependencies = [
"serde",
"serde_ignored",
"serde_json",
"serde_urlencoded",
"serde_with",
"serde_yaml",
"thiserror 2.0.18",
@ -5287,6 +5293,7 @@ dependencies = [
"logos-blockchain-common-http-client",
"logos-blockchain-core",
"logos-blockchain-groth16",
"logos-blockchain-http-api-common",
"logos-blockchain-key-management-system-service",
"num-bigint",
"rand 0.8.6",

1
Cargo.toml
View File

@ -261,6 +261,7 @@ serde_arrays = { default-features = false, version = "0.2.
serde_ignored = { default-features = false, version = "0.1" }
serde_json = { default-features = false, version = "1.0" }
serde_path_to_error = { default-features = false, version = "0.1" }
serde_urlencoded = { default-features = false, version = "0.7.1" }
serde_with = { default-features = false, version = "3.14.0" }
serde_yaml = { default-features = false, version = "0.9.33" }
serde_yml = { default-features = false, version = "0.0.12" }

5
nodes/api-common/Cargo.toml
View File

@ -19,9 +19,14 @@ lb-key-management-system-keys = { workspace = true }
lb-tracing = { workspace = true }
serde = { features = ["alloc", "derive"], workspace = true }
serde_json = { workspace = true }
serde_urlencoded = { workspace = true }
serde_with = { features = ["macros"], workspace = true }
time = { workspace = true }
tokio = { features = ["time"], optional = true, workspace = true }
tracing = { workspace = true }
url = { workspace = true }
utoipa = { workspace = true }
validator = { features = ["derive"], workspace = true }
[target.'cfg(not(windows))'.dependencies]
pprof = { features = ["criterion", "flamegraph", "protobuf-codec"], optional = true, workspace = true }

1
nodes/api-common/src/lib.rs
View File

@ -3,6 +3,7 @@ pub mod metrics;
pub mod paths;
#[cfg(feature = "profiling")]
pub mod pprof;
pub mod queries;
pub mod settings;
#[cfg(all(feature = "profiling", target_os = "windows"))]

206
nodes/api-common/src/queries.rs Normal file
View File

@ -0,0 +1,206 @@
use std::num::NonZero;
use serde::{Deserialize, Serialize};
use url::Url;
use utoipa::IntoParams;
use validator::Validate;
use crate::{MAX_BLOCKS_STREAM_BLOCKS, MAX_BLOCKS_STREAM_CHUNK_SIZE};
/// Query parameters for the blocks stream endpoint, with validation and
/// `OpenAPI` schema generation. Note: Literals in `param` are duplicated due to
/// utoipa attribute limitations.
#[derive(Debug, Copy, Clone, PartialEq, Eq, IntoParams, Deserialize, Serialize, Validate)]
#[into_params(parameter_in = Query)]
pub struct BlocksStreamQuery {
/// If omitted, the server chooses a default lower bound.
/// For descending streams this is `slot 0` (bounded by `blocks_limit`).
/// For ascending streams, `slot_from` is estimated from the average
/// slots-per-block and `blocks_limit`, biased so the stream ends near
/// `slot_to`. This may return fewer than `blocks_limit` blocks; callers
/// can refine by specifying `slot_from` explicitly.
#[serde(default, skip_serializing_if = "Option::is_none")]
#[param(minimum = 0)]
pub slot_from: Option<u64>,
/// Upper bound slot (inclusive). Defaults to tip slot, or LIB slot when
/// `immutable_only=true`.
#[serde(default, skip_serializing_if = "Option::is_none")]
#[param(minimum = 0)]
pub slot_to: Option<u64>,
/// Sort direction. Defaults to descending.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub order: Option<BlockSortOrder>,
/// The maximum number of actual blocks to return. If omitted:
/// - explicit bounded slot range (`slot_from` and `slot_to`) defaults to
/// the server maximum (`630_720_000`);
/// - otherwise defaults to `100`.
#[serde(default, skip_serializing_if = "Option::is_none")]
#[validate(custom(function = "validate_blocks_limit"))]
#[param(minimum = 1, maximum = 630_720_000, default = 100, example = 100)]
pub blocks_limit: Option<NonZero<usize>>,
/// Server chunk size hint for streamed delivery. Defaults to `100` ,
/// maximum `1000`.
#[serde(default, skip_serializing_if = "Option::is_none")]
#[validate(custom(function = "validate_server_batch_size"))]
#[param(minimum = 1, maximum = 1_000, default = 100, example = 100)]
pub server_batch_size: Option<NonZero<usize>>,
/// When true, include only immutable blocks.
/// If `slot_to` is omitted, the default anchor is LIB slot.
/// Defaults to `false`.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub block_filter: Option<BlockFilter>,
}
#[expect(
clippy::trivially_copy_pass_by_ref,
reason = "validator derive calls custom validators by reference"
)]
fn validate_blocks_limit(v: &NonZero<usize>) -> Result<(), validator::ValidationError> {
if v.get() > MAX_BLOCKS_STREAM_BLOCKS {
let mut err = validator::ValidationError::new("out_of_range");
err.message =
Some(format!("'blocks_limit' must be in [1, {MAX_BLOCKS_STREAM_BLOCKS}]").into());
err.add_param("field".into(), &"blocks_limit");
err.add_param("min".into(), &1);
err.add_param("max".into(), &MAX_BLOCKS_STREAM_BLOCKS);
err.add_param("value".into(), &v.get());
return Err(err);
}
Ok(())
}
#[expect(
clippy::trivially_copy_pass_by_ref,
reason = "validator derive calls custom validators by reference"
)]
fn validate_server_batch_size(v: &NonZero<usize>) -> Result<(), validator::ValidationError> {
if v.get() > MAX_BLOCKS_STREAM_CHUNK_SIZE {
let mut err = validator::ValidationError::new("out_of_range");
err.message = Some(
format!("'server_batch_size' must be in [1, {MAX_BLOCKS_STREAM_CHUNK_SIZE}]").into(),
);
err.add_param("field".into(), &"server_batch_size");
err.add_param("min".into(), &1);
err.add_param("max".into(), &MAX_BLOCKS_STREAM_CHUNK_SIZE);
err.add_param("value".into(), &v.get());
return Err(err);
}
Ok(())
}
impl BlocksStreamQuery {
const fn params_is_none(&self) -> bool {
self.slot_from.is_none()
&& self.slot_to.is_none()
&& self.order.is_none()
&& self.blocks_limit.is_none()
&& self.server_batch_size.is_none()
&& self.block_filter.is_none()
}
/// Append query parameters to the given URL.
///
/// Fields that are `None` are omitted so server defaults apply.
pub fn append_to_url(&self, request_url: &mut Url) {
if self.params_is_none() {
return;
}
if let Ok(encoded) = serde_urlencoded::to_string(self) {
request_url.query_pairs_mut().extend_pairs(
url::form_urlencoded::parse(encoded.as_bytes())
.map(|(k, v)| (k.into_owned(), v.into_owned())),
);
}
}
}
/// Sort order for blocks in the `get_blocks_range_stream` method.
#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BlockSortOrder {
/// Ascending order (oldest to newest).
Ascending,
/// Descending order (newest to oldest).
Descending,
}
/// Filter for block types in the `get_blocks_range_stream` method.
#[derive(Debug, Copy, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum BlockFilter {
/// Includes only immutable blocks.
ImmutableOnly,
/// Includes mutable and immutable blocks.
MutableAndImmutable,
}
#[cfg(test)]
mod tests {
use std::num::NonZero;
use url::Url;
use crate::queries::{BlockFilter, BlockSortOrder, BlocksStreamQuery};
#[test]
fn blocks_stream_query_encodes_to_server_query_shape() {
let mut request_url = Url::parse("http://localhost:8080/cryptarchia/blocks_range").unwrap();
let params = BlocksStreamQuery {
slot_from: Some(10),
slot_to: Some(20),
order: Some(BlockSortOrder::Ascending),
blocks_limit: NonZero::new(50),
server_batch_size: NonZero::new(5),
block_filter: Some(BlockFilter::ImmutableOnly),
};
params.append_to_url(&mut request_url);
assert_eq!(
request_url.query(),
Some(
"slot_from=10&slot_to=20&order=ascending&blocks_limit=50&server_batch_size=5&block_filter=immutable_only"
)
);
let params = BlocksStreamQuery {
slot_from: Some(10),
slot_to: Some(20),
order: Some(BlockSortOrder::Descending),
blocks_limit: NonZero::new(50),
server_batch_size: NonZero::new(5),
block_filter: Some(BlockFilter::MutableAndImmutable),
};
let mut request_url = Url::parse("http://localhost:8080/cryptarchia/blocks_range").unwrap();
params.append_to_url(&mut request_url);
assert_eq!(
request_url.query(),
Some(
"slot_from=10&slot_to=20&order=descending&blocks_limit=50&server_batch_size=5&block_filter=mutable_and_immutable"
)
);
}
#[test]
fn blocks_stream_query_omits_unspecified_values() {
let mut request_url = Url::parse("http://localhost:8080/cryptarchia/blocks_range").unwrap();
let params = BlocksStreamQuery {
slot_from: None,
slot_to: None,
order: None,
blocks_limit: None,
server_batch_size: None,
block_filter: None,
};
params.append_to_url(&mut request_url);
assert_eq!(
request_url.as_str(),
"http://localhost:8080/cryptarchia/blocks_range"
);
assert!(request_url.query().is_none());
}
}

3
nodes/node/binary/Cargo.toml
View File

@ -80,7 +80,8 @@ validator = { features = ["derive"], workspace = true }
tikv-jemallocator = { optional = true, workspace = true }
[dev-dependencies]
rand = { workspace = true }
rand = { workspace = true }
serde_urlencoded = { workspace = true }
[features]
default = ["tracing"]

46
nodes/node/binary/src/api/handlers.rs
View File

@ -38,6 +38,7 @@ use lb_http_api_common::{
},
},
paths,
queries::BlocksStreamQuery,
};
use lb_libp2p::libp2p::bytes::Bytes;
use lb_network_service::backends::libp2p::Libp2p as Libp2pNetworkBackend;
@ -58,11 +59,12 @@ use overwatch::{
};
use serde::{Deserialize, Serialize};
use tokio_stream::StreamExt as _;
use tracing::debug;
use crate::api::{
errors::{BlocksStreamHandlerError, BlocksStreamWindowError},
openapi::schema,
queries::{BlockRangeQuery, BlocksStreamQuery, BlocksStreamRequest},
queries::{BlockRangeQuery, BlocksStreamRequest},
responses::{self, overwatch::get_relay_or_500},
serializers::{
blocks::{ApiBlock, ApiProcessedBlockEvent},
@ -70,6 +72,8 @@ use crate::api::{
},
};
const TARGET: &str = "node::binary::api";
#[derive(Debug)]
struct ResolvedBlocksStreamWindow {
slot_from: Slot,
@ -104,18 +108,24 @@ fn resolve_blocks_stream_window(
} else {
chain_info.slot
};
let slot_to = request.slot_to.map_or(max_slot_to, Slot::new);
let mut slot_to = request.slot_to.map_or(max_slot_to, Slot::new);
if slot_to > max_slot_to {
slot_to = max_slot_to;
let anchor = if request.immutable_only {
"lib_slot"
} else {
"tip_slot"
};
return Err(BlocksStreamWindowError::SlotToAboveAnchor {
anchor,
slot_to: slot_to.into_inner(),
max_slot_to: max_slot_to.into_inner(),
});
debug!(
target: TARGET,
"{}: clamping to {}",
BlocksStreamWindowError::SlotToAboveAnchor {
anchor,
slot_to: slot_to.into_inner(),
max_slot_to: max_slot_to.into_inner(),
}.to_string(),
max_slot_to.into_inner()
);
}
let slot_from = request.slot_from.map_or_else(
@ -1554,31 +1564,25 @@ mod tests {
}
#[test]
fn rejects_slot_to_above_tip() {
let err = resolve_blocks_stream_window(
fn clamps_slot_to_above_tip() {
let window = resolve_blocks_stream_window(
&request(None, Some(TIP_SLOT + 1), true, DEFAULT_LIMIT, false),
&chain_info(),
)
.unwrap_err();
.unwrap();
assert!(matches!(
err,
BlocksStreamWindowError::SlotToAboveAnchor { .. }
));
assert_eq!(window.slot_to, Slot::new(TIP_SLOT));
}
#[test]
fn rejects_slot_to_above_lib_when_immutable_only() {
let err = resolve_blocks_stream_window(
fn clamps_slot_to_above_lib_when_immutable_only() {
let window = resolve_blocks_stream_window(
&request(None, Some(LIB_SLOT + 1), true, DEFAULT_LIMIT, true),
&chain_info(),
)
.unwrap_err();
.unwrap();
assert!(matches!(
err,
BlocksStreamWindowError::SlotToAboveAnchor { .. }
));
assert_eq!(window.slot_to, Slot::new(LIB_SLOT));
}
#[test]

240
nodes/node/binary/src/api/queries.rs
View File

@ -2,11 +2,11 @@ use std::num::{NonZero, NonZeroUsize};
use lb_http_api_common::{
DEFAULT_BLOCKS_STREAM_CHUNK_SIZE, DEFAULT_NUMBER_OF_BLOCKS_TO_STREAM, MAX_BLOCKS_STREAM_BLOCKS,
MAX_BLOCKS_STREAM_CHUNK_SIZE,
queries::{BlockFilter, BlockSortOrder, BlocksStreamQuery},
};
use serde::Deserialize;
use utoipa::IntoParams;
use validator::Validate;
use validator::Validate as _;
use crate::api::errors::BlocksStreamRequestError;
@ -20,117 +20,6 @@ pub struct BlockRangeQuery {
pub slot_to: usize,
}
/// Query parameters for the blocks stream endpoint, with validation and
/// `OpenAPI` schema generation. Note: Literals in `param` are duplicated due to
/// utoipa attribute limitations.
#[derive(IntoParams, Deserialize, Validate)]
#[into_params(parameter_in = Query)]
pub struct BlocksStreamQuery {
/// If omitted, the server chooses a default lower bound.
/// For descending streams this is `slot 0` (bounded by `blocks_limit`).
/// For ascending streams, `slot_from` is estimated from the average
/// slots-per-block and `blocks_limit`, biased so the stream ends near
/// `slot_to`. This may return fewer than `blocks_limit` blocks; callers
/// can refine by specifying `slot_from` explicitly.
#[serde(default)]
#[param(minimum = 0)]
pub slot_from: Option<u64>,
/// Upper bound slot (inclusive). Defaults to tip slot, or LIB slot when
/// `immutable_only=true`.
#[serde(default)]
#[param(minimum = 0)]
pub slot_to: Option<u64>,
/// Sort direction. Defaults to descending (`true`).
#[serde(default)]
pub descending: Option<bool>,
/// The maximum number of actual blocks to return. If omitted:
/// - explicit bounded slot range (`slot_from` and `slot_to`) defaults to
/// the server maximum (`630_720_000`);
/// - otherwise defaults to `100`.
#[serde(default)]
#[validate(custom(function = "validate_blocks_limit"))]
#[param(minimum = 1, maximum = 630_720_000, default = 100, example = 100)]
pub blocks_limit: Option<usize>,
/// Server chunk size hint for streamed delivery. Defaults to `100` ,
/// maximum `1000`.
#[serde(default)]
#[validate(custom(function = "validate_server_batch_size"))]
#[param(minimum = 1, maximum = 1_000, default = 100, example = 100)]
pub server_batch_size: Option<usize>,
/// When true, include only immutable blocks.
/// If `slot_to` is omitted, the default anchor is LIB slot.
#[serde(default)]
pub immutable_only: Option<bool>,
}
fn validate_blocks_limit(v: usize) -> Result<(), validator::ValidationError> {
if v == 0 || v > MAX_BLOCKS_STREAM_BLOCKS {
let mut err = validator::ValidationError::new("out_of_range");
err.message =
Some(format!("'blocks_limit' must be in [1, {MAX_BLOCKS_STREAM_BLOCKS}]").into());
err.add_param("field".into(), &"blocks_limit");
err.add_param("min".into(), &1);
err.add_param("max".into(), &MAX_BLOCKS_STREAM_BLOCKS);
err.add_param("value".into(), &v);
return Err(err);
}
Ok(())
}
fn validate_server_batch_size(v: usize) -> Result<(), validator::ValidationError> {
if v == 0 || v > MAX_BLOCKS_STREAM_CHUNK_SIZE {
let mut err = validator::ValidationError::new("out_of_range");
err.message = Some(
format!("'server_batch_size' must be in [1, {MAX_BLOCKS_STREAM_CHUNK_SIZE}]").into(),
);
err.add_param("field".into(), &"server_batch_size");
err.add_param("min".into(), &1);
err.add_param("max".into(), &MAX_BLOCKS_STREAM_CHUNK_SIZE);
err.add_param("value".into(), &v);
return Err(err);
}
Ok(())
}
impl TryFrom<BlocksStreamQuery> for BlocksStreamRequest {
type Error = BlocksStreamRequestError;
// Parse and validate the query parameters for the blocks stream endpoint,
// applying defaults where necessary.
fn try_from(query: BlocksStreamQuery) -> Result<Self, Self::Error> {
query.validate()?;
let blocks_limit = query.blocks_limit.unwrap_or_else(|| {
if query.slot_from.is_some() && query.slot_to.is_some() {
MAX_BLOCKS_STREAM_BLOCKS
} else {
DEFAULT_NUMBER_OF_BLOCKS_TO_STREAM
}
});
let server_batch_size = query
.server_batch_size
.unwrap_or(DEFAULT_BLOCKS_STREAM_CHUNK_SIZE);
if let (Some(slot_from), Some(slot_to)) = (query.slot_from, query.slot_to)
&& slot_from > slot_to
{
return Err(BlocksStreamRequestError::InvalidSlotRange { slot_from, slot_to });
}
Ok(Self {
slot_from: query.slot_from,
slot_to: query.slot_to,
descending: query.descending.unwrap_or(true),
blocks_limit: NonZeroUsize::new(blocks_limit)
.expect("'blocks_limit' is always >= 1 in schema"),
server_batch_size: NonZeroUsize::new(server_batch_size)
.expect("'server_batch_size' is always >= 1 in schema"),
immutable_only: query.immutable_only.unwrap_or_default(),
})
}
}
/// This is a processed `BlocksStreamQuery` with all defaults applied and
/// validated, ready to be used for fetching blocks.
pub struct BlocksStreamRequest {
@ -148,19 +37,68 @@ pub struct BlocksStreamRequest {
pub immutable_only: bool,
}
impl TryFrom<BlocksStreamQuery> for BlocksStreamRequest {
type Error = BlocksStreamRequestError;
// Parse and validate the query parameters for the blocks stream endpoint,
// applying defaults where necessary.
fn try_from(query: BlocksStreamQuery) -> Result<Self, Self::Error> {
query.validate()?;
let blocks_limit = query.blocks_limit.unwrap_or_else(|| {
if query.slot_from.is_some() && query.slot_to.is_some() {
NonZeroUsize::new(MAX_BLOCKS_STREAM_BLOCKS).unwrap()
} else {
NonZeroUsize::new(DEFAULT_NUMBER_OF_BLOCKS_TO_STREAM).unwrap()
}
});
let server_batch_size = query
.server_batch_size
.unwrap_or(NonZeroUsize::new(DEFAULT_BLOCKS_STREAM_CHUNK_SIZE).unwrap());
if let (Some(slot_from), Some(slot_to)) = (query.slot_from, query.slot_to)
&& slot_from > slot_to
{
return Err(BlocksStreamRequestError::InvalidSlotRange { slot_from, slot_to });
}
Ok(Self {
slot_from: query.slot_from,
slot_to: query.slot_to,
descending: query.order.unwrap_or(BlockSortOrder::Descending)
== BlockSortOrder::Descending,
blocks_limit,
server_batch_size,
immutable_only: query
.block_filter
.unwrap_or(BlockFilter::MutableAndImmutable)
== BlockFilter::ImmutableOnly,
})
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::num::{NonZero, NonZeroUsize};
use lb_http_api_common::{
DEFAULT_BLOCKS_STREAM_CHUNK_SIZE, DEFAULT_NUMBER_OF_BLOCKS_TO_STREAM,
MAX_BLOCKS_STREAM_BLOCKS,
queries::{BlockFilter, BlockSortOrder, BlocksStreamQuery},
};
use crate::api::{errors::BlocksStreamRequestError, queries::BlocksStreamRequest};
#[test]
fn blocks_stream_request_defaults_to_recent_blocks_from_tip() {
let request = BlocksStreamRequest::try_from(BlocksStreamQuery {
slot_from: None,
slot_to: None,
descending: None,
order: None,
blocks_limit: None,
server_batch_size: None,
immutable_only: None,
block_filter: None,
})
.expect("query without explicit range should parse");
@ -183,10 +121,10 @@ mod tests {
let request = BlocksStreamRequest::try_from(BlocksStreamQuery {
slot_from: None,
slot_to: None,
descending: None,
blocks_limit: Some(7),
order: None,
blocks_limit: NonZeroUsize::new(7),
server_batch_size: None,
immutable_only: None,
block_filter: None,
})
.expect("query with explicit blocks_limit should parse");
@ -199,10 +137,10 @@ mod tests {
let request = BlocksStreamRequest::try_from(BlocksStreamQuery {
slot_from: Some(10),
slot_to: Some(20),
descending: Some(true),
order: Some(BlockSortOrder::Descending),
blocks_limit: None,
server_batch_size: None,
immutable_only: None,
block_filter: None,
})
.expect("query with explicit slot range should parse");
@ -216,37 +154,41 @@ mod tests {
}
#[test]
fn blocks_stream_request_rejects_zero_limit() {
let result = BlocksStreamRequest::try_from(BlocksStreamQuery {
slot_from: None,
slot_to: None,
descending: None,
blocks_limit: Some(0),
server_batch_size: None,
immutable_only: None,
});
fn blocks_stream_request_maps_typed_order_and_filter() {
let request = BlocksStreamRequest::try_from(BlocksStreamQuery {
slot_from: Some(10),
slot_to: Some(20),
order: Some(BlockSortOrder::Ascending),
blocks_limit: NonZeroUsize::new(7),
server_batch_size: NonZeroUsize::new(3),
block_filter: Some(BlockFilter::ImmutableOnly),
})
.expect("typed query should parse");
match result.err() {
Some(BlocksStreamRequestError::Validation { .. }) => {}
_ => panic!("Expected validation error for 'blocks_limit'"),
}
assert!(!request.descending);
assert_eq!(request.blocks_limit, NonZero::new(7).unwrap());
assert_eq!(request.server_batch_size, NonZero::new(3).unwrap());
assert!(request.immutable_only);
}
#[test]
fn blocks_stream_request_rejects_zero_batch_size() {
let result = BlocksStreamRequest::try_from(BlocksStreamQuery {
slot_from: None,
slot_to: None,
descending: None,
blocks_limit: None,
server_batch_size: Some(0),
immutable_only: None,
});
fn blocks_stream_query_rejects_zero_limit_at_deserialization() {
let result = serde_urlencoded::from_str::<BlocksStreamQuery>("blocks_limit=0");
match result.err() {
Some(BlocksStreamRequestError::Validation { .. }) => {}
_ => panic!("Expected validation error for 'server_batch_size'"),
}
assert!(
result.is_err(),
"zero blocks_limit should fail to deserialize"
);
}
#[test]
fn blocks_stream_query_rejects_zero_batch_size_at_deserialization() {
let result = serde_urlencoded::from_str::<BlocksStreamQuery>("server_batch_size=0");
assert!(
result.is_err(),
"zero server_batch_size should fail to deserialize"
);
}
#[test]
@ -254,10 +196,10 @@ mod tests {
let result = BlocksStreamRequest::try_from(BlocksStreamQuery {
slot_from: Some(9),
slot_to: Some(7),
descending: None,
order: None,
blocks_limit: None,
server_batch_size: None,
immutable_only: None,
block_filter: None,
});
match result.err() {

73
nodes/node/http-client/src/lib.rs
View File

@ -1,4 +1,4 @@
use std::{num::NonZero, sync::Arc};
use std::sync::Arc;
use futures::{Stream, StreamExt as _, TryStreamExt as _};
pub use lb_chain_broadcast_service::BlockInfo;
@ -22,6 +22,7 @@ use lb_http_api_common::{
CRYPTARCHIA_LIB_STREAM, MEMPOOL_ADD_TX, SDP_POST_DECLARATION,
wallet::{BALANCE, TRANSACTIONS_TRANSFER_FUNDS},
},
queries::BlocksStreamQuery,
settings::default_max_body_size,
};
use lb_key_management_system_keys::keys::ZkPublicKey;
@ -88,41 +89,6 @@ impl BasicAuthCredentials {
}
}
#[derive(Default, Clone, Debug)]
struct BlocksStreamQueryParams {
blocks_limit: Option<NonZero<usize>>,
slot_from: Option<u64>,
slot_to: Option<u64>,
descending: Option<bool>,
server_batch_size: Option<NonZero<usize>>,
immutable_only: Option<bool>,
}
impl BlocksStreamQueryParams {
fn append_to_url(&self, request_url: &mut Url) {
let mut query = request_url.query_pairs_mut();
if let Some(blocks_limit) = self.blocks_limit {
query.append_pair("blocks_limit", &blocks_limit.to_string());
}
if let Some(slot_from) = self.slot_from {
query.append_pair("slot_from", &slot_from.to_string());
}
if let Some(slot_to) = self.slot_to {
query.append_pair("slot_to", &slot_to.to_string());
}
if let Some(descending) = self.descending {
query.append_pair("descending", &descending.to_string());
}
if let Some(server_batch_size) = self.server_batch_size {
query.append_pair("server_batch_size", &server_batch_size.to_string());
}
if self.immutable_only == Some(true) {
query.append_pair("immutable_only", "true");
}
}
}
#[derive(Clone)]
pub struct CommonHttpClient {
client: Arc<Client>,
@ -347,9 +313,11 @@ impl CommonHttpClient {
}
}
fn build_blocks_range_stream_request_url(
/// Build the request URL for the `get_blocks_range_stream` method with the
/// given parameters.
pub fn build_blocks_range_stream_request_url(
base_url: &Url,
params: &BlocksStreamQueryParams,
params: &BlocksStreamQuery,
) -> Result<Url, Error> {
let mut request_url = base_url
.join(BLOCKS_RANGE_STREAM.trim_start_matches('/'))
@ -388,10 +356,10 @@ impl CommonHttpClient {
// Helper function to validate inputs for block streaming methods.
fn verify_inputs(
blocks_limit: Option<NonZero<usize>>,
blocks_limit: Option<std::num::NonZero<usize>>,
slot_from: Option<u64>,
slot_to: Option<u64>,
server_batch_size: Option<NonZero<usize>>,
server_batch_size: Option<std::num::NonZero<usize>>,
) -> Result<(), Error> {
if let Some(blocks) = blocks_limit
&& blocks.get() > MAX_BLOCKS_STREAM_BLOCKS
@ -422,29 +390,18 @@ impl CommonHttpClient {
///
/// `server_batch_size` lets callers request smaller chunks; the server
/// still enforces its own upper bound.
#[expect(clippy::too_many_arguments, reason = "Need all args")]
pub async fn get_blocks_range_stream(
&self,
base_url: Url,
blocks_limit: Option<NonZero<usize>>,
slot_from: Option<u64>,
slot_to: Option<u64>,
descending: Option<bool>,
server_batch_size: Option<NonZero<usize>>,
immutable_only: Option<bool>,
params: BlocksStreamQuery,
) -> Result<impl Stream<Item = ProcessedBlockEvent> + use<>, Error> {
Self::verify_inputs(blocks_limit, slot_from, slot_to, server_batch_size)?;
let params = BlocksStreamQueryParams {
blocks_limit,
slot_from,
slot_to,
descending,
server_batch_size,
immutable_only,
};
let request_url = Self::build_blocks_range_stream_request_url(&base_url, &params)?;
Self::verify_inputs(
params.blocks_limit,
params.slot_from,
params.slot_to,
params.server_batch_size,
)?;
let response = self.send_blocks_range_stream_request(request_url).await?;
Ok(Self::parse_processed_blocks_range_event_stream(response))
}

26
services/api/src/http/mantle.rs
View File

@ -415,17 +415,31 @@ where
let mut blocks = Vec::with_capacity(limit.min(1024));
let mut current_id = chain_info.tip;
let gated_slot_from = slot_from.max(chain_info.lib_slot + 1);
if gated_slot_from > slot_to {
return Ok(Vec::new());
}
let mut retried = false;
loop {
// This function only serves the mutable window. Once we hit LIB we are below
// the requested mutable range and should stop without loading the body.
if current_id == chain_info.lib {
break;
}
let Some(block) = storage_adapter.get_block(&current_id).await else {
if retried {
return Err(format!(
"canonical chain inconsistency: missing block for canonical header {current_id}"
"canonical chain inconsistency: missing block {current_id} while traversing \
mutable chain anchored at LIB {}",
chain_info.lib
)
.into());
}
// Retry once from the latest tip if the original tip is not yet available.
// The original LIB remains the anchor for this request.
let refreshed_info =
crate::http::consensus::cryptarchia_info::<RuntimeServiceId>(handle).await?;
current_id = refreshed_info.cryptarchia_info.tip;
@ -438,7 +452,7 @@ where
let slot = header.slot();
let parent_id = header.parent_block();
if slot < slot_from {
if slot < gated_slot_from {
break;
}
@ -456,8 +470,14 @@ where
}
}
// Defensive guard against malformed/self-parenting headers.
if parent_id == current_id {
break;
return Err(format!(
"canonical chain inconsistency: block {current_id} at slot {slot:?} is its own\
parent before anchored LIB {}",
chain_info.lib
)
.into());
}
current_id = parent_id;
}

1
tests/src/cucumber/defaults.rs
View File

@ -16,6 +16,7 @@ const TARGET: &str = "cucumber_defaults";
const LOGOS_BLOCKCHAIN_TESTS_TRACING: &str = "LOGOS_BLOCKCHAIN_TESTS_TRACING";
const TF_KEEP_LOGS: &str = "TF_KEEP_LOGS";
pub const E2E_KEEP_LOGS: &str = "E2E_KEEP_LOGS";
const CUCUMBER_LOG_LEVEL: &str = "CUCUMBER_LOG_LEVEL";
const RUST_LOG: &str = "RUST_LOG";
const LOGOS_BLOCKCHAIN_LOG_DIR: &str = "LOGOS_BLOCKCHAIN_LOG_DIR";

636
tests/src/tests/cryptarchia/blocks_streaming.rs
View File

@ -7,7 +7,11 @@ use std::{
use futures::stream::{self, StreamExt as _};
use lb_common_http_client::ProcessedBlockEvent;
use lb_core::header::HeaderId;
use lb_http_api_common::{DEFAULT_NUMBER_OF_BLOCKS_TO_STREAM, paths::BLOCKS_RANGE_STREAM};
use lb_http_api_common::{
DEFAULT_NUMBER_OF_BLOCKS_TO_STREAM, MAX_BLOCKS_STREAM_BLOCKS, MAX_BLOCKS_STREAM_CHUNK_SIZE,
paths::BLOCKS_RANGE_STREAM,
queries::{BlockFilter, BlockSortOrder, BlocksStreamQuery},
};
use lb_node::config::RunConfig;
use lb_testing_framework::{
DeploymentBuilder, LbcEnv, NodeHttpClient, TopologyConfig as TfTopologyConfig,
@ -236,29 +240,23 @@ fn slot_for_height(chain: &CanonicalChain, height: usize) -> u64 {
async fn request_stream_events(
node: &NodeHttpClient,
blocks_limit: Option<NonZero<usize>>,
slot_from: Option<u64>,
slot_to: Option<u64>,
descending: Option<bool>,
chunk_size: Option<NonZero<usize>>,
immutable_only: Option<bool>,
params: BlocksStreamQuery,
) -> Vec<ProcessedBlockEvent> {
let start = Instant::now();
print!(
" request_stream_events: blocks_limit={blocks_limit:?}, slot_from={slot_from:?}, \
slot_to={slot_to:?}, descending={descending:?}, chunk_size={chunk_size:?}, \
immutable_only={immutable_only:?}"
" request_stream_events: blocks_limit={:?}, slot_from={:?}, \
slot_to={:?}, descending={:?}, chunk_size={:?}, \
immutable_only={:?}",
params.blocks_limit,
params.slot_from,
params.slot_to,
params.order,
params.server_batch_size,
params.block_filter
);
let stream = node
.blocks_range_stream(
blocks_limit,
slot_from,
slot_to,
descending,
chunk_size,
immutable_only,
)
.blocks_range_stream(params)
.await
.expect("blocks stream request should succeed");
@ -338,6 +336,16 @@ fn ids_in_slot_range(
Box::new(chain.heights_by_slot.range(slot_from..=slot_to))
};
let max_height = chain
.ids_by_height
.keys()
.collect::<Vec<_>>()
.iter()
.max()
.map(|&&v| nz(v))
.expect("slot-mapped height must exist in canonical chain")
.get();
let blocks_limit = blocks_limit.min(max_height);
iter.take(blocks_limit)
.map(|(_, height)| {
*chain
@ -398,12 +406,19 @@ async fn test_blocks_streaming() {
chain = refresh_chain(node, &chain).await;
let events = request_stream_events(
node,
blocks_limit,
None,
None,
Some(descending),
chunk_size,
Some(true),
BlocksStreamQuery {
slot_from: None,
slot_to: None,
order: if descending {
// None should default to `descending = true`
None
} else {
Some(BlockSortOrder::Ascending)
},
blocks_limit,
server_batch_size: chunk_size,
block_filter: Some(BlockFilter::ImmutableOnly),
},
)
.await;
@ -419,222 +434,93 @@ async fn test_blocks_streaming() {
}
}
// case: single block below LIB
println!("case: single block below LIB");
// single block below LIB, at LIB and above LIB
println!("case: single block below LIB, at LIB and above LIB");
chain = refresh_chain(node, &chain).await;
let target_height = nz(chain.lib_height - 3);
let (slot_from, slot_to) = blocks_request(&chain, target_height, target_height);
let events = request_stream_events(
node,
None,
Some(slot_from),
Some(slot_to),
None,
None,
Some(false),
)
.await;
assert_eq!(events.len(), 1);
let expected_id = *chain
.ids_by_height
.get(&target_height.get())
.expect("target height should exist on canonical chain");
assert_eq!(
events[0].block.header.id, expected_id,
"slot range should include requested header"
);
assert_stream_integrity(&chain, &events);
// case: single block at LIB
println!("case: single block at LIB");
chain = refresh_chain(node, &chain).await;
let target_height = nz(chain.lib_height);
let (slot_from, slot_to) = (chain.lib_slot, chain.lib_slot);
let events = request_stream_events(
node,
None,
Some(slot_from),
Some(slot_to),
None,
None,
Some(false),
)
.await;
let expected_id = *chain
.ids_by_height
.get(&target_height.get())
.expect("target height should exist on canonical chain");
assert_eq!(
events[0].block.header.id, expected_id,
"slot range should include requested header"
);
assert_stream_integrity(&chain, &events);
// case: single block above LIB
println!("case: single block above LIB");
chain = refresh_chain(node, &chain).await;
let target_height = nz(chain.lib_height + 1);
let (slot_from, slot_to) = blocks_request(&chain, target_height, target_height);
let events = request_stream_events(
node,
None,
Some(slot_from),
Some(slot_to),
None,
None,
Some(false),
)
.await;
let expected_id = *chain
.ids_by_height
.get(&target_height.get())
.expect("target height should exist on canonical chain");
assert_eq!(
events[0].block.header.id, expected_id,
"slot range should include requested header"
);
assert_stream_integrity(&chain, &events);
// case: three blocks up to LIB, various chunk sizes
println!("case: three blocks up to LIB, various chunk sizes");
let blocks_limit = None;
for chunk_size in [
None,
Some(nz(1)),
Some(nz(4)),
Some(nz(chain.lib_height)),
Some(nz(chain.tip_height + 10)),
for target_height in [
nz(chain.lib_height - 3),
nz(chain.lib_height),
nz(chain.lib_height + 1),
] {
for descending in [false, true] {
chain = refresh_chain(node, &chain).await;
let blocks_from = nz(chain.lib_height - 2);
let blocks_to = nz(chain.lib_height);
let (slot_from, slot_to) = blocks_request(&chain, blocks_from, blocks_to);
let events = request_stream_events(
node,
blocks_limit,
Some(slot_from),
Some(slot_to),
Some(descending),
chunk_size,
Some(false),
)
.await;
chain = refresh_chain(node, &chain).await;
let (slot_from, slot_to) = blocks_request(&chain, target_height, target_height);
let events = request_stream_events(
node,
BlocksStreamQuery {
slot_from: Some(slot_from),
slot_to: Some(slot_to),
order: None,
blocks_limit: None,
server_batch_size: None,
block_filter: None,
},
)
.await;
let expected_ids =
ids_in_slot_range(&chain, slot_from, slot_to, descending, blocks_limit);
assert_event_order_matches_expected(&events, &expected_ids);
assert_stream_integrity(&chain, &events);
}
let expected_id = *chain
.ids_by_height
.get(&target_height.get())
.expect("target height should exist on canonical chain");
assert_eq!(
events[0].block.header.id, expected_id,
"slot range should include requested header"
);
assert_stream_integrity(&chain, &events);
}
// case: three blocks from LIB and up, various chunk sizes
println!("case: three blocks from LIB and up, various chunk sizes");
// case: three blocks up to LIB, three blocks from LIB, all blocks, limited
// blocks, various chunk sizes
println!(
"case: three blocks up to LIB, three blocks from LIB, all blocks, limited blocks, various \
chunk sizes"
);
let blocks_limit = None;
for chunk_size in [
None,
Some(nz(1)),
Some(nz(4)),
Some(nz(chain.lib_height)),
Some(nz(chain.tip_height + 10)),
for (blocks_limit, blocks_from, blocks_to) in [
(None, nz(chain.lib_height - 2), nz(chain.lib_height)),
(None, nz(chain.lib_height), nz(chain.lib_height + 2)),
(None, nz(1), nz(chain.tip_height)),
(Some(nz(3)), nz(1), nz(chain.tip_height)),
(
Some(nz(MAX_BLOCKS_STREAM_BLOCKS)),
nz(1),
nz(chain.tip_height),
),
] {
for descending in [false, true] {
chain = refresh_chain(node, &chain).await;
let blocks_from = nz(chain.lib_height);
let blocks_to = nz(chain.lib_height + 2);
let (slot_from, slot_to) = blocks_request(&chain, blocks_from, blocks_to);
let events = request_stream_events(
node,
blocks_limit,
Some(slot_from),
Some(slot_to),
Some(descending),
chunk_size,
Some(false),
)
.await;
for chunk_size in [
None,
Some(nz(1)),
Some(nz(4)),
Some(nz(chain.lib_height)),
Some(nz(chain.tip_height + 10)),
Some(nz(MAX_BLOCKS_STREAM_CHUNK_SIZE)),
] {
for descending in [false, true] {
chain = refresh_chain(node, &chain).await;
let (slot_from, slot_to) = blocks_request(&chain, blocks_from, blocks_to);
let events = request_stream_events(
node,
BlocksStreamQuery {
slot_from: Some(slot_from),
slot_to: Some(slot_to),
order: Some(if descending {
BlockSortOrder::Descending
} else {
BlockSortOrder::Ascending
}),
blocks_limit,
server_batch_size: chunk_size,
block_filter: Some(BlockFilter::MutableAndImmutable),
},
)
.await;
let expected_ids =
ids_in_slot_range(&chain, slot_from, slot_to, descending, blocks_limit);
assert_event_order_matches_expected(&events, &expected_ids);
assert_stream_integrity(&chain, &events);
}
}
// case: all blocks, various chunk sizes
println!("case: all blocks, various chunk sizes");
let blocks_limit = None;
for chunk_size in [
None,
Some(nz(1)),
Some(nz(4)),
Some(nz(chain.lib_height)),
Some(nz(chain.tip_height + 10)),
] {
for descending in [false, true] {
chain = refresh_chain(node, &chain).await;
let (slot_from, slot_to) = (0, chain.tip_slot);
let events = request_stream_events(
node,
blocks_limit,
Some(slot_from),
Some(slot_to),
Some(descending),
chunk_size,
Some(false),
)
.await;
let expected_ids =
ids_in_slot_range(&chain, slot_from, slot_to, descending, blocks_limit);
assert_event_order_matches_expected(&events, &expected_ids);
assert_stream_integrity(&chain, &events);
}
}
// case: limited blocks, various chunk sizes
println!("case: limited blocks, various chunk sizes");
let blocks_limit = Some(nz(3));
for chunk_size in [
None,
Some(nz(1)),
Some(nz(4)),
Some(nz(chain.lib_height)),
Some(nz(chain.tip_height + 10)),
] {
for descending in [false, true] {
chain = refresh_chain(node, &chain).await;
let (slot_from, slot_to) = (0, chain.tip_slot);
let events = request_stream_events(
node,
blocks_limit,
Some(slot_from),
Some(slot_to),
Some(descending),
chunk_size,
Some(false),
)
.await;
let expected_ids =
ids_in_slot_range(&chain, slot_from, slot_to, descending, blocks_limit);
assert_event_order_matches_expected(&events, &expected_ids);
assert_stream_integrity(&chain, &events);
let expected_ids =
ids_in_slot_range(&chain, slot_from, slot_to, descending, blocks_limit);
assert_event_order_matches_expected(&events, &expected_ids);
assert_stream_integrity(&chain, &events);
}
}
}
@ -646,12 +532,14 @@ async fn test_blocks_streaming() {
let blocks_limit = 7;
let events = request_stream_events(
node,
Some(nz(blocks_limit)),
None,
Some(tip_slot),
Some(false),
None,
Some(false),
BlocksStreamQuery {
slot_from: None,
slot_to: Some(tip_slot),
order: Some(BlockSortOrder::Ascending),
blocks_limit: Some(nz(blocks_limit)),
server_batch_size: None,
block_filter: Some(BlockFilter::MutableAndImmutable),
},
)
.await;
@ -670,108 +558,59 @@ async fn test_blocks_streaming() {
"best-effort ascending request should preserve ascending order"
);
// ============== Failure modes =============
// ============== Clamping (soft failures) =============
// case: single block above LIB (immutable only, should fail)
println!("case: single block above LIB (immutable only, should fail)");
chain = refresh_chain(node, &chain).await;
let target_height = nz(chain.lib_height + 3);
let (slot_from, slot_to) = blocks_request(&chain, target_height, target_height);
let Err(err) = node
.blocks_range_stream(None, Some(slot_from), Some(slot_to), None, None, Some(true))
.await
else {
panic!("immutable-only request above LIB should fail");
};
assert!(
matches!(err, lb_common_http_client::Error::Server(ref message) if message.contains("lib_slot")),
"immutable-only request above LIB should mention lib_slot, got: {err}"
);
// case: three blocks from LIB and up (immutable only, should fail)
println!("case: three blocks from LIB and up (immutable only, should fail)");
// case: three blocks from LIB and up (immutable only, slot_to clamps to LIB)
println!("case: three blocks from LIB and up (immutable only, slot_to clamps to LIB)");
chain = refresh_chain(node, &chain).await;
let blocks_from = nz(chain.lib_height);
let blocks_to = nz(chain.lib_height + 2);
let (slot_from, slot_to) = blocks_request(&chain, blocks_from, blocks_to);
let Err(err) = node
.blocks_range_stream(None, Some(slot_from), Some(slot_to), None, None, Some(true))
.await
else {
panic!("immutable-only request above LIB should fail");
};
assert!(
matches!(err, lb_common_http_client::Error::Server(ref message) if message.contains("lib_slot")),
"immutable-only request above LIB should mention lib_slot, got: {err}"
);
let events = request_stream_events(
node,
BlocksStreamQuery {
slot_from: Some(slot_from),
slot_to: Some(slot_to),
order: None,
blocks_limit: None,
server_batch_size: None,
block_filter: Some(BlockFilter::ImmutableOnly),
},
)
.await;
let expected_ids = ids_in_slot_range(&chain, slot_from, chain.lib_slot, true, None);
assert_event_order_matches_expected(&events, &expected_ids);
assert_stream_integrity(&chain, &events);
// case: all blocks, small chunked (immutable only, should fail above LIB)
println!("case: all blocks, small chunked (immutable only, should fail above LIB)");
// case: all blocks, small chunked (immutable only, slot_to clamps to LIB)
println!("case: all blocks, small chunked (immutable only, slot_to clamps to LIB)");
chain = refresh_chain(node, &chain).await;
let blocks_from = nz(1);
let blocks_to = nz(chain.tip_height);
let (slot_from, slot_to) = blocks_request(&chain, blocks_from, blocks_to);
let Err(err) = node
.blocks_range_stream(
None,
Some(slot_from),
Some(slot_to),
None,
Some(nz(4)),
Some(true),
)
.await
else {
panic!("immutable-only request above LIB should fail");
};
assert!(
matches!(err, lb_common_http_client::Error::Server(ref message) if message.contains("lib_slot")),
"immutable-only request above LIB should mention lib_slot, got: {err}"
);
let events = request_stream_events(
node,
BlocksStreamQuery {
slot_from: Some(slot_from),
slot_to: Some(slot_to),
order: None,
blocks_limit: None,
server_batch_size: Some(nz(4)),
block_filter: Some(BlockFilter::ImmutableOnly),
},
)
.await;
let expected_ids = ids_in_slot_range(&chain, slot_from, chain.lib_slot, true, None);
assert_event_order_matches_expected(&events, &expected_ids);
assert_stream_integrity(&chain, &events);
// case: blocks_limit=0 should fail (400) via raw HTTP query
println!("case: blocks_limit=0 should fail (400) via raw HTTP query");
// case: slot_to above tip should clamp to tip and succeed
println!("case: slot_to above tip should clamp to tip and succeed via raw HTTP query");
let tip_slot = u64::from(
node.consensus_info()
.await
.expect("fetching consensus info should succeed")
.cryptarchia_info
.slot,
);
let client = reqwest::Client::new();
let mut url = node.base_url().clone();
url.set_path(BLOCKS_RANGE_STREAM);
url.set_query(Some(&format!("blocks_limit=0&slot_to={tip_slot}")));
let resp = client
.get(url)
.send()
.await
.expect("raw blocks/stream request should complete");
assert_eq!(
resp.status(),
reqwest::StatusCode::BAD_REQUEST,
"blocks_limit=0 must return 400"
);
let body = resp
.text()
.await
.expect("error response body should be readable");
assert!(
body.contains("blocks_limit"),
"400 body should mention blocks_limit, got: {body}"
);
// case: slot_to above tip should fail (400)
println!("case: slot_to above tip should fail (400) via raw HTTP query");
chain = refresh_chain(node, &chain).await;
let mut url = node.base_url().clone();
url.set_path(BLOCKS_RANGE_STREAM);
url.set_query(Some(&format!("slot_to={}", tip_slot + 1)));
@ -784,31 +623,33 @@ async fn test_blocks_streaming() {
assert_eq!(
resp.status(),
reqwest::StatusCode::BAD_REQUEST,
"slot_to above tip must return 400"
reqwest::StatusCode::OK,
"slot_to above tip must clamp and return 200"
);
let body = resp
.text()
.await
.expect("error response body should be readable");
assert!(
body.contains("tip_slot"),
"400 body should mention tip_slot, got: {body}"
);
.expect("stream response body should be readable");
let events = body
.lines()
.map(|line| {
serde_json::from_str::<ProcessedBlockEvent>(line)
.expect("ndjson line should decode into ProcessedBlockEvent")
})
.collect::<Vec<_>>();
let expected_ids = ids_in_slot_range(&chain, 0, chain.tip_slot, true, None);
assert_event_order_matches_expected(&events, &expected_ids);
assert_stream_integrity(&chain, &events);
// case: immutable_only=true with slot_to above LIB should fail (400)
// case: immutable_only=true with slot_to above LIB should clamp to LIB and
// succeed
println!(
"case: immutable_only=true with slot_to above LIB should fail (400) via raw HTTP query"
"case: immutable_only=true with slot_to above LIB should clamp to LIB and succeed via raw HTTP query"
);
let lib_slot = u64::from(
node.consensus_info()
.await
.expect("fetching consensus info should succeed")
.cryptarchia_info
.lib_slot,
);
chain = refresh_chain(node, &chain).await;
let (_, lib_slot) = current_tip_lib_slot(node).await;
let mut url = node.base_url().clone();
url.set_path(BLOCKS_RANGE_STREAM);
url.set_query(Some(&format!(
@ -824,20 +665,123 @@ async fn test_blocks_streaming() {
assert_eq!(
resp.status(),
reqwest::StatusCode::BAD_REQUEST,
"immutable_only=true with slot_to above LIB must return 400"
reqwest::StatusCode::OK,
"immutable_only=true with slot_to above LIB must clamp and return 200"
);
let body = resp
.text()
.await
.expect("error response body should be readable");
.expect("stream response body should be readable");
let events = body
.lines()
.map(|line| {
serde_json::from_str::<ProcessedBlockEvent>(line)
.expect("ndjson line should decode into ProcessedBlockEvent")
})
.collect::<Vec<_>>();
let expected_ids = ids_in_slot_range(&chain, 0, chain.lib_slot, true, None);
assert_event_order_matches_expected(&events, &expected_ids);
assert_stream_integrity(&chain, &events);
// ============== Failure modes =============
// case: single block above LIB (immutable only, explicit slot_from stays above
// clamped LIB)
println!(
"case: single block above LIB (immutable only, explicit slot_from stays above clamped LIB)"
);
chain = refresh_chain(node, &chain).await;
let target_height = nz(chain.lib_height + 3);
let (slot_from, slot_to) = blocks_request(&chain, target_height, target_height);
let Err(err) = node
.blocks_range_stream(BlocksStreamQuery {
slot_from: Some(slot_from),
slot_to: Some(slot_to),
order: None,
blocks_limit: None,
server_batch_size: None,
block_filter: Some(BlockFilter::ImmutableOnly),
})
.await
else {
panic!("explicit slot_from above clamped LIB should fail");
};
assert!(
body.contains("lib_slot"),
"400 body should mention lib_slot, got: {body}"
matches!(err, lb_common_http_client::Error::Server(ref message) if message.contains("slot_from")),
"invalid clamped range should mention 'slot_from', got: {err}"
);
// case: blocks_limit=0 and server_batch_size=0 should fail (400) via raw HTTP
// query
println!("case: blocks_limit=0 and server_batch_size=0 should fail (400) via raw HTTP query");
let (tip_slot, _) = current_tip_lib_slot(node).await;
for query_part in ["blocks_limit=0", "server_batch_size=0"] {
let mut url = node.base_url().clone();
url.set_path(BLOCKS_RANGE_STREAM);
url.set_query(Some(&format!("{query_part}&slot_to={tip_slot}")));
let resp = client
.get(url)
.send()
.await
.expect("raw blocks/stream request should complete");
assert_eq!(
resp.status(),
reqwest::StatusCode::BAD_REQUEST,
"'{query_part}' must return 400"
);
let body = resp
.text()
.await
.expect("error response body should be readable");
assert!(
body.contains("expected a nonzero usize"),
"400 body should mention 'expected a nonzero usize', got: {body}"
);
}
// case: slot_from > slot_to should fail client-side validation
println!("case: slot_from > slot_to should fail client-side validation");
chain = refresh_chain(node, &chain).await;
let Err(err) = node
.blocks_range_stream(BlocksStreamQuery {
slot_from: Some(chain.lib_slot),
slot_to: Some(chain.lib_slot / 2),
order: None,
blocks_limit: None,
server_batch_size: None,
block_filter: Some(BlockFilter::MutableAndImmutable),
})
.await
else {
panic!("slot_from > slot_to should fail client-side validation");
};
assert!(
matches!(err, lb_common_http_client::Error::Client(ref message)
if message.contains("slot_from") && message.contains("slot_to")),
"expected client-side slot range validation error, got: {err}"
);
}
const fn nz(value: usize) -> NonZero<usize> {
NonZero::new(value).unwrap()
}
async fn current_tip_lib_slot(node: &NodeHttpClient) -> (u64, u64) {
let cryptarchia_info = node
.consensus_info()
.await
.expect("fetching consensus info should succeed")
.cryptarchia_info;
(
u64::from(cryptarchia_info.slot),
u64::from(cryptarchia_info.lib_slot),
)
}

20
tests/testing_framework/src/node/http_client.rs
View File

@ -1,4 +1,4 @@
use std::{net::SocketAddr, num::NonZero, pin::Pin};
use std::{net::SocketAddr, pin::Pin};
use common_http_client::{
ApiBlock, BasicAuthCredentials, CommonHttpClient, Error, ProcessedBlockEvent,
@ -12,6 +12,7 @@ use lb_http_api_common::{
WalletTransferFundsRequestBody, WalletTransferFundsResponseBody,
},
paths::{BLEND_NETWORK_INFO, DIAL_PEER, MANTLE_METRICS, MANTLE_SDP_DECLARATIONS, NETWORK_INFO},
queries::BlocksStreamQuery,
};
use lb_libp2p::{Multiaddr, PeerId};
use lb_network_service::backends::libp2p::Libp2pInfo;
@ -113,24 +114,11 @@ impl NodeHttpClient {
/// range.
pub async fn blocks_range_stream(
&self,
blocks_limit: Option<NonZero<usize>>,
slot_from: Option<u64>,
slot_to: Option<u64>,
descending: Option<bool>,
server_batch_size: Option<NonZero<usize>>,
immutable_only: Option<bool>,
params: BlocksStreamQuery,
) -> Result<Pin<Box<dyn Stream<Item = ProcessedBlockEvent> + Send + '_>>, Error> {
let stream = self
.http_client
.get_blocks_range_stream(
self.base_url.clone(),
blocks_limit,
slot_from,
slot_to,
descending,
server_batch_size,
immutable_only,
)
.get_blocks_range_stream(self.base_url.clone(), params)
.await?;
Ok(Box::pin(stream))
}

1
zone-sdk/Cargo.toml
View File

@ -20,6 +20,7 @@ hex = { workspace = true }
lb-common-http-client = { workspace = true }
lb-core = { workspace = true }
lb-groth16 = { workspace = true }
lb-http-api-common = { workspace = true }
lb-key-management-system-service = { workspace = true }
rand = { workspace = true }
reqwest = { workspace = true }

27
zone-sdk/src/adapter.rs
View File

@ -1,4 +1,4 @@
use std::{num::NonZero, pin::Pin};
use std::pin::Pin;
use async_trait::async_trait;
use futures::{Stream, stream};
@ -9,6 +9,7 @@ use lb_core::{
header::HeaderId,
mantle::{Op, SignedMantleTx, channel::ChannelState, ops::channel::ChannelId},
};
use lb_http_api_common::queries::BlocksStreamQuery;
use reqwest::Url;
use crate::{Deposit, Withdraw, ZoneBlock, ZoneMessage};
@ -24,12 +25,7 @@ pub trait Node {
async fn blocks_range_stream(
&self,
blocks_limit: Option<NonZero<usize>>,
slot_from: Option<u64>,
slot_to: Option<u64>,
descending: Option<bool>,
server_batch_size: Option<NonZero<usize>>,
immutable_only: Option<bool>,
params: BlocksStreamQuery,
) -> Result<BoxStream<ProcessedBlockEvent>, Error>;
async fn lib_stream(&self) -> Result<BoxStream<BlockInfo>, Error>;
@ -86,24 +82,11 @@ impl Node for NodeHttpClient {
async fn blocks_range_stream(
&self,
blocks_limit: Option<NonZero<usize>>,
slot_from: Option<u64>,
slot_to: Option<u64>,
descending: Option<bool>,
server_batch_size: Option<NonZero<usize>>,
immutable_only: Option<bool>,
params: BlocksStreamQuery,
) -> Result<BoxStream<ProcessedBlockEvent>, Error> {
let stream = self
.client
.get_blocks_range_stream(
self.base_url.clone(),
blocks_limit,
slot_from,
slot_to,
descending,
server_batch_size,
immutable_only,
)
.get_blocks_range_stream(self.base_url.clone(), params)
.await?;
Ok(Box::pin(stream))
}

9
zone-sdk/src/indexer.rs
View File

@ -143,7 +143,6 @@ fn should_skip(message: &ZoneMessage, slot: Slot, skip_until: &mut Option<(MsgId
#[cfg(test)]
mod tests {
use std::num::NonZero;
use async_trait::async_trait;
use lb_common_http_client::{
@ -155,6 +154,7 @@ mod tests {
mantle::{NoteId, SignedMantleTx, ledger::Inputs},
};
use lb_groth16::Fr;
use lb_http_api_common::queries::BlocksStreamQuery;
use super::*;
use crate::{Deposit, ZoneBlock, adapter::BoxStream};
@ -390,12 +390,7 @@ mod tests {
async fn blocks_range_stream(
&self,
_blocks_limit: Option<NonZero<usize>>,
_slot_from: Option<u64>,
_slot_to: Option<u64>,
_descending: Option<bool>,
_server_batch_size: Option<NonZero<usize>>,
_immutable_only: Option<bool>,
_params: BlocksStreamQuery,
) -> Result<BoxStream<ProcessedBlockEvent>, lb_common_http_client::Error> {
Ok(Box::pin(futures::stream::empty()))
}

9
zone-sdk/src/sequencer.rs
View File

@ -1921,7 +1921,6 @@ fn sign_tx(tx_hash: TxHash, signing_key: &Ed25519Key) -> Ed25519Signature {
#[cfg(test)]
mod tests {
use std::num::NonZero;
use async_trait::async_trait;
use lb_common_http_client::{
@ -1932,6 +1931,7 @@ mod tests {
mantle::{Note, Utxo, ledger::Inputs, ops::channel::deposit::DepositOp},
proofs::leader_proof::Groth16LeaderProof,
};
use lb_http_api_common::queries::BlocksStreamQuery;
use lb_key_management_system_service::keys::ZkKey;
use num_bigint::BigUint;
use rand::{RngCore as _, thread_rng};
@ -2190,12 +2190,7 @@ mod tests {
async fn blocks_range_stream(
&self,
_blocks_limit: Option<NonZero<usize>>,
_slot_from: Option<u64>,
_slot_to: Option<u64>,
_descending: Option<bool>,
_server_batch_size: Option<NonZero<usize>>,
_immutable_only: Option<bool>,
_params: BlocksStreamQuery,
) -> Result<BoxStream<ProcessedBlockEvent>, lb_common_http_client::Error> {
unimplemented!()
}