Skip to content

Commit e0ccadb

Browse files
dappliondapplion
authored
Move sync active requests to own modules (#6272)
* Move sync active requests to own modules * Merge branch 'unstable' into sync-requests-modules
1 parent 351dd6c commit e0ccadb

File tree

3 files changed

+161
-154
lines changed

3 files changed

+161
-154
lines changed

‎beacon_node/network/src/sync/network_context/requests.rs

+5-154
Original file line numberDiff line numberDiff line change
@@ -1,18 +1,14 @@
1-
use beacon_chain::get_block_root;
2-
use lighthouse_network::{
3-
rpc::{methods::BlobsByRootRequest, BlocksByRootRequest},
4-
PeerId,
5-
};
6-
use std::sync::Arc;
71
use strum::IntoStaticStr;
8-
use types::{
9-
blob_sidecar::BlobIdentifier, BlobSidecar, ChainSpec, EthSpec, Hash256, SignedBeaconBlock,
10-
};
2+
use types::Hash256;
113

4+
pub use blobs_by_root::{ActiveBlobsByRootRequest, BlobsByRootSingleBlockRequest};
5+
pub use blocks_by_root::{ActiveBlocksByRootRequest, BlocksByRootSingleRequest};
126
pub use data_columns_by_root::{
137
ActiveDataColumnsByRootRequest, DataColumnsByRootSingleBlockRequest,
148
};
159

10+
mod blobs_by_root;
11+
mod blocks_by_root;
1612
mod data_columns_by_root;
1713

1814
#[derive(Debug, PartialEq, Eq, IntoStaticStr)]
@@ -25,148 +21,3 @@ pub enum LookupVerifyError {
2521
InvalidInclusionProof,
2622
DuplicateData,
2723
}
28-
29-
pub struct ActiveBlocksByRootRequest {
30-
request: BlocksByRootSingleRequest,
31-
resolved: bool,
32-
pub(crate) peer_id: PeerId,
33-
}
34-
35-
impl ActiveBlocksByRootRequest {
36-
pub fn new(request: BlocksByRootSingleRequest, peer_id: PeerId) -> Self {
37-
Self {
38-
request,
39-
resolved: false,
40-
peer_id,
41-
}
42-
}
43-
44-
/// Append a response to the single chunk request. If the chunk is valid, the request is
45-
/// resolved immediately.
46-
/// The active request SHOULD be dropped after `add_response` returns an error
47-
pub fn add_response<E: EthSpec>(
48-
&mut self,
49-
block: Arc<SignedBeaconBlock<E>>,
50-
) -> Result<Arc<SignedBeaconBlock<E>>, LookupVerifyError> {
51-
if self.resolved {
52-
return Err(LookupVerifyError::TooManyResponses);
53-
}
54-
55-
let block_root = get_block_root(&block);
56-
if self.request.0 != block_root {
57-
return Err(LookupVerifyError::UnrequestedBlockRoot(block_root));
58-
}
59-
60-
// Valid data, blocks by root expects a single response
61-
self.resolved = true;
62-
Ok(block)
63-
}
64-
65-
pub fn terminate(self) -> Result<(), LookupVerifyError> {
66-
if self.resolved {
67-
Ok(())
68-
} else {
69-
Err(LookupVerifyError::NoResponseReturned)
70-
}
71-
}
72-
}
73-
74-
#[derive(Debug, Copy, Clone)]
75-
pub struct BlocksByRootSingleRequest(pub Hash256);
76-
77-
impl BlocksByRootSingleRequest {
78-
pub fn into_request(self, spec: &ChainSpec) -> BlocksByRootRequest {
79-
BlocksByRootRequest::new(vec![self.0], spec)
80-
}
81-
}
82-
83-
#[derive(Debug, Clone)]
84-
pub struct BlobsByRootSingleBlockRequest {
85-
pub block_root: Hash256,
86-
pub indices: Vec<u64>,
87-
}
88-
89-
impl BlobsByRootSingleBlockRequest {
90-
pub fn into_request(self, spec: &ChainSpec) -> BlobsByRootRequest {
91-
BlobsByRootRequest::new(
92-
self.indices
93-
.into_iter()
94-
.map(|index| BlobIdentifier {
95-
block_root: self.block_root,
96-
index,
97-
})
98-
.collect(),
99-
spec,
100-
)
101-
}
102-
}
103-
104-
pub struct ActiveBlobsByRootRequest<E: EthSpec> {
105-
request: BlobsByRootSingleBlockRequest,
106-
blobs: Vec<Arc<BlobSidecar<E>>>,
107-
resolved: bool,
108-
pub(crate) peer_id: PeerId,
109-
}
110-
111-
impl<E: EthSpec> ActiveBlobsByRootRequest<E> {
112-
pub fn new(request: BlobsByRootSingleBlockRequest, peer_id: PeerId) -> Self {
113-
Self {
114-
request,
115-
blobs: vec![],
116-
resolved: false,
117-
peer_id,
118-
}
119-
}
120-
121-
/// Appends a chunk to this multi-item request. If all expected chunks are received, this
122-
/// method returns `Some`, resolving the request before the stream terminator.
123-
/// The active request SHOULD be dropped after `add_response` returns an error
124-
pub fn add_response(
125-
&mut self,
126-
blob: Arc<BlobSidecar<E>>,
127-
) -> Result<Option<Vec<Arc<BlobSidecar<E>>>>, LookupVerifyError> {
128-
if self.resolved {
129-
return Err(LookupVerifyError::TooManyResponses);
130-
}
131-
132-
let block_root = blob.block_root();
133-
if self.request.block_root != block_root {
134-
return Err(LookupVerifyError::UnrequestedBlockRoot(block_root));
135-
}
136-
if !blob.verify_blob_sidecar_inclusion_proof() {
137-
return Err(LookupVerifyError::InvalidInclusionProof);
138-
}
139-
if !self.request.indices.contains(&blob.index) {
140-
return Err(LookupVerifyError::UnrequestedIndex(blob.index));
141-
}
142-
if self.blobs.iter().any(|b| b.index == blob.index) {
143-
return Err(LookupVerifyError::DuplicateData);
144-
}
145-
146-
self.blobs.push(blob);
147-
if self.blobs.len() >= self.request.indices.len() {
148-
// All expected chunks received, return result early
149-
self.resolved = true;
150-
Ok(Some(std::mem::take(&mut self.blobs)))
151-
} else {
152-
Ok(None)
153-
}
154-
}
155-
156-
pub fn terminate(self) -> Result<(), LookupVerifyError> {
157-
if self.resolved {
158-
Ok(())
159-
} else {
160-
Err(LookupVerifyError::NotEnoughResponsesReturned {
161-
expected: self.request.indices.len(),
162-
actual: self.blobs.len(),
163-
})
164-
}
165-
}
166-
167-
/// Mark request as resolved (= has returned something downstream) while marking this status as
168-
/// true for future calls.
169-
pub fn resolve(&mut self) -> bool {
170-
std::mem::replace(&mut self.resolved, true)
171-
}
172-
}

‎beacon_node/network/src/sync/network_context/requests/blobs_by_root.rs

+96
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
use lighthouse_network::{rpc::methods::BlobsByRootRequest, PeerId};
2+
use std::sync::Arc;
3+
use types::{blob_sidecar::BlobIdentifier, BlobSidecar, ChainSpec, EthSpec, Hash256};
4+
5+
use super::LookupVerifyError;
6+
7+
#[derive(Debug, Clone)]
8+
pub struct BlobsByRootSingleBlockRequest {
9+
pub block_root: Hash256,
10+
pub indices: Vec<u64>,
11+
}
12+
13+
impl BlobsByRootSingleBlockRequest {
14+
pub fn into_request(self, spec: &ChainSpec) -> BlobsByRootRequest {
15+
BlobsByRootRequest::new(
16+
self.indices
17+
.into_iter()
18+
.map(|index| BlobIdentifier {
19+
block_root: self.block_root,
20+
index,
21+
})
22+
.collect(),
23+
spec,
24+
)
25+
}
26+
}
27+
28+
pub struct ActiveBlobsByRootRequest<E: EthSpec> {
29+
request: BlobsByRootSingleBlockRequest,
30+
blobs: Vec<Arc<BlobSidecar<E>>>,
31+
resolved: bool,
32+
pub(crate) peer_id: PeerId,
33+
}
34+
35+
impl<E: EthSpec> ActiveBlobsByRootRequest<E> {
36+
pub fn new(request: BlobsByRootSingleBlockRequest, peer_id: PeerId) -> Self {
37+
Self {
38+
request,
39+
blobs: vec![],
40+
resolved: false,
41+
peer_id,
42+
}
43+
}
44+
45+
/// Appends a chunk to this multi-item request. If all expected chunks are received, this
46+
/// method returns `Some`, resolving the request before the stream terminator.
47+
/// The active request SHOULD be dropped after `add_response` returns an error
48+
pub fn add_response(
49+
&mut self,
50+
blob: Arc<BlobSidecar<E>>,
51+
) -> Result<Option<Vec<Arc<BlobSidecar<E>>>>, LookupVerifyError> {
52+
if self.resolved {
53+
return Err(LookupVerifyError::TooManyResponses);
54+
}
55+
56+
let block_root = blob.block_root();
57+
if self.request.block_root != block_root {
58+
return Err(LookupVerifyError::UnrequestedBlockRoot(block_root));
59+
}
60+
if !blob.verify_blob_sidecar_inclusion_proof() {
61+
return Err(LookupVerifyError::InvalidInclusionProof);
62+
}
63+
if !self.request.indices.contains(&blob.index) {
64+
return Err(LookupVerifyError::UnrequestedIndex(blob.index));
65+
}
66+
if self.blobs.iter().any(|b| b.index == blob.index) {
67+
return Err(LookupVerifyError::DuplicateData);
68+
}
69+
70+
self.blobs.push(blob);
71+
if self.blobs.len() >= self.request.indices.len() {
72+
// All expected chunks received, return result early
73+
self.resolved = true;
74+
Ok(Some(std::mem::take(&mut self.blobs)))
75+
} else {
76+
Ok(None)
77+
}
78+
}
79+
80+
pub fn terminate(self) -> Result<(), LookupVerifyError> {
81+
if self.resolved {
82+
Ok(())
83+
} else {
84+
Err(LookupVerifyError::NotEnoughResponsesReturned {
85+
expected: self.request.indices.len(),
86+
actual: self.blobs.len(),
87+
})
88+
}
89+
}
90+
91+
/// Mark request as resolved (= has returned something downstream) while marking this status as
92+
/// true for future calls.
93+
pub fn resolve(&mut self) -> bool {
94+
std::mem::replace(&mut self.resolved, true)
95+
}
96+
}

‎beacon_node/network/src/sync/network_context/requests/blocks_by_root.rs

+60
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,60 @@
1+
use beacon_chain::get_block_root;
2+
use lighthouse_network::{rpc::BlocksByRootRequest, PeerId};
3+
use std::sync::Arc;
4+
use types::{ChainSpec, EthSpec, Hash256, SignedBeaconBlock};
5+
6+
use super::LookupVerifyError;
7+
8+
#[derive(Debug, Copy, Clone)]
9+
pub struct BlocksByRootSingleRequest(pub Hash256);
10+
11+
impl BlocksByRootSingleRequest {
12+
pub fn into_request(self, spec: &ChainSpec) -> BlocksByRootRequest {
13+
BlocksByRootRequest::new(vec![self.0], spec)
14+
}
15+
}
16+
17+
pub struct ActiveBlocksByRootRequest {
18+
request: BlocksByRootSingleRequest,
19+
resolved: bool,
20+
pub(crate) peer_id: PeerId,
21+
}
22+
23+
impl ActiveBlocksByRootRequest {
24+
pub fn new(request: BlocksByRootSingleRequest, peer_id: PeerId) -> Self {
25+
Self {
26+
request,
27+
resolved: false,
28+
peer_id,
29+
}
30+
}
31+
32+
/// Append a response to the single chunk request. If the chunk is valid, the request is
33+
/// resolved immediately.
34+
/// The active request SHOULD be dropped after `add_response` returns an error
35+
pub fn add_response<E: EthSpec>(
36+
&mut self,
37+
block: Arc<SignedBeaconBlock<E>>,
38+
) -> Result<Arc<SignedBeaconBlock<E>>, LookupVerifyError> {
39+
if self.resolved {
40+
return Err(LookupVerifyError::TooManyResponses);
41+
}
42+
43+
let block_root = get_block_root(&block);
44+
if self.request.0 != block_root {
45+
return Err(LookupVerifyError::UnrequestedBlockRoot(block_root));
46+
}
47+
48+
// Valid data, blocks by root expects a single response
49+
self.resolved = true;
50+
Ok(block)
51+
}
52+
53+
pub fn terminate(self) -> Result<(), LookupVerifyError> {
54+
if self.resolved {
55+
Ok(())
56+
} else {
57+
Err(LookupVerifyError::NoResponseReturned)
58+
}
59+
}
60+
}

0 commit comments

Comments
 (0)