Skip to content
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

refactor(0.7.x): 0.7.5 with exposed non-retryable s3_fetchers #98

Merged
merged 2 commits into from
Dec 19, 2023
Merged

refactor(0.7.x): 0.7.5 with exposed non-retryable s3_fetchers #98

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
bcb2864
refactor(0.7.x): 0.7.5 with exposed non-retryable s3_fetchers
khorolets Dec 18, 2023
fd6c30c
LakeError error to debug message
khorolets Dec 19, 2023
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
10 changes: 8 additions & 2 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,9 +5,15 @@ All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).

## [Unreleased](https://github.com/near/near-lake-framework/compare/v0.7.4...HEAD)
## [Unreleased](https://github.com/near/near-lake-framework/compare/v0.7.5...HEAD)

## [0.7.3](https://github.com/near/near-lake-framework/compare/v0.7.2...0.7.4)
## [0.7.5](https://github.com/near/near-lake-framework/compare/v0.7.3...0.7.5)

* Refactor `s3_fetchers` module to allow exposing the underlying functionality:
* `s3_fetchers::fetch_block` (without retrying)
* `s3_fetchers::fetch_shard` (without retrying)

## [0.7.4](https://github.com/near/near-lake-framework/compare/v0.7.2...0.7.4)

* Upgrade all `aws` crates to the latest version
* Undeylying AWS error will now be printed
Expand Down
4 changes: 2 additions & 2 deletions Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,14 +12,15 @@ rust-version = "1.58.1"

# cargo-workspaces
[workspace.metadata.workspaces]
version = "0.7.4"
version = "0.7.5"

[dependencies]
anyhow = "1.0.51"
aws-config = { version = "1.0.0", features = ["behavior-version-latest"] }
aws-types = "1.0.0"
aws-credential-types = "1.0.0"
aws-sdk-s3 = "0.39.1"
aws-smithy-types = "1.0.1"
Copy link
Member Author

Choose a reason for hiding this comment

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

I had to move it from dev-dependencies to a general one since I introduced a variant with aws-smithy-types error to LakeError regarding the byte-stream

async-stream = "0.3.3"
async-trait = "0.1.64"
derive_builder = "0.11.2"
Expand All @@ -35,7 +36,6 @@ near-indexer-primitives = "0.17"

[dev-dependencies]
aws-smithy-http = "0.60.0"
aws-smithy-types = "1.0.1"

[lib]
doctest = false
179 changes: 110 additions & 69 deletions src/s3_fetchers.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,50 +142,71 @@ pub(crate) async fn fetch_streamer_message(
}

