From a60157c51aeba11d852d3e4ee7e23d67a081df22 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Mon, 19 Jun 2023 18:18:14 +0200 Subject: [PATCH 01/15] [Fix] Premature end for exported snapshots --- src/chain/store/chain_store.rs | 8 +++++++- src/utils/io/writer_checksum.rs | 11 +++++++---- 2 files changed, 14 insertions(+), 5 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 69f5bd62b526..732ddecc5003 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -35,6 +35,7 @@ use bls_signatures::Serialize as SerializeBls; use cid::Cid; use digest::Digest; use futures::{io::BufWriter, AsyncWrite}; +use futures_util::AsyncWriteExt; use fvm_ipld_amt::Amtv0 as Amt; use fvm_ipld_blockstore::Blockstore; use fvm_ipld_car::CarHeader; @@ -625,7 +626,12 @@ where let mut writer = writer.lock().await; let digest = match &mut *writer { - Either::Left(left) => left.get_mut().finalize().await, + Either::Left(left) => { + left.close() + .await + .map_err(|e| Error::Other(e.to_string()))?; + left.get_mut().finalize().await + } Either::Right(right) => right.finalize().await, } .map_err(|e| Error::Other(e.to_string()))?; diff --git a/src/utils/io/writer_checksum.rs b/src/utils/io/writer_checksum.rs index 27b92ac98ef1..95cb8db83d6d 100644 --- a/src/utils/io/writer_checksum.rs +++ b/src/utils/io/writer_checksum.rs @@ -4,7 +4,8 @@ use std::{pin::Pin, task::Poll}; use async_trait::async_trait; use digest::{Digest, Output}; -use futures::{io::BufWriter, AsyncWrite, AsyncWriteExt}; +use futures::{io::BufWriter, AsyncWrite}; +use futures_util::AsyncWriteExt; use pin_project_lite::pin_project; pin_project! { @@ -25,7 +26,7 @@ pub trait Checksum { async fn finalize(&mut self) -> std::io::Result>>; } -impl AsyncWrite for AsyncWriterWithChecksum { +impl AsyncWrite for AsyncWriterWithChecksum { fn poll_write( mut self: std::pin::Pin<&mut Self>, cx: &mut std::task::Context<'_>, @@ -59,9 +60,11 @@ impl AsyncWrite for AsyncWriterWithChecksum Checksum for AsyncWriterWithChecksum { +impl Checksum + for AsyncWriterWithChecksum +{ async fn finalize(&mut self) -> std::io::Result>> { - self.inner.flush().await?; + self.inner.get_mut().flush().await; if let Some(hasher) = &mut self.hasher { let hasher = std::mem::replace(hasher, D::new()); Ok(Some(hasher.finalize())) From 7bdff520df49b29268ab21354afd5d8fb233cf59 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Mon, 19 Jun 2023 18:26:05 +0200 Subject: [PATCH 02/15] add snapshot validity test --- scripts/tests/calibnet_export_check.sh | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/scripts/tests/calibnet_export_check.sh b/scripts/tests/calibnet_export_check.sh index d10ac02cdba7..5e7109f3479c 100755 --- a/scripts/tests/calibnet_export_check.sh +++ b/scripts/tests/calibnet_export_check.sh @@ -12,5 +12,8 @@ forest_init echo "Exporting zstd compressed snapshot" $FOREST_CLI_PATH snapshot export -echo "Verifing snapshot checksum" +echo "Testing snapshot validity" +zstd -t ./*.car.zst + +echo "Verifying snapshot checksum" sha256sum -c ./*.sha256sum From 700b462fcd8eda563af66bc0bd6a65d601e57ce9 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Mon, 19 Jun 2023 18:53:01 +0200 Subject: [PATCH 03/15] changelog entry --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 604f5673fc27..ecb947384ece 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -35,6 +35,9 @@ ### Fixed +- [#3006](https://github.com/ChainSafe/forest/issues/3006): Fix `premature end` + error when exporting a snapshot. + ## Forest v0.9.0 "Fellowship" Notable updates: From 5956044a259edff6dee271831f9e5c5747b5328c Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Mon, 19 Jun 2023 19:28:52 +0200 Subject: [PATCH 04/15] remove obsolete code, introduce flush where it needs to be --- src/chain/store/chain_store.rs | 3 +++ src/utils/io/writer_checksum.rs | 1 - 2 files changed, 3 insertions(+), 1 deletion(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 732ddecc5003..727ae8e3cfc2 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -627,6 +627,9 @@ where let mut writer = writer.lock().await; let digest = match &mut *writer { Either::Left(left) => { + left.flush() + .await + .map_err(|e| Error::Other(e.to_string()))?; left.close() .await .map_err(|e| Error::Other(e.to_string()))?; diff --git a/src/utils/io/writer_checksum.rs b/src/utils/io/writer_checksum.rs index 95cb8db83d6d..6faac309898b 100644 --- a/src/utils/io/writer_checksum.rs +++ b/src/utils/io/writer_checksum.rs @@ -64,7 +64,6 @@ impl Checksum for AsyncWriterWithChecksum { async fn finalize(&mut self) -> std::io::Result>> { - self.inner.get_mut().flush().await; if let Some(hasher) = &mut self.hasher { let hasher = std::mem::replace(hasher, D::new()); Ok(Some(hasher.finalize())) From 49b10613c9dba4e7407013384021c241dac2dbc9 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 08:26:30 +0200 Subject: [PATCH 05/15] make sure the writer is flushed and closed --- src/chain/store/chain_store.rs | 12 +++++++++++- src/utils/io/writer_checksum.rs | 1 + 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 727ae8e3cfc2..8b7c40d2fbaf 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -635,7 +635,17 @@ where .map_err(|e| Error::Other(e.to_string()))?; left.get_mut().finalize().await } - Either::Right(right) => right.finalize().await, + Either::Right(right) => { + right + .flush() + .await + .map_err(|e| Error::Other(e.to_string()))?; + right + .close() + .await + .map_err(|e| Error::Other(e.to_string()))?; + right.finalize().await + } } .map_err(|e| Error::Other(e.to_string()))?; diff --git a/src/utils/io/writer_checksum.rs b/src/utils/io/writer_checksum.rs index 6faac309898b..7aa96dc78b26 100644 --- a/src/utils/io/writer_checksum.rs +++ b/src/utils/io/writer_checksum.rs @@ -133,6 +133,7 @@ mod test { let mut bytes = [0; 1024]; OsRng.fill_bytes(&mut bytes); temp_file_writer.write_all(&bytes).await?; + temp_file_writer.flush().await?; } let checksum = temp_file_writer.finalize().await?; From 249ea990de690906b9401f999dc4fc727cbe77e3 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 08:30:50 +0200 Subject: [PATCH 06/15] comment and more changes --- src/chain/store/chain_store.rs | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 8b7c40d2fbaf..bfcd0189935f 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -630,6 +630,8 @@ where left.flush() .await .map_err(|e| Error::Other(e.to_string()))?; + // Note: this ZstdEncoder also needs to be closed, otherwise the resulting archive + // might end up being invalid. left.close() .await .map_err(|e| Error::Other(e.to_string()))?; @@ -640,10 +642,6 @@ where .flush() .await .map_err(|e| Error::Other(e.to_string()))?; - right - .close() - .await - .map_err(|e| Error::Other(e.to_string()))?; right.finalize().await } } From d771f1499e5fe26b0e3b613e045c9430acd162c3 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 08:31:07 +0200 Subject: [PATCH 07/15] fix --- src/chain/store/chain_store.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index bfcd0189935f..faf82f7c90e3 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -630,7 +630,7 @@ where left.flush() .await .map_err(|e| Error::Other(e.to_string()))?; - // Note: this ZstdEncoder also needs to be closed, otherwise the resulting archive + // Note: the ZstdEncoder also needs to be closed, otherwise the resulting archive // might end up being invalid. left.close() .await From 0fa29f97d9a8e266cc81710eedf66f12e213ef07 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 08:40:32 +0200 Subject: [PATCH 08/15] properly fix the test and impl --- src/chain/store/chain_store.rs | 6 ++++-- src/utils/io/writer_checksum.rs | 4 +++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index faf82f7c90e3..8b7c40d2fbaf 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -630,8 +630,6 @@ where left.flush() .await .map_err(|e| Error::Other(e.to_string()))?; - // Note: the ZstdEncoder also needs to be closed, otherwise the resulting archive - // might end up being invalid. left.close() .await .map_err(|e| Error::Other(e.to_string()))?; @@ -642,6 +640,10 @@ where .flush() .await .map_err(|e| Error::Other(e.to_string()))?; + right + .close() + .await + .map_err(|e| Error::Other(e.to_string()))?; right.finalize().await } } diff --git a/src/utils/io/writer_checksum.rs b/src/utils/io/writer_checksum.rs index 7aa96dc78b26..c9b83043e072 100644 --- a/src/utils/io/writer_checksum.rs +++ b/src/utils/io/writer_checksum.rs @@ -133,9 +133,11 @@ mod test { let mut bytes = [0; 1024]; OsRng.fill_bytes(&mut bytes); temp_file_writer.write_all(&bytes).await?; - temp_file_writer.flush().await?; } + temp_file_writer.flush().await?; + temp_file_writer.close().await?; + let checksum = temp_file_writer.finalize().await?; let file_hash = { From 95f8276ac6d95f75e3fbc3de325c51b08127494f Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 09:54:54 +0200 Subject: [PATCH 09/15] fix --- .github/workflows/forest.yml | 2 +- scripts/tests/calibnet_export_check.sh | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/.github/workflows/forest.yml b/.github/workflows/forest.yml index 4a5d4ddd6243..dba615fa1b33 100644 --- a/.github/workflows/forest.yml +++ b/.github/workflows/forest.yml @@ -225,4 +225,4 @@ jobs: run: | chmod +x ~/.cargo/bin/forest* - name: Snapshot export check - run: ./scripts/tests/calibnet_export_check.sh + run: diff --git a/scripts/tests/calibnet_export_check.sh b/scripts/tests/calibnet_export_check.sh index 5e7109f3479c..023e9f298b08 100755 --- a/scripts/tests/calibnet_export_check.sh +++ b/scripts/tests/calibnet_export_check.sh @@ -9,6 +9,9 @@ source "$(dirname "$0")/harness.sh" forest_init +echo "Cleaning up the initial snapshot" +rm -rf ./*.car.* + echo "Exporting zstd compressed snapshot" $FOREST_CLI_PATH snapshot export From 82b49da45d3c699abac1fc869e5e5dc5b3586672 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 10:11:39 +0200 Subject: [PATCH 10/15] Update scripts/tests/calibnet_export_check.sh Co-authored-by: David Himmelstrup --- scripts/tests/calibnet_export_check.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/tests/calibnet_export_check.sh b/scripts/tests/calibnet_export_check.sh index 023e9f298b08..69c6a6c874b4 100755 --- a/scripts/tests/calibnet_export_check.sh +++ b/scripts/tests/calibnet_export_check.sh @@ -16,7 +16,7 @@ echo "Exporting zstd compressed snapshot" $FOREST_CLI_PATH snapshot export echo "Testing snapshot validity" -zstd -t ./*.car.zst +zstd --test ./*.car.zst echo "Verifying snapshot checksum" sha256sum -c ./*.sha256sum From 4bcc41d4ba3aca50b37bbf5d35d67d825e26dfb9 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 10:11:46 +0200 Subject: [PATCH 11/15] Update scripts/tests/calibnet_export_check.sh Co-authored-by: David Himmelstrup --- scripts/tests/calibnet_export_check.sh | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/scripts/tests/calibnet_export_check.sh b/scripts/tests/calibnet_export_check.sh index 69c6a6c874b4..2cdf3dce02a9 100755 --- a/scripts/tests/calibnet_export_check.sh +++ b/scripts/tests/calibnet_export_check.sh @@ -19,4 +19,4 @@ echo "Testing snapshot validity" zstd --test ./*.car.zst echo "Verifying snapshot checksum" -sha256sum -c ./*.sha256sum +sha256sum --check ./*.sha256sum From 32be55ad901446d51250c7893f8d10f29e93176f Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 10:28:38 +0200 Subject: [PATCH 12/15] implement AsyncWrite for Either --- .github/workflows/forest.yml | 2 +- src/chain/store/chain_store.rs | 30 ++++++++++-------------------- src/utils/misc/either.rs | 32 +++++++++++++++++++++++++++++++- 3 files changed, 42 insertions(+), 22 deletions(-) diff --git a/.github/workflows/forest.yml b/.github/workflows/forest.yml index dba615fa1b33..4a5d4ddd6243 100644 --- a/.github/workflows/forest.yml +++ b/.github/workflows/forest.yml @@ -225,4 +225,4 @@ jobs: run: | chmod +x ~/.cargo/bin/forest* - name: Snapshot export check - run: + run: ./scripts/tests/calibnet_export_check.sh diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 8b7c40d2fbaf..c897ed5da0ef 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -625,27 +625,17 @@ where ); let mut writer = writer.lock().await; + writer + .flush() + .await + .map_err(|e| Error::Other(e.to_string()))?; + writer + .close() + .await + .map_err(|e| Error::Other(e.to_string()))?; let digest = match &mut *writer { - Either::Left(left) => { - left.flush() - .await - .map_err(|e| Error::Other(e.to_string()))?; - left.close() - .await - .map_err(|e| Error::Other(e.to_string()))?; - left.get_mut().finalize().await - } - Either::Right(right) => { - right - .flush() - .await - .map_err(|e| Error::Other(e.to_string()))?; - right - .close() - .await - .map_err(|e| Error::Other(e.to_string()))?; - right.finalize().await - } + Either::Left(left) => left.get_mut().finalize().await, + Either::Right(right) => right.finalize().await, } .map_err(|e| Error::Other(e.to_string()))?; diff --git a/src/utils/misc/either.rs b/src/utils/misc/either.rs index 06372a67e829..f50af9d69c91 100644 --- a/src/utils/misc/either.rs +++ b/src/utils/misc/either.rs @@ -1,12 +1,15 @@ // Copyright 2019-2023 ChainSafe Systems // SPDX-License-Identifier: Apache-2.0, MIT +use std::io::IoSlice; use std::{ pin::Pin, task::{Context, Poll}, }; -use futures::AsyncRead; +use futures::{AsyncRead, AsyncWrite}; +use futures_util::io::{Close, Flush, IntoSink, Write, WriteAll, WriteVectored}; +use futures_util::AsyncWriteExt; pub enum Either { Left(L), @@ -25,3 +28,30 @@ impl AsyncRead for Either { } } } + +impl AsyncWrite for Either { + fn poll_write( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + buf: &[u8], + ) -> Poll> { + match Pin::into_inner(self) { + Self::Left(left) => Pin::new(left).poll_write(cx, buf), + Self::Right(right) => Pin::new(right).poll_write(cx, buf), + } + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match Pin::into_inner(self) { + Self::Left(left) => Pin::new(left).poll_flush(cx), + Self::Right(right) => Pin::new(right).poll_flush(cx), + } + } + + fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + match Pin::into_inner(self) { + Self::Left(left) => Pin::new(left).poll_close(cx), + Self::Right(right) => Pin::new(right).poll_close(cx), + } + } +} From 7f186090aaea699cfb58cd3a52dc5bb6bd3b0a35 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 10:30:25 +0200 Subject: [PATCH 13/15] fmt --- src/chain/store/chain_store.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index c897ed5da0ef..df0752f16157 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -633,6 +633,7 @@ where .close() .await .map_err(|e| Error::Other(e.to_string()))?; + let digest = match &mut *writer { Either::Left(left) => left.get_mut().finalize().await, Either::Right(right) => right.finalize().await, From eb8636b973f68bb9be09c92a7b10b8d35a05f256 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 10:39:26 +0200 Subject: [PATCH 14/15] Update src/chain/store/chain_store.rs Co-authored-by: David Himmelstrup --- src/chain/store/chain_store.rs | 10 ++-------- 1 file changed, 2 insertions(+), 8 deletions(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index df0752f16157..5f991a0f0fe8 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -625,14 +625,8 @@ where ); let mut writer = writer.lock().await; - writer - .flush() - .await - .map_err(|e| Error::Other(e.to_string()))?; - writer - .close() - .await - .map_err(|e| Error::Other(e.to_string()))?; + writer.flush().await.context("failed to flush")?; + writer.close().await.context("failed to close")?; let digest = match &mut *writer { Either::Left(left) => left.get_mut().finalize().await, From d4213fb732bd5c219193af44603187a2628b56c2 Mon Sep 17 00:00:00 2001 From: Roman Useinov Date: Tue, 20 Jun 2023 10:40:07 +0200 Subject: [PATCH 15/15] add Context import --- src/chain/store/chain_store.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/chain/store/chain_store.rs b/src/chain/store/chain_store.rs index 5f991a0f0fe8..c00fbda67947 100644 --- a/src/chain/store/chain_store.rs +++ b/src/chain/store/chain_store.rs @@ -29,7 +29,7 @@ use crate::utils::{ misc::Either, }; use ahash::{HashMap, HashMapExt, HashSet}; -use anyhow::Result; +use anyhow::{Context, Result}; use async_compression::futures::write::ZstdEncoder; use bls_signatures::Serialize as SerializeBls; use cid::Cid;