Repository: p2panda/aquadoggo (language: rust)

Issue:
Node crashes when receiving blob
Steps to reproduce:

1. Start two nodes
2. Send a blob to one of them
3. Restart one node

```
[2023-09-12T08:46:11Z INFO  aquadoggo::manager] Start materializer service
[2023-09-12T08:46:11Z INFO  aquadoggo::materializer::worker] Register reduce worker with pool size 16
[2023-09-12T08:46:11Z INFO  aquadoggo::materializer::worker] Register dependency worker with pool size 16
[2023-09-12T08:46:11Z INFO  aquadoggo::materializer::worker] Register schema worker with pool size 16
[2023-09-12T08:46:11Z INFO  aquadoggo::materializer::worker] Register blob worker with pool size 16
[2023-09-12T08:46:11Z INFO  aquadoggo::materializer::worker] Register garbage_collection worker with pool size 16
[2023-09-12T08:46:11Z DEBUG aquadoggo::materializer::service] Dispatch 0 pending tasks from last runtime
[2023-09-12T08:46:11Z DEBUG aquadoggo::materializer::service] Materialiser service is ready
[2023-09-12T08:46:11Z INFO  aquadoggo::manager] Start http service
[2023-09-12T08:46:11Z DEBUG aquadoggo::graphql::schema] Subscribing GraphQL manager to schema provider
[2023-09-12T08:46:11Z DEBUG aquadoggo::graphql::schema] Finished building initial GraphQL schema
HTTP port 2020 was already taken, try random port instead ..
Go to http://0.0.0.0:35569/graphql to use GraphQL playground
[2023-09-12T08:46:11Z DEBUG aquadoggo::http::service] HTTP service is ready
[2023-09-12T08:46:11Z INFO  aquadoggo::manager] Start network service
Peer id: 12D3KooWDPDvhGBK249kVkak3BGmfXuQvo6VSDErqc2BH2gqGFHG
[2023-09-12T08:46:11Z INFO  aquadoggo::network::service] Networking service initializing...
[2023-09-12T08:46:11Z DEBUG aquadoggo::network::behaviour] mDNS network behaviour enabled
[2023-09-12T08:46:11Z DEBUG aquadoggo::network::behaviour] Relay client network behaviour enabled
QUIC port 2022 was already taken, try random port instead ..
[2023-09-12T08:46:11Z INFO  aquadoggo::network::service] Network service ready!
[2023-09-12T08:46:11Z INFO  aquadoggo::manager] Start replication service
Node is listening on 0.0.0.0:47501
[2023-09-12T08:46:11Z DEBUG aquadoggo::network::service] mDNS discovered a new peer: 12D3KooWApRV2vny2MQb6Psbc5JxHw8ynJe3tXHfBVPNNcDRieMq
[2023-09-12T08:46:11Z INFO  aquadoggo::replication::service] Established connection with peer: 12D3KooWApRV2vny2MQb6Psbc5JxHw8ynJe3tXHfBVPNNcDRieMq (1)
[2023-09-12T08:46:11Z INFO  aquadoggo::replication::service] Established connection with peer: 12D3KooWApRV2vny2MQb6Psbc5JxHw8ynJe3tXHfBVPNNcDRieMq (2)
[2023-09-12T08:46:16Z INFO  aquadoggo::replication::manager] Accept inbound replication session with peer 12D3KooWApRV2vny2MQb6Psbc5JxHw8ynJe3tXHfBVPNNcDRieMq (1)
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer reduce task with input <TaskInput 0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51/-> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Working on <TaskInput 0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51/->
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer reduce task with input <TaskInput 00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce/-> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Working on <TaskInput 00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce/->
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Created <Document 16bd51>
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Dispatch dependency task for view with id: 0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer dependency task with input <TaskInput -/0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Working on <TaskInput -/0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51>
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer reduce task with input <TaskInput 002083088e6d6d9f94f86d54c836bc8343a8305ad3698a3ec6ebcd3a0895aa6b7a33/-> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Working on <TaskInput 002083088e6d6d9f94f86d54c836bc8343a8305ad3698a3ec6ebcd3a0895aa6b7a33/->
[2023-09-12T08:46:16Z INFO  aquadoggo::replication::service] Finished replication with peer 12D3KooWApRV2vny2MQb6Psbc5JxHw8ynJe3tXHfBVPNNcDRieMq (1)
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer reduce task with input <TaskInput 0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4/-> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Working on <TaskInput 0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4/->
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Created <Document 20abd4>
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Dispatch dependency task for view with id: 0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer dependency task with input <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Working on <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4>
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Get view for pinned relation with id: 0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Created <Document 6b7a33>
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Dispatch dependency task for view with id: 002083088e6d6d9f94f86d54c836bc8343a8305ad3698a3ec6ebcd3a0895aa6b7a33
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer dependency task with input <TaskInput -/002083088e6d6d9f94f86d54c836bc8343a8305ad3698a3ec6ebcd3a0895aa6b7a33> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Working on <TaskInput -/002083088e6d6d9f94f86d54c836bc8343a8305ad3698a3ec6ebcd3a0895aa6b7a33>
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Scheduling 2 tasks
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer blob task with input <TaskInput -/002083088e6d6d9f94f86d54c836bc8343a8305ad3698a3ec6ebcd3a0895aa6b7a33> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Duplicate materializer dependency task already in progress, setting re-queue flag for task with input <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4> and not adding this task to the queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::blob] Working on <TaskInput -/002083088e6d6d9f94f86d54c836bc8343a8305ad3698a3ec6ebcd3a0895aa6b7a33>
[2023-09-12T08:46:16Z INFO  aquadoggo::materializer::tasks::blob] Creating blob at path /tmp/.tmpJA3c6E/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Created <Document d457ce>
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::reduce] Dispatch dependency task for view with id: 00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer dependency task with input <TaskInput -/00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Working on <TaskInput -/00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce>
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Scheduling 2 tasks
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer blob task with input <TaskInput -/0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Materializer dependency task with input <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4> not sent to queue as a task for this document has already been re-queued.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::blob] Working on <TaskInput -/0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51>
[2023-09-12T08:46:16Z INFO  aquadoggo::materializer::tasks::blob] Creating blob at path /tmp/.tmpJA3c6E/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] View found for pinned relation: 0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Get view for pinned relation with id: 00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Scheduling 2 tasks
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Materializer dependency task with input <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4> not sent to queue as a task for this document has already been re-queued.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer blob task with input <TaskInput -/00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::blob] Working on <TaskInput -/00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce>
[2023-09-12T08:46:16Z INFO  aquadoggo::materializer::tasks::blob] Creating blob at path /tmp/.tmpJA3c6E/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4
[2023-09-12T08:46:16Z ERROR aquadoggo::materializer::worker] Critical error in worker blob with task <QueueItem 0 w. <TaskInput -/002083088e6d6d9f94f86d54c836bc8343a8305ad3698a3ec6ebcd3a0895aa6b7a33>>: Could not write blob file @ /tmp/.tmpJA3c6E/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4: Some pieces missing for the requested blob
[2023-09-12T08:46:16Z INFO  aquadoggo::manager] Received shutdown signal
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] View found for pinned relation: 00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Get view for pinned relation with id: 002083088e6d6d9f94f86d54c836bc8343a8305ad3698a3ec6ebcd3a0895aa6b7a33
[2023-09-12T08:46:16Z INFO  aquadoggo::replication::manager] Initiate outbound replication session with peer 12D3KooWApRV2vny2MQb6Psbc5JxHw8ynJe3tXHfBVPNNcDRieMq (1)
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] View found for pinned relation: 002083088e6d6d9f94f86d54c836bc8343a8305ad3698a3ec6ebcd3a0895aa6b7a33
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Scheduling 1 tasks
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer blob task with input <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::blob] Working on <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4>
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::worker] Sending materializer dependency task with input <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4> to the task queue.
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Working on <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4>
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Get view for pinned relation with id: 0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51
[2023-09-12T08:46:16Z INFO  aquadoggo::materializer::tasks::blob] Creating blob at path /tmp/.tmpJA3c6E/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4
[2023-09-12T08:46:16Z INFO  aquadoggo::replication::service] Closed connection with peer: 12D3KooWApRV2vny2MQb6Psbc5JxHw8ynJe3tXHfBVPNNcDRieMq (2)
[2023-09-12T08:46:16Z INFO  aquadoggo::replication::service] Closed connection with peer: 12D3KooWApRV2vny2MQb6Psbc5JxHw8ynJe3tXHfBVPNNcDRieMq (1)
[2023-09-12T08:46:16Z ERROR aquadoggo::materializer::worker] Critical error in worker blob with task <QueueItem 1 w. <TaskInput -/0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51>>: Could not write blob file @ /tmp/.tmpJA3c6E/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4: background task failed
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] View found for pinned relation: 0020725a1aeb32236b41e3d24e89f9ec1ec253ba588a2d350d40e47629b23616bd51
[2023-09-12T08:46:16Z DEBUG aquadoggo::materializer::tasks::dependency] Get view for pinned relation with id: 00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce
[2023-09-12T08:46:16Z ERROR aquadoggo::materializer::worker] Critical error in worker dependency with task <QueueItem 4 w. <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4>>: A fatal error occured in DocumentStore: attempted to acquire a connection on a closed pool
[2023-09-12T08:46:16Z ERROR aquadoggo::materializer::worker] Critical error in worker blob with task <QueueItem 2 w. <TaskInput -/00207d38b972a8eb1407b169bf4388a77763bc7b19428f6fecef9fdcd46970d457ce>>: Could not write blob file @ /tmp/.tmpJA3c6E/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4: background task failed
[2023-09-12T08:46:16Z ERROR aquadoggo::materializer::worker] Critical error in worker blob with task <QueueItem 3 w. <TaskInput -/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4>>: Could not write blob file @ /tmp/.tmpJA3c6E/0020f56b7856b3e383fd969ffd47375ab43d14cad30ebaa6cf5340cc1500ee20abd4: background task failed
```


