fix: P2P: block merkle roots not validated before inserting into archive DB

lib/archive.rs

@@ -730,6 +730,36 @@ impl Archive {
             })
     }
 
+    /// Delete a stored body, reversing the effects of [`Self::put_body`].
+    ///
+    /// Used to discard a body that was stored before its contents could be
+    /// validated (e.g. a body received from a peer) and later turned out to be
+    /// invalid, so that the block is reported missing again by
+    /// [`Self::iter_missing_bodies`] and the real body is re-requested.
+    pub fn delete_body(
+        &self,
+        rwtxn: &mut RwTxn,
+        block_hash: BlockHash,
+        body: &Body,
+    ) -> Result<(), Error> {
+        self.bodies
+            .delete(rwtxn, &block_hash)
+            .map_err(DbError::from)?;
+        body.transactions.iter().try_for_each(|tx| {
+            let txid = tx.txid();
+            let mut inclusions = self.get_tx_inclusions(rwtxn, txid)?;
+            inclusions.remove(&block_hash);
+            if inclusions.is_empty() {
+                self.txid_to_inclusions
+                    .delete(rwtxn, &txid)
+                    .map_err(DbError::from)?;
+            } else {
+                self.txid_to_inclusions.put(rwtxn, &txid, &inclusions)?;
+            }
+            Ok(())
+        })
+    }
+
     /// Store a header.
     ///
     /// The following predicates MUST be met before calling this function:

lib/node/net_task.rs

@@ -381,15 +381,26 @@ fn reorg_to_tip(
             }
             two_way_peg_data
         };
-        let () = connect_tip_(
+        if let Err(err) = connect_tip_(
             &mut rwtxn,
             archive,
             mempool,
             state,
             &header,
             &body,
             &two_way_peg_data,
-        )?;
+        ) {
+            // The stored body for this block failed validation (e.g. a peer
+            // supplied a body whose contents do not match the header's merkle
+            // root). Abort the reorg and discard the invalid body from the
+            // archive so that the block is reported missing again and the real
+            // body is re-requested, instead of the archive staying poisoned.
+            drop(rwtxn);
+            let mut rwtxn = env.write_txn().map_err(EnvError::from)?;
+            let () = archive.delete_body(&mut rwtxn, header.hash(), &body)?;
+            rwtxn.commit().map_err(RwTxnError::from)?;
+            return Err(err);
+        }
         let new_tip_hash = state.try_get_tip(&rwtxn)?.unwrap();
         let bmm_verification =
             archive.get_best_main_verification(&rwtxn, new_tip_hash)?;
@@ -997,16 +1008,35 @@ impl NetTask {
                         }
                     }
                 }
-                MailboxItem::NewTipReady(new_tip, _addr, resp_tx) => {
-                    let reorg_applied = task::block_in_place(|| {
+                MailboxItem::NewTipReady(new_tip, addr, resp_tx) => {
+                    let reorg_result = task::block_in_place(|| {
                         reorg_to_tip(
                             &self.ctxt.env,
                             &self.ctxt.archive,
                             &self.ctxt.mempool,
                             &self.ctxt.state,
                             new_tip,
                         )
-                    })?;
+                    });
+                    // A tip supplied by a peer may carry a body that fails
+                    // validation. Do not let that error propagate out of `run`
+                    // and halt the whole net task: log it and drop the
+                    // offending peer instead. `reorg_to_tip` has already
+                    // discarded the invalid body so it will be re-requested.
+                    let reorg_applied = match reorg_result {
+                        Ok(reorg_applied) => reorg_applied,
+                        Err(err) => {
+                            tracing::warn!(
+                                ?addr,
+                                %err,
+                                "Failed to reorg to new tip; dropping peer"
+                            );
+                            if let Some(addr) = addr {
+                                let () = self.ctxt.net.remove_active_peer(addr);
+                            }
+                            false
+                        }
+                    };
                     if let Some(resp_tx) = resp_tx {
                         let () = resp_tx
                             .send(reorg_applied)