chore: Diagnose early EOF stream

subconsciousnetwork · Oct 23, 2023 · 1adba02 · 1adba02
1 parent f7dd38d
commit 1adba02
Show file tree

Hide file tree

Showing 5 changed files with 52 additions and 29 deletions.
diff --git a/.github/workflows/run_test_suite.yaml b/.github/workflows/run_test_suite.yaml
@@ -37,6 +37,25 @@ jobs:
           swift build --sanitize=address
           swift test --sanitize=address
 
+  run-linting-linux:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - uses: Swatinem/rust-cache@v2
+      - name: 'Setup Rust'
+        run: |
+          curl -sSf https://sh.rustup.rs | sh -s -- -y
+          rustup component add clippy
+          rustup component add rustfmt
+      - name: 'Install environment packages'
+        run: |
+          sudo apt-get update -qqy
+          sudo apt-get install jq protobuf-compiler cmake
+      - name: 'Check Format'
+        run: cargo fmt --all -- --check
+      - name: 'Run Linter'
+        run: cargo clippy --all -- -D warnings
+
   run-test-suite-windows:
     runs-on: windows-latest
     steps:
@@ -49,34 +68,20 @@ jobs:
         run: |
           choco install -y cmake protoc openssl
         shell: sh
+      - name: Install cargo-binstall
+        uses: cargo-bins/cargo-binstall@v1.4.4
+      - name: Install binaries from cargo
+        run: |
+          cargo binstall cargo-nextest --no-confirm
       - name: 'Install IPFS Kubo'
         uses: ibnesayeed/setup-ipfs@master
         with:
           ipfs_version: v0.17.0
           run_daemon: true
       - name: 'Run Rust native target tests'
-        run: cargo test --features test-kubo,helpers
+        run: cargo nextest run --features test-kubo,headers --retries 5
         env:
-          NOOSPHERE_LOG: deafening
-
-  run-linting-linux:
-    runs-on: ubuntu-latest
-    steps:
-      - uses: actions/checkout@v3
-      - uses: Swatinem/rust-cache@v2
-      - name: 'Setup Rust'
-        run: |
-          curl -sSf https://sh.rustup.rs | sh -s -- -y
-          rustup component add clippy
-          rustup component add rustfmt
-      - name: 'Install environment packages'
-        run: |
-          sudo apt-get update -qqy
-          sudo apt-get install jq protobuf-compiler cmake
-      - name: 'Check Format'
-        run: cargo fmt --all -- --check
-      - name: 'Run Linter'
-        run: cargo clippy --all -- -D warnings
+          NOOSPHERE_LOG: academic
 
   run-test-suite-linux:
     runs-on: ubuntu-latest
@@ -90,13 +95,20 @@ jobs:
         run: |
           sudo apt-get update -qqy
           sudo apt-get install jq protobuf-compiler cmake
+      - name: Install cargo-binstall
+        uses: cargo-bins/cargo-binstall@v1.4.4
+      - name: Install binaries from cargo
+        run: |
+          cargo binstall cargo-nextest --no-confirm
       - name: 'Install IPFS Kubo'
         uses: ibnesayeed/setup-ipfs@master
         with:
           ipfs_version: v0.17.0
           run_daemon: true
       - name: 'Run Rust native target tests'
-        run: NOOSPHERE_LOG=deafening cargo test --features test-kubo,headers
+        run: cargo nextest run --features test-kubo,headers --retries 5
+        env:
+          NOOSPHERE_LOG: academic
 
   run-test-suite-linux-rocksdb:
     runs-on: ubuntu-latest
@@ -116,7 +128,7 @@ jobs:
           ipfs_version: v0.17.0
           run_daemon: true
       - name: 'Run Rust native target tests (RocksDB)'
-        run: NOOSPHERE_LOG=defeaning cargo test -p noosphere -p noosphere-storage --features rocksdb,test-kubo
+        run: NOOSPHERE_LOG=academic cargo test -p noosphere -p noosphere-storage --features rocksdb,test-kubo
 
   run-test-suite-linux-c:
     runs-on: ubuntu-latest

diff --git a/rust/noosphere-core/src/stream/memo.rs b/rust/noosphere-core/src/stream/memo.rs
@@ -82,7 +82,7 @@ where
                         };
 
                         if replicate_authority {
-                            debug!("Replicating authority...");
+                            trace!("Replicating authority...");
                             let authority = sphere.get_authority().await?;
                             let store = store.clone();
 
@@ -105,7 +105,7 @@ where
                         }
 
                         if replicate_address_book {
-                            debug!("Replicating address book...");
+                            trace!("Replicating address book...");
                             let address_book = sphere.get_address_book().await?;
                             let identities = address_book.get_identities().await?;
 
@@ -121,7 +121,7 @@ where
                         }
 
                         if replicate_content {
-                            debug!("Replicating content...");
+                            trace!("Replicating content...");
                             let content = sphere.get_content().await?;
 
                             tasks.spawn(walk_versioned_map_changes_and(content, store.clone(), move |_, link, store| async move {

diff --git a/rust/noosphere-core/src/view/content.rs b/rust/noosphere-core/src/view/content.rs
@@ -17,7 +17,7 @@ impl<'a, 'b, S: BlockStore> BodyChunkDecoder<'a, 'b, S> {
         let store = self.1.clone();
         Box::pin(try_stream! {
             while let Some(cid) = next {
-                debug!("Unpacking block {}...", cid);
+                trace!("Unpacking block {}...", cid);
                 let chunk = store.load::<DagCborCodec, BodyChunkIpld>(&cid).await.map_err(|error| {
                     std::io::Error::new(std::io::ErrorKind::UnexpectedEof, error.to_string())
                 })?;

diff --git a/rust/noosphere-gateway/src/handlers/v0alpha2/push.rs b/rust/noosphere-gateway/src/handlers/v0alpha2/push.rs
@@ -121,8 +121,16 @@ where
             })?;
 
             for await block in new_blocks {
-                yield block?;
+                match block {
+                    Ok(block) => yield block,
+                    Err(error) => {
+                        warn!("Failed stream final gateway blocks: {}", error);
+                        Err(error)?;
+                    }
+                }
             }
+
+            info!("Finished gateway push routine!");
         };
 
         Ok(to_car_stream(roots, block_stream))
@@ -216,7 +224,7 @@ where
 
             for step in history.into_iter().rev() {
                 let (cid, sphere) = step?;
-                debug!("Hydrating {}", cid);
+                trace!("Hydrating {}", cid);
                 sphere.hydrate().await?;
             }
 
@@ -337,6 +345,7 @@ where
 
     /// Notify the name system that new names may need to be resolved
     async fn notify_name_resolver(&self, push_body: &PushBody) -> Result<()> {
+        debug!("Notifying name system of new link record...");
         if let Some(name_record) = &push_body.name_record {
             if let Err(error) = self.name_system_tx.send(NameSystemJob::Publish {
                 context: self.sphere_context.clone(),
@@ -359,6 +368,7 @@ where
 
     /// Request that new history be syndicated to IPFS
     async fn notify_ipfs_syndicator(&self, next_version: Link<MemoIpld>) -> Result<()> {
+        debug!("Notifying syndication worker of new blocks...");
         // TODO(#156): This should not be happening on every push, but rather on
         // an explicit publish action. Move this to the publish handler when we
         // have added it to the gateway.

diff --git a/rust/noosphere/tests/distributed_stress.rs b/rust/noosphere/tests/distributed_stress.rs
@@ -150,6 +150,7 @@ mod multiplayer {
     use serde_json::Value;
     use url::Url;
 
+    #[cfg(not(feature = "rocksdb"))]
     #[tokio::test(flavor = "multi_thread")]
     async fn orb_can_render_peers_in_the_sphere_address_book() -> Result<()> {
         initialize_tracing(None);