Discussion / hints:

Fix patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9be30737a..b9bd3e910 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -27,6 +27,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Make sure temporary directory does not run out of scope [#557](https://github.com/p2panda/aquadoggo/pull/557)
 - Deduplicate generated schema field by key in proptests [#558](https://github.com/p2panda/aquadoggo/pull/558)
+- Do not panic when blob does not have all pieces yet [#563](https://github.com/p2panda/aquadoggo/pull/563)
 
 ## [0.5.0]
 
diff --git a/Cargo.lock b/Cargo.lock
index 30715f103..9d53d3b9b 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3118,7 +3118,7 @@ checksum = "04744f49eae99ab78e0d5c0b603ab218f515ea8cfe5a456d7629ad883a3b6e7d"
 [[package]]
 name = "p2panda-rs"
 version = "0.7.1"
-source = "git+https://github.com/p2panda/p2panda?rev=3ee5b0e543ccb24d548e70143b3adf6f1d5e4d01#3ee5b0e543ccb24d548e70143b3adf6f1d5e4d01"
+source = "git+https://github.com/p2panda/p2panda?rev=c0b382c9b31425748bba7c150ec9ad7a784d000c#c0b382c9b31425748bba7c150ec9ad7a784d000c"
 dependencies = [
  "arrayvec 0.5.2",
  "async-trait",
diff --git a/aquadoggo/Cargo.toml b/aquadoggo/Cargo.toml
index ba48a9dbd..d53d289df 100644
--- a/aquadoggo/Cargo.toml
+++ b/aquadoggo/Cargo.toml
@@ -57,7 +57,7 @@ lipmaa-link = "0.2.2"
 log = "0.4.19"
 once_cell = "1.18.0"
 openssl-probe = "0.1.5"
-p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "3ee5b0e543ccb24d548e70143b3adf6f1d5e4d01", features = [
+p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "c0b382c9b31425748bba7c150ec9ad7a784d000c", features = [
     "storage-provider",
 ] }
 rand = "0.8.5"
@@ -98,7 +98,7 @@ http = "0.2.9"
 hyper = "0.14.19"
 libp2p-swarm-test = "0.2.0"
 once_cell = "1.17.0"
-p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "3ee5b0e543ccb24d548e70143b3adf6f1d5e4d01", features = [
+p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "c0b382c9b31425748bba7c150ec9ad7a784d000c", features = [
     "test-utils",
     "storage-provider",
 ] }
diff --git a/aquadoggo/src/db/errors.rs b/aquadoggo/src/db/errors.rs
index d76f6b061..d03d28046 100644
--- a/aquadoggo/src/db/errors.rs
+++ b/aquadoggo/src/db/errors.rs
@@ -53,11 +53,7 @@ pub enum BlobStoreError {
     #[error("Missing \"pieces\" field on blob document")]
     NotBlobDocument,
 
-    /// Error when no pieces found for existing blob document.
-    #[error("No pieces found for the requested blob")]
-    NoBlobPiecesFound,
-
-    /// Error when some pieces not found for existing blob document.
+    /// Error when some or all pieces not found for existing blob document.
     #[error("Some pieces missing for the requested blob")]
     MissingPieces,
 
diff --git a/aquadoggo/src/db/stores/blob.rs b/aquadoggo/src/db/stores/blob.rs
index 2b7ad75c4..f3b4c8fbc 100644
--- a/aquadoggo/src/db/stores/blob.rs
+++ b/aquadoggo/src/db/stores/blob.rs
@@ -113,11 +113,6 @@ impl BlobStream {
     /// This method is called _after_ the stream has ended. We compare the values with what we've
     /// expected and find inconsistencies and invalid blobs.
     fn validate(&self) -> Result<(), BlobStoreError> {
-        // No pieces were found
-        if self.length == 0 {
-            return Err(BlobStoreError::NoBlobPiecesFound);
-        };
-
         // Not all pieces were found
         if self.expected_num_pieces != self.num_pieces {
             return Err(BlobStoreError::MissingPieces);
@@ -160,6 +155,7 @@ impl SqlStore {
     /// Get data stream for one blob from the store, identified by it's document id.
     pub async fn get_blob(&self, id: &DocumentId) -> Result<Option<BlobStream>, BlobStoreError> {
         if let Some(document) = self.get_document(id).await? {
+            let document = validate_blob_pieces(self, document).await?;
             Ok(Some(BlobStream::new(self, document)?))
         } else {
             Ok(None)
@@ -172,6 +168,7 @@ impl SqlStore {
         view_id: &DocumentViewId,
     ) -> Result<Option<BlobStream>, BlobStoreError> {
         if let Some(document) = self.get_document_by_view_id(view_id).await? {
+            let document = validate_blob_pieces(self, document).await?;
             Ok(Some(BlobStream::new(self, document)?))
         } else {
             Ok(None)
@@ -234,6 +231,43 @@ impl SqlStore {
     }
 }
 
+/// Throws an error when database does not contain all related blob pieces yet.
+async fn validate_blob_pieces(
+    store: &SqlStore,
+    document: impl AsDocument,
+) -> Result<impl AsDocument, BlobStoreError> {
+    let schema = Schema::get_system(SchemaId::BlobPiece(1)).expect("System schema is given");
+    let list = RelationList::new_pinned(document.view_id(), "pieces");
+
+    let args = Query::new(
+        &Pagination::new(
+            &NonZeroU64::new(BLOB_QUERY_PAGE_SIZE).unwrap(),
+            None,
+            &vec![PaginationField::TotalCount],
+        ),
+        &Select::default(),
+        &Filter::default(),
+        &Order::default(),
+    );
+
+    let (pagination_data, _) = store.query(schema, &args, Some(&list)).await?;
+    let total_count = pagination_data
+        .total_count
+        .expect("total count is selected");
+
+    // Get the number of pieces in the blob
+    let expected_num_pieces = match document.get("pieces").unwrap() {
+        OperationValue::PinnedRelationList(list) => list.len(),
+        _ => unreachable!(), // We already validated that this is a blob document
+    };
+
+    if total_count as usize == expected_num_pieces {
+        Ok(document)
+    } else {
+        Err(BlobStoreError::MissingPieces)
+    }
+}
+
 /// Helper for getting the document ids of any document which relates to the specified document.
 ///
 /// Optionally pass in a `SchemaId` to restrict the results to documents of a certain schema.
@@ -370,17 +404,8 @@ mod tests {
             let blob_document_id: DocumentId = blob_view_id.to_string().parse().unwrap();
 
             // We get the correct `NoBlobPiecesFound` error.
-            let stream = node
-                .context
-                .store
-                .get_blob(&blob_document_id)
-                .await
-                .unwrap();
-            let collected_data = read_data_from_stream(stream.unwrap()).await;
-            assert!(matches!(
-                collected_data,
-                Err(BlobStoreError::NoBlobPiecesFound)
-            ),);
+            let stream = node.context.store.get_blob(&blob_document_id).await;
+            assert!(matches!(stream, Err(BlobStoreError::MissingPieces)));
 
             // Publish one blob piece.
             let blob_piece_view_id_1 = add_document(
@@ -410,14 +435,8 @@ mod tests {
             let blob_document_id: DocumentId = blob_view_id.to_string().parse().unwrap();
 
             // We should get the correct `MissingBlobPieces` error.
-            let stream = node
-                .context
-                .store
-                .get_blob(&blob_document_id)
-                .await
-                .unwrap();
-            let collected_data = read_data_from_stream(stream.unwrap()).await;
-            assert!(matches!(collected_data, Err(BlobStoreError::MissingPieces)),);
+            let stream = node.context.store.get_blob(&blob_document_id).await;
+            assert!(matches!(stream, Err(BlobStoreError::MissingPieces)));
 
             // Publish one more blob piece, but it doesn't contain the correct number of bytes.
             let blob_piece_view_id_2 = add_document(
diff --git a/aquadoggo/src/materializer/tasks/blob.rs b/aquadoggo/src/materializer/tasks/blob.rs
index b9c1cc28f..01c068744 100644
--- a/aquadoggo/src/materializer/tasks/blob.rs
+++ b/aquadoggo/src/materializer/tasks/blob.rs
@@ -1,6 +1,5 @@
 // SPDX-License-Identifier: AGPL-3.0-or-later
 
-use anyhow::anyhow;
 use futures::{pin_mut, StreamExt};
 use log::{debug, info};
 use p2panda_rs::document::traits::AsDocument;
@@ -48,7 +47,8 @@ pub async fn blob_task(context: Context, input: TaskInput) -> TaskResult<TaskInp
             Ok(vec![document])
         }
 
-        // This task is about an updated blob piece document that may be used in one or more blob documents.
+        // This task is about an updated blob piece document that may be used in one or more blob
+        // documents.
         SchemaId::BlobPiece(_) => get_related_blobs(&input_view_id, &context).await,
         _ => Err(TaskError::Critical(format!(
             "Unknown system schema id: {}",
@@ -71,7 +71,7 @@ pub async fn blob_task(context: Context, input: TaskInput) -> TaskResult<TaskInp
             .get_blob_by_view_id(blob_document.view_id())
             .await
             // We don't raise a critical error here, as it is possible that this method returns an
-            // error
+            // error, for example when not all blob pieces are available yet for materialisation
             .map_err(|err| TaskError::Failure(err.to_string()))?
             .expect("Blob data exists at this point");
 
@@ -104,16 +104,18 @@ pub async fn blob_task(context: Context, input: TaskInput) -> TaskResult<TaskInp
 
         while let Some(value) = stream.next().await {
             match value {
-                Ok(buf) => file.write(&buf).await.map_err(|err| anyhow!(err)),
-                Err(err) => Err(anyhow!(err)),
-            }
-            .map_err(|err| {
-                TaskError::Critical(format!(
-                    "Could not write blob file @ {}: {}",
-                    blob_view_path.display(),
+                Ok(buf) => file.write(&buf).await.map_err(|err| {
+                    TaskError::Critical(format!(
+                        "Error occurred when writing to blob file @ {}: {}",
+                        blob_view_path.display(),
+                        err
+                    ))
+                }),
+                Err(err) => Err(TaskError::Failure(format!(
+                    "Blob data is invalid and can not be materialised: {}",
                     err
-                ))
-            })?;
+                ))),
+            }?;
         }
     }
 
diff --git a/aquadoggo_cli/Cargo.toml b/aquadoggo_cli/Cargo.toml
index 3fb0304e8..8962a80bb 100644
--- a/aquadoggo_cli/Cargo.toml
+++ b/aquadoggo_cli/Cargo.toml
@@ -29,7 +29,7 @@ figment = { version = "0.10.10", features = ["toml", "env"] }
 hex = "0.4.3"
 libp2p = "0.52.0"
 log = "0.4.20"
-p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "3ee5b0e543ccb24d548e70143b3adf6f1d5e4d01" }
+p2panda-rs = { git = "https://github.com/p2panda/p2panda", rev = "c0b382c9b31425748bba7c150ec9ad7a784d000c" }
 path-clean = "1.0.1"
 serde = { version = "1.0.185", features = ["serde_derive"] }
 tempfile = "3.7.0"


Test patch: