Skip to content

Restore lost concurrency when scanning for triggers #3444

New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

Merged
tilacog merged 1 commit into from
Apr 8, 2022
Merged

Restore lost concurrency when scanning for triggers #3444

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
40 changes: 22 additions & 18 deletions chain/ethereum/src/ethereum_adapter.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@ use graph::prelude::tokio::try_join;
use graph::{
blockchain::{block_stream::BlockWithTriggers, BlockPtr, IngestorError},
prelude::{
anyhow::{self, anyhow, bail, ensure},
anyhow::{self, anyhow, bail, ensure, Context},
async_trait, debug, error, ethabi,
futures03::{self, compat::Future01CompatExt, FutureExt, StreamExt, TryStreamExt},
hex, info, retry, serde_json as json, stream, tiny_keccak, trace, warn,
Expand Down Expand Up @@ -1347,24 +1347,28 @@ pub(crate) async fn blocks_with_triggers(
trigger_futs.push(block_future)
}

// join on triger futures
let triggers: Vec<EthereumTrigger> = trigger_futs.try_concat().await?;

// get hash for "to" block
let to_hash = match adapter
// Get hash for "to" block
let to_hash_fut = adapter
.block_hash_by_block_number(&logger, to)
.compat()
.await?
{
Some(hash) => hash,
None => {
warn!(logger,
"Ethereum endpoint is behind";
"url" => eth.url_hostname()
);
bail!("Block {} not found in the chain", to)
}
};
.and_then(|hash| match hash {
Some(hash) => Ok(hash),
None => {
warn!(logger,
"Ethereum endpoint is behind";
"url" => eth.url_hostname()
);
bail!("Block {} not found in the chain", to)
}
})
.compat();

// Join on triggers and block hash resolution
let (triggers, to_hash) = futures03::join!(trigger_futs.try_concat(), to_hash_fut);

// Unpack and handle possible errors in the previously joined futures
let triggers =
triggers.with_context(|| format!("Failed to obtain triggers for block {}", to))?;
let to_hash = to_hash.with_context(|| format!("Failed to infer hash for block {}", to))?;
Comment on lines +1365 to +1371
Copy link
Copy Markdown
Contributor Author

@tilacog tilacog Apr 7, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I decided not to use try_join! as it discards one error if the two operations fail.

Copy link
Copy Markdown
Collaborator

@leoyvens leoyvens Apr 8, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Well, this will discard the to_hash error if both fail, so it's not so different from try_join, which would be fine imo.

Copy link
Copy Markdown
Contributor Author

@tilacog tilacog Apr 8, 2022 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You have a point.

I was reluctant to use try_join because it would only return the fastest error to resolve and discard the slowest, so I was concerned about indeterminism in this case.

If omitting a second error situation isn't a problem, I can change this to use try_join or just leave it as it is.
If not, I can rewrite this to provide a combined error message for when we have two error values.

Which one do you think would be preferable?

Copy link
Copy Markdown
Collaborator

@leoyvens leoyvens Apr 8, 2022

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

interesting point about determinism. I think we can keep this as-is then.


let mut block_hashes: HashSet<H256> =
triggers.iter().map(EthereumTrigger::block_hash).collect();
Expand Down