/// Fetches the block data JSON from AWS S3 and returns the `BlockView`
pub async fn fetch_block_or_retry(
pub async fn fetch_block(
lake_s3_client: &impl S3Client,
s3_bucket_name: &str,
block_height: crate::types::BlockHeight,
) -> Result<
near_indexer_primitives::views::BlockView,
crate::types::LakeError<aws_sdk_s3::operation::get_object::GetObjectError>,
> {
let body_bytes = loop {
match lake_s3_client
.get_object(s3_bucket_name, &format!("{:0>12}/block.json", block_height))
.await
{
Ok(response) => {
match response.body.collect().await {
Ok(bytes_stream) => break bytes_stream.into_bytes(),
Err(err) => {
tracing::debug!(
target: crate::LAKE_FRAMEWORK,
"Failed to read bytes from the block #{:0>12} response. Retrying immediately.\n{:#?}",
block_height,
err,
);
}
};
}
Err(err) => {
tracing::debug!(
target: crate::LAKE_FRAMEWORK,
"Failed to get {:0>12}/block.json. Retrying immediately\n{:#?}",
block_height,
err
);
}
};
};
let body_bytes = lake_s3_client
.get_object(s3_bucket_name, &format!("{:0>12}/block.json", block_height))
.await?
.body
.collect()
.await?
Copy link
Member Author

Choose a reason for hiding this comment

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

This one will convert into new LakeError::AwsSmithyError

.into_bytes();

Ok(serde_json::from_slice::<
near_indexer_primitives::views::BlockView,
>(body_bytes.as_ref())?)
}

/// Fetches the block data JSON from AWS S3 and returns the `BlockView` retrying until it succeeds (indefinitely)
pub async fn fetch_block_or_retry(
lake_s3_client: &impl S3Client,
s3_bucket_name: &str,
block_height: crate::types::BlockHeight,
) -> Result<
near_indexer_primitives::views::BlockView,
crate::types::LakeError<aws_sdk_s3::operation::get_object::GetObjectError>,
> {
loop {
match fetch_block(lake_s3_client, s3_bucket_name, block_height).await {
Ok(block_view) => break Ok(block_view),
Err(err) => match err {
crate::types::LakeError::AwsError { .. } => {
tracing::debug!(
target: crate::LAKE_FRAMEWORK,
"Block #{:0>12} not found. Retrying in immediately...\n{:#?}",
block_height,
err,
);
}
crate::types::LakeError::AwsSmithyError { .. } => {
tracing::debug!(
target: crate::LAKE_FRAMEWORK,
"Failed to read bytes from the block #{:0>12} response. Retrying immediately.\n{:#?}",
block_height,
err,
);
}
_ => {
tracing::debug!(
target: crate::LAKE_FRAMEWORK,
"Failed to fetch block #{}, retrying immediately\n{:#?}",
block_height,
err
);
}
},
}
}
}

/// Fetches the shard data JSON from AWS S3 and returns the `IndexerShard`
pub async fn fetch_shard_or_retry(
pub async fn fetch_shard(
lake_s3_client: &impl S3Client,
s3_bucket_name: &str,
block_height: crate::types::BlockHeight,
Expand All @@ -194,48 +215,68 @@ pub async fn fetch_shard_or_retry(
near_indexer_primitives::IndexerShard,
crate::types::LakeError<aws_sdk_s3::operation::get_object::GetObjectError>,
> {
let body_bytes = loop {
match lake_s3_client
.get_object(
s3_bucket_name,
&format!("{:0>12}/shard_{}.json", block_height, shard_id),
)
.await
{
Ok(response) => {
let body_bytes = match response.body.collect().await {
Ok(body) => body.into_bytes(),
Err(err) => {
tracing::debug!(
target: crate::LAKE_FRAMEWORK,
"Failed to read the {:0>12}/shard_{}.json. Retrying in 1s...\n {:#?}",
block_height,
shard_id,
err,
);
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
continue;
}
};

break body_bytes;
}
Err(err) => {
tracing::debug!(
target: crate::LAKE_FRAMEWORK,
"Failed to fetch shard #{}, retrying immediately\n{:#?}",
shard_id,
err
);
}
}
};
let body_bytes = lake_s3_client
.get_object(
s3_bucket_name,
&format!("{:0>12}/shard_{}.json", block_height, shard_id),
)
.await?
.body
.collect()
.await?
.into_bytes();

Ok(serde_json::from_slice::<
near_indexer_primitives::IndexerShard,
>(body_bytes.as_ref())?)
}

/// Fetches the shard data JSON from AWS S3 and returns the `IndexerShard`
pub async fn fetch_shard_or_retry(
lake_s3_client: &impl S3Client,
s3_bucket_name: &str,
block_height: crate::types::BlockHeight,
shard_id: u64,
) -> Result<
near_indexer_primitives::IndexerShard,
crate::types::LakeError<aws_sdk_s3::operation::get_object::GetObjectError>,
> {
loop {
match fetch_shard(lake_s3_client, s3_bucket_name, block_height, shard_id).await {
Ok(shard) => break Ok(shard),
Err(err) => match err {
crate::types::LakeError::AwsError { .. } => {
tracing::debug!(
target: crate::LAKE_FRAMEWORK,
"Shard {} of block #{:0>12} not found. Retrying in immediately...\n{:#?}",
shard_id,
block_height,
err,
);
}
crate::types::LakeError::AwsSmithyError { .. } => {
tracing::debug!(
target: crate::LAKE_FRAMEWORK,
"Failed to read bytes from the shard {} of block #{:0>12} response. Retrying immediately.\n{:#?}",
shard_id,
block_height,
err,
);
}
_ => {
tracing::debug!(
target: crate::LAKE_FRAMEWORK,
"Failed to fetch shard {} of block #{}, retrying immediately\n{:#?}",
shard_id,
block_height,
err
);
}
},
}
}
}

#[cfg(test)]
mod test {
use super::*;
Expand All @@ -255,7 +296,7 @@ mod test {
impl S3Client for LakeS3Client {
async fn get_object(
&self,
bucket: &str,
_bucket: &str,
prefix: &str,
) -> Result<
aws_sdk_s3::operation::get_object::GetObjectOutput,
Expand All @@ -269,8 +310,8 @@ mod test {

async fn list_objects(
&self,
bucket: &str,
start_after: &str,
_bucket: &str,
_start_after: &str,
) -> Result<
aws_sdk_s3::operation::list_objects_v2::ListObjectsV2Output,
aws_sdk_s3::error::SdkError<aws_sdk_s3::operation::list_objects_v2::ListObjectsV2Error>,
Expand Down
7 changes: 6 additions & 1 deletion src/types.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,9 +125,14 @@ pub enum LakeError<E> {
#[from]
error: aws_sdk_s3::error::SdkError<E>,
},
#[error("Failed to convert integer")]
#[error("Failed to convert integer: {error}")]
IntConversionError {
#[from]
error: std::num::TryFromIntError,
},
#[error("AWS Smithy byte_stream error: {error}")]
Copy link
Contributor

Choose a reason for hiding this comment

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

Suggested change
#[error("AWS Smithy byte_stream error: {error}")]
#[error("AWS Smithy byte_stream error: {error:?}")]

Can we print the debug message, the normal display message doesn't provide much information. Same for AwsError too please 🙏🏼

AwsSmithyError {
#[from]
error: aws_smithy_types::byte_stream::error::Error,
},
}
Loading