Skip to content
New issue

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

#

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

Already on GitHub? # to your account

The SubstreamsBlockStream was using request.clone() side stepping latest cursor value #4228

Merged
merged 1 commit into from
Dec 6, 2022
Merged

The SubstreamsBlockStream was using request.clone() side stepping latest cursor value #4228

Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
26 changes: 13 additions & 13 deletions graph/src/blockchain/substreams_block_stream.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,18 +178,6 @@ fn stream_blocks<C: Blockchain, F: SubstreamsMapper<C>>(

let stop_block_num = manifest_end_block_num as u64;

let request = Request {
start_block_num,
start_cursor: latest_cursor.clone(),
stop_block_num,
fork_steps: vec![StepNew as i32, StepUndo as i32],
irreversibility_condition: "".to_string(),
modules,
output_modules: vec![module_name],
production_mode: true,
..Default::default()
};

// Back off exponentially whenever we encounter a connection error or a stream with bad data
let mut backoff = ExponentialBackoff::new(Duration::from_millis(500), Duration::from_secs(45));

Expand All @@ -211,7 +199,19 @@ fn stream_blocks<C: Blockchain, F: SubstreamsMapper<C>>(
skip_backoff = false;

let mut connect_start = Instant::now();
let result = endpoint.clone().substreams(request.clone()).await;
let request = Request {
start_block_num,
start_cursor: latest_cursor.clone(),
stop_block_num,
fork_steps: vec![StepNew as i32, StepUndo as i32],
irreversibility_condition: "".to_string(),
modules: modules.clone(),
output_modules: vec![module_name.clone()],
production_mode: true,
..Default::default()
};

let result = endpoint.clone().substreams(request).await;

match result {
Ok(stream) => {
Expand Down