A0-0000: Don't import unimportable blocks (#1959) (#1960)

# Description

Our sync tried importing everything that was required. This was not a
major issue before 15.1, since at worst the import failed and whatever,
but with optimistic imports we gotta be relatively confident that the
import will actually work.

## Type of change

- Bug fix (non-breaking change which fixes an issue)

# Checklist:

- I have added tests
- I have made corresponding changes to the existing documentation
- I have created new documentation

Author: timorleph

bin/node/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "aleph-node"
-version = "15.1.0"
+version = "15.2.0"
 description = "Aleph node binary"
 build = "build.rs"
 license = "GPL-3.0-or-later"

finality-aleph/src/sync/forest/mod.rs

@@ -608,11 +608,29 @@ where
         }
     }
 
-    pub fn start_import(&mut self, id: &BlockId) {
+    /// Start an import of a block. Returns whether this is possible, i.e. the parent block is
+    /// imported.
+    pub fn start_import(&mut self, id: &BlockId) -> bool {
+        let parent_id = match self.get(id) {
+            VertexHandle::Candidate(vertex) => vertex.vertex.parent(),
+            _ => return false,
+        };
+        let parent_id = match parent_id {
+            Some(parent_id) => parent_id,
+            None => return false,
+        };
+        // At this point this is equivalent to being imported, as we never get here for blocks that
+        // are below minimal.
+        if !self.skippable(&parent_id) {
+            return false;
+        }
         use VertexHandleMut::Candidate;
         if let Candidate(mut vertex) = self.get_mut(id) {
             vertex.get_mut().vertex.start_import();
+            return true;
         }
+        // This should never happen in practice.
+        false
     }
 
     fn know_most(&self, id: &BlockId) -> HashSet<I> {
@@ -1007,6 +1025,16 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn accepts_first_import() {
+        let (initial_header, mut forest) = setup();
+        let child = initial_header.random_child();
+        assert!(forest
+            .update_header(&child, None, true)
+            .expect("header was correct"));
+        assert!(forest.start_import(&child.id()));
+    }
+
     #[test]
     fn accepts_first_body() {
         let (initial_header, mut forest) = setup();
@@ -1019,6 +1047,17 @@ mod tests {
         assert_eq!(forest.extension_request(), Noop);
     }
 
+    #[test]
+    fn rejects_import_when_parent_unimported() {
+        let (initial_header, mut forest) = setup();
+        let child = initial_header.random_child();
+        let grandchild = child.random_child();
+        assert!(forest
+            .update_header(&child, None, false)
+            .expect("header was correct"));
+        assert!(!forest.start_import(&grandchild.id()));
+    }
+
     #[test]
     fn rejects_body_when_parent_unimported() {
         let (initial_header, mut forest) = setup();

finality-aleph/src/sync/handler/mod.rs

@@ -570,8 +570,15 @@ where
                             }),
                         );
                     }
-                    last_imported = Some(b.header().id());
-                    self.forest.start_import(&b.header().id());
+                    let block_id = b.header().id();
+                    if !self.forest.start_import(&block_id) {
+                        return (
+                            new_highest,
+                            equivocation_proofs,
+                            Some(Error::BlockNotImportable(block_id)),
+                        );
+                    }
+                    last_imported = Some(block_id);
                     self.block_importer.import_block(b, false);
                 }
             }
@@ -2566,6 +2573,20 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn rejects_parentless_block() {
+        let (mut handler, _backend, _notifier, genesis) = setup();
+
+        // Create 2 new blocks and make the handler aware of them.
+        let mut branch = grow_light_branch(&mut handler, &genesis, 2, 0);
+
+        // Give only the latter to handler.
+        let parentless_block = MockBlock::new(branch.pop().expect("just created"), true);
+        let (_, _, e) =
+            handler.handle_request_response(vec![ResponseItem::Block(parentless_block)], 1);
+        assert!(matches!(e, Some(Error::BlockNotImportable(_))));
+    }
+
     #[tokio::test]
     async fn handles_wide_fork_scenario() {
         let (mut handler, mut backend, mut notifier, genesis) = setup();