diff --git a/Cargo.lock b/Cargo.lock
index 9ced235d..d13a5a83 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -28,7 +28,7 @@ dependencies = [
 
 [[package]]
 name = "aleph-bft"
-version = "0.41.0"
+version = "0.42.0"
 dependencies = [
  "aleph-bft-mock",
  "aleph-bft-rmc",
diff --git a/README.md b/README.md
index 0f978745..11c68054 100644
--- a/README.md
+++ b/README.md
@@ -60,7 +60,7 @@ More details are available [in the book][reference-link-implementation-details].
 - Import AlephBFT in your crate
   ```toml
   [dependencies]
-  aleph-bft = "^0.41"
+  aleph-bft = "^0.42"
   ```
 - The main entry point is the `run_session` function, which returns a Future that runs the
   consensus algorithm.
diff --git a/consensus/Cargo.toml b/consensus/Cargo.toml
index 0ccbbaa7..e8122373 100644
--- a/consensus/Cargo.toml
+++ b/consensus/Cargo.toml
@@ -1,6 +1,6 @@
 [package]
 name = "aleph-bft"
-version = "0.41.0"
+version = "0.42.0"
 edition = "2021"
 authors = ["Cardinal Cryptography"]
 categories = ["algorithms", "data-structures", "cryptography", "database"]
diff --git a/consensus/src/creation/collector.rs b/consensus/src/creation/collector.rs
new file mode 100644
index 00000000..0d8d5100
--- /dev/null
+++ b/consensus/src/creation/collector.rs
@@ -0,0 +1,342 @@
+use crate::{units::Unit, Hasher, NodeCount, NodeIndex, NodeMap, Round};
+use anyhow::Result;
+use thiserror::Error;
+
+#[derive(Eq, Error, Debug, PartialEq)]
+pub enum ConstraintError {
+    #[error("Not enough parents.")]
+    NotEnoughParents,
+    #[error("Missing own parent.")]
+    MissingOwnParent,
+}
+
+#[derive(Clone)]
+pub struct UnitsCollector<H: Hasher> {
+    candidates: NodeMap<(H::Hash, Round)>,
+    for_round: Round,
+    direct_parents: NodeCount,
+}
+
+impl<H: Hasher> UnitsCollector<H> {
+    pub fn new_initial(n_members: NodeCount) -> Self {
+        UnitsCollector {
+            candidates: NodeMap::with_size(n_members),
+            for_round: 1,
+            direct_parents: NodeCount(0),
+        }
+    }
+
+    pub fn from_previous(previous: &UnitsCollector<H>) -> Self {
+        UnitsCollector {
+            candidates: previous.candidates.clone(),
+            for_round: previous.for_round + 1,
+            direct_parents: NodeCount(0),
+        }
+    }
+
+    pub fn add_unit<U: Unit<Hasher = H>>(&mut self, unit: &U) {
+        let node_id = unit.creator();
+        let hash = unit.hash();
+        let round = unit.round();
+
+        if round >= self.for_round {
+            return;
+        }
+
+        let to_insert = match self.candidates.get(node_id) {
+            None => Some((hash, round)),
+            Some((_, r)) if *r < round => Some((hash, round)),
+            _ => None,
+        };
+
+        if let Some(data) = to_insert {
+            self.candidates.insert(node_id, data);
+            if round == self.for_round - 1 {
+                self.direct_parents += NodeCount(1);
+            }
+        }
+    }
+
+    pub fn prospective_parents(
+        &self,
+        node_id: NodeIndex,
+    ) -> Result<&NodeMap<(H::Hash, Round)>, ConstraintError> {
+        if self.direct_parents < self.candidates.size().consensus_threshold() {
+            return Err(ConstraintError::NotEnoughParents);
+        }
+        match self.candidates.get(node_id) {
+            Some((_, r)) if *r == self.for_round - 1 => Ok(&self.candidates),
+            _ => Err(ConstraintError::MissingOwnParent),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::{
+        creation::collector::{ConstraintError, UnitsCollector},
+        units::{random_full_parent_units_up_to, Unit},
+        NodeCount, NodeIndex,
+    };
+    use aleph_bft_mock::Hasher64;
+
+    #[test]
+    fn initial_fails_without_parents() {
+        let n_members = NodeCount(4);
+        let units_collector = UnitsCollector::<Hasher64>::new_initial(n_members);
+
+        let err = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect_err("should fail without parents");
+        assert_eq!(err, ConstraintError::NotEnoughParents);
+    }
+
+    #[test]
+    fn initial_fails_with_too_few_parents() {
+        let n_members = NodeCount(4);
+        let mut units_collector = UnitsCollector::new_initial(n_members);
+        let units = random_full_parent_units_up_to(0, n_members, 43);
+        units_collector.add_unit(&units[0][0]);
+
+        let err = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect_err("should fail without parents");
+        assert_eq!(err, ConstraintError::NotEnoughParents);
+    }
+
+    #[test]
+    fn initial_fails_without_own_parent() {
+        let n_members = NodeCount(4);
+        let mut units_collector = UnitsCollector::new_initial(n_members);
+        let units = random_full_parent_units_up_to(0, n_members, 43);
+        for unit in units[0].iter().skip(1) {
+            units_collector.add_unit(unit);
+        }
+
+        let err = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect_err("should fail without parents");
+        assert_eq!(err, ConstraintError::MissingOwnParent);
+    }
+
+    #[test]
+    fn initial_successfully_computes_minimal_parents() {
+        let n_members = NodeCount(4);
+        let mut units_collector = UnitsCollector::new_initial(n_members);
+        let units = random_full_parent_units_up_to(0, n_members, 43);
+        for unit in units[0].iter().take(3) {
+            units_collector.add_unit(unit);
+        }
+
+        let parents = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect("we should be able to retrieve parents");
+        assert_eq!(parents.item_count(), 3);
+
+        let new_units: Vec<_> = units[0]
+            .iter()
+            .take(3)
+            .map(|unit| (unit.hash(), unit.round()))
+            .collect();
+        let selected_parents: Vec<_> = parents.values().cloned().collect();
+        assert_eq!(new_units, selected_parents);
+    }
+
+    #[test]
+    fn initial_successfully_computes_full_parents() {
+        let n_members = NodeCount(4);
+        let mut units_collector = UnitsCollector::new_initial(n_members);
+        let units = random_full_parent_units_up_to(0, n_members, 43);
+        for unit in &units[0] {
+            units_collector.add_unit(unit);
+        }
+
+        let parents = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect("we should be able to retrieve parents");
+        assert_eq!(parents.item_count(), 4);
+
+        let new_units: Vec<_> = units[0]
+            .iter()
+            .map(|unit| (unit.hash(), unit.round()))
+            .collect();
+        let selected_parents: Vec<_> = parents.values().cloned().collect();
+        assert_eq!(new_units, selected_parents);
+    }
+
+    #[test]
+    fn following_fails_without_parents() {
+        let n_members = NodeCount(4);
+        let initial_units_collector = UnitsCollector::<Hasher64>::new_initial(n_members);
+        let units_collector = UnitsCollector::from_previous(&initial_units_collector);
+
+        let err = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect_err("should fail without parents");
+        assert_eq!(err, ConstraintError::NotEnoughParents);
+    }
+
+    #[test]
+    fn following_fails_with_too_few_parents() {
+        let n_members = NodeCount(4);
+        let initial_units_collector = UnitsCollector::<Hasher64>::new_initial(n_members);
+        let mut units_collector = UnitsCollector::from_previous(&initial_units_collector);
+        let units = random_full_parent_units_up_to(1, n_members, 43);
+        units_collector.add_unit(&units[1][0]);
+
+        let err = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect_err("should fail without parents");
+        assert_eq!(err, ConstraintError::NotEnoughParents);
+    }
+
+    #[test]
+    fn following_fails_with_too_old_parents() {
+        let n_members = NodeCount(4);
+        let initial_units_collector = UnitsCollector::<Hasher64>::new_initial(n_members);
+        let mut units_collector = UnitsCollector::from_previous(&initial_units_collector);
+        let units = random_full_parent_units_up_to(0, n_members, 43);
+        for unit in &units[0] {
+            units_collector.add_unit(unit);
+        }
+
+        let err = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect_err("should fail without parents");
+        assert_eq!(err, ConstraintError::NotEnoughParents);
+    }
+
+    #[test]
+    fn following_fails_without_own_parent() {
+        let n_members = NodeCount(4);
+        let initial_units_collector = UnitsCollector::<Hasher64>::new_initial(n_members);
+        let mut units_collector = UnitsCollector::from_previous(&initial_units_collector);
+        let units = random_full_parent_units_up_to(1, n_members, 43);
+        for unit in units[1].iter().skip(1) {
+            units_collector.add_unit(unit);
+        }
+
+        let err = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect_err("should fail without parents");
+        assert_eq!(err, ConstraintError::MissingOwnParent);
+    }
+
+    #[test]
+    fn following_fails_with_too_old_own_parent() {
+        let n_members = NodeCount(4);
+        let initial_units_collector = UnitsCollector::<Hasher64>::new_initial(n_members);
+        let mut units_collector = UnitsCollector::from_previous(&initial_units_collector);
+        let units = random_full_parent_units_up_to(1, n_members, 43);
+        for unit in units[1].iter().skip(1) {
+            units_collector.add_unit(unit);
+        }
+        units_collector.add_unit(&units[0][0]);
+
+        let err = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect_err("should fail without parents");
+        assert_eq!(err, ConstraintError::MissingOwnParent);
+    }
+
+    #[test]
+    fn following_successfully_computes_minimal_parents() {
+        let n_members = NodeCount(4);
+        let initial_units_collector = UnitsCollector::<Hasher64>::new_initial(n_members);
+        let mut units_collector = UnitsCollector::from_previous(&initial_units_collector);
+        let units = random_full_parent_units_up_to(1, n_members, 43);
+        for unit in units[1].iter().take(3) {
+            units_collector.add_unit(unit);
+        }
+
+        let parents = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect("we should be able to retrieve parents");
+        assert_eq!(parents.item_count(), 3);
+
+        let new_units: Vec<_> = units[1]
+            .iter()
+            .take(3)
+            .map(|unit| (unit.hash(), unit.round()))
+            .collect();
+        let selected_parents: Vec<_> = parents.values().cloned().collect();
+        assert_eq!(new_units, selected_parents);
+    }
+
+    #[test]
+    fn following_successfully_computes_minimal_parents_with_ancient() {
+        let n_members = NodeCount(4);
+        let initial_units_collector = UnitsCollector::<Hasher64>::new_initial(n_members);
+        let mut units_collector = UnitsCollector::from_previous(&initial_units_collector);
+        let units = random_full_parent_units_up_to(1, n_members, 43);
+        for unit in units[1].iter().take(3) {
+            units_collector.add_unit(unit);
+        }
+        units_collector.add_unit(&units[0][3]);
+
+        let parents = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect("we should be able to retrieve parents");
+        assert_eq!(parents.item_count(), 4);
+
+        let mut new_units: Vec<_> = units[1]
+            .iter()
+            .take(3)
+            .map(|unit| (unit.hash(), unit.round()))
+            .collect();
+        new_units.push((units[0][3].hash(), units[0][3].round()));
+        let selected_parents: Vec<_> = parents.values().cloned().collect();
+        assert_eq!(new_units, selected_parents);
+    }
+
+    #[test]
+    fn following_successfully_computes_full_parents() {
+        let n_members = NodeCount(4);
+        let initial_units_collector = UnitsCollector::<Hasher64>::new_initial(n_members);
+        let mut units_collector = UnitsCollector::from_previous(&initial_units_collector);
+        let units = random_full_parent_units_up_to(1, n_members, 43);
+        for unit in &units[1] {
+            units_collector.add_unit(unit);
+        }
+
+        let parents = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect("we should be able to retrieve parents");
+        assert_eq!(parents.item_count(), 4);
+
+        let new_units: Vec<_> = units[1]
+            .iter()
+            .map(|unit| (unit.hash(), unit.round()))
+            .collect();
+        let selected_parents: Vec<_> = parents.values().cloned().collect();
+        assert_eq!(new_units, selected_parents);
+    }
+
+    #[test]
+    fn following_inherits_units() {
+        let n_members = NodeCount(4);
+        let mut initial_units_collector = UnitsCollector::<Hasher64>::new_initial(n_members);
+        let units = random_full_parent_units_up_to(1, n_members, 43);
+        for unit in &units[0] {
+            initial_units_collector.add_unit(unit);
+        }
+        let mut units_collector = UnitsCollector::from_previous(&initial_units_collector);
+        for unit in units[1].iter().take(3) {
+            units_collector.add_unit(unit);
+        }
+
+        let parents = units_collector
+            .prospective_parents(NodeIndex(0))
+            .expect("we should be able to retrieve parents");
+        assert_eq!(parents.item_count(), 4);
+
+        let mut new_units: Vec<_> = units[1]
+            .iter()
+            .take(3)
+            .map(|unit| (unit.hash(), unit.round()))
+            .collect();
+        new_units.push((units[0][3].hash(), units[0][3].round()));
+        let selected_parents: Vec<_> = parents.values().cloned().collect();
+        assert_eq!(new_units, selected_parents);
+    }
+}
diff --git a/consensus/src/creation/creator.rs b/consensus/src/creation/creator.rs
index a8653cc8..182e8a76 100644
--- a/consensus/src/creation/creator.rs
+++ b/consensus/src/creation/creator.rs
@@ -1,55 +1,10 @@
 use crate::{
+    creation::collector::{ConstraintError, UnitsCollector},
     units::{ControlHash, PreUnit, Unit},
     Hasher, NodeCount, NodeIndex, NodeMap, Round,
 };
 use anyhow::Result;
-use thiserror::Error;
-
-#[derive(Eq, Error, Debug, PartialEq)]
-enum ConstraintError {
-    #[error("Not enough parents.")]
-    NotEnoughParents,
-    #[error("Missing own parent.")]
-    MissingOwnParent,
-}
-
-#[derive(Clone)]
-struct UnitsCollector<H: Hasher> {
-    candidates: NodeMap<H::Hash>,
-    n_candidates: NodeCount,
-}
-
-impl<H: Hasher> UnitsCollector<H> {
-    pub fn new(n_members: NodeCount) -> Self {
-        Self {
-            candidates: NodeMap::with_size(n_members),
-            n_candidates: NodeCount(0),
-        }
-    }
-
-    pub fn add_unit<U: Unit<Hasher = H>>(&mut self, unit: &U) {
-        let node_id = unit.creator();
-        let hash = unit.hash();
-
-        if self.candidates.get(node_id).is_none() {
-            self.candidates.insert(node_id, hash);
-            self.n_candidates += NodeCount(1);
-        }
-    }
-
-    pub fn prospective_parents(
-        &self,
-        node_id: NodeIndex,
-    ) -> Result<&NodeMap<H::Hash>, ConstraintError> {
-        if self.n_candidates < self.candidates.size().consensus_threshold() {
-            return Err(ConstraintError::NotEnoughParents);
-        }
-        if self.candidates.get(node_id).is_none() {
-            return Err(ConstraintError::MissingOwnParent);
-        }
-        Ok(&self.candidates)
-    }
-}
+use std::cmp;
 
 pub struct Creator<H: Hasher> {
     round_collectors: Vec<UnitsCollector<H>>,
@@ -62,7 +17,7 @@ impl<H: Hasher> Creator<H> {
         Creator {
             node_id,
             n_members,
-            round_collectors: vec![UnitsCollector::new(n_members)],
+            round_collectors: vec![UnitsCollector::new_initial(n_members)],
         }
     }
 
@@ -72,59 +27,53 @@ impl<H: Hasher> Creator<H> {
 
     // gets or initializes a unit collector for a given round (and all between if not there)
     fn get_or_initialize_collector_for_round(&mut self, round: Round) -> &mut UnitsCollector<H> {
-        let round_ix = usize::from(round);
-        if round > self.current_round() {
-            let new_size = round_ix + 1;
-            self.round_collectors
-                .resize(new_size, UnitsCollector::new(self.n_members));
-        };
-        &mut self.round_collectors[round_ix]
+        while round > self.current_round() {
+            let next_collector = UnitsCollector::from_previous(
+                self.round_collectors
+                    .last()
+                    .expect("we always have at least one"),
+            );
+            self.round_collectors.push(next_collector);
+        }
+        &mut self.round_collectors[round as usize]
     }
 
     /// To create a new unit, we need to have at least the consensus threshold of parents available in previous round.
     /// Additionally, our unit from previous round must be available.
     pub fn create_unit(&self, round: Round) -> Result<PreUnit<H>> {
-        let parents = match round.checked_sub(1) {
-            None => NodeMap::with_size(self.n_members),
-            Some(prev_round) => {
-                let parents = self
-                    .round_collectors
+        let control_hash = match round.checked_sub(1) {
+            None => ControlHash::new(&NodeMap::with_size(self.n_members)),
+            Some(prev_round) => ControlHash::new(
+                self.round_collectors
                     .get(usize::from(prev_round))
                     .ok_or(ConstraintError::NotEnoughParents)?
-                    .prospective_parents(self.node_id)?
-                    .clone();
-                let mut parents_with_rounds_and_hashes = NodeMap::with_size(parents.size());
-                for (parent_index, hash) in parents.into_iter() {
-                    // we cannot have here round 0 units
-                    parents_with_rounds_and_hashes.insert(parent_index, (hash, prev_round));
-                }
-                parents_with_rounds_and_hashes
-            }
+                    .prospective_parents(self.node_id)?,
+            ),
         };
 
-        Ok(PreUnit::new(
-            self.node_id,
-            round,
-            ControlHash::new(&parents),
-        ))
+        Ok(PreUnit::new(self.node_id, round, control_hash))
     }
 
     pub fn add_unit<U: Unit<Hasher = H>>(&mut self, unit: &U) {
-        self.get_or_initialize_collector_for_round(unit.round())
-            .add_unit(unit);
+        let start_round = unit.round();
+        let end_round = cmp::max(start_round, self.current_round());
+        for round in start_round..=end_round {
+            self.get_or_initialize_collector_for_round(round)
+                .add_unit(unit);
+        }
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use super::{Creator as GenericCreator, UnitsCollector};
     use crate::{
-        creation::creator::ConstraintError,
-        units::{create_preunits, creator_set, preunit_to_full_unit, Unit},
+        creation::creator::Creator as GenericCreator,
+        units::{
+            create_preunits, creator_set, preunit_to_full_unit, random_full_parent_units_up_to,
+        },
         NodeCount, NodeIndex,
     };
     use aleph_bft_mock::Hasher64;
-    use std::collections::HashSet;
 
     type Creator = GenericCreator<Hasher64>;
 
@@ -274,71 +223,58 @@ mod tests {
     }
 
     #[test]
-    fn units_collector_successfully_computes_parents() {
-        let n_members = NodeCount(4);
-        let creators = creator_set(n_members);
-        let new_units = create_preunits(creators.iter(), 0);
-        let new_units: Vec<_> = new_units
-            .into_iter()
-            .map(|pu| preunit_to_full_unit(pu, 0))
-            .collect();
-
-        let mut units_collector = UnitsCollector::new(n_members);
-        new_units
-            .iter()
-            .for_each(|unit| units_collector.add_unit(unit));
-
-        let parents = units_collector
-            .prospective_parents(NodeIndex(0))
-            .expect("we should be able to retrieve parents");
-        assert_eq!(parents.item_count(), 4);
-
-        let new_units: HashSet<_> = new_units.iter().map(|unit| unit.hash()).collect();
-        let selected_parents = parents.values().cloned().collect();
-        assert_eq!(new_units, selected_parents);
-    }
-
-    #[test]
-    fn units_collector_returns_err_when_not_enough_parents() {
-        let n_members = NodeCount(4);
-        let creators = creator_set(n_members);
-        let new_units = create_preunits(creators.iter().take(2), 0);
-        let new_units: Vec<_> = new_units
-            .into_iter()
-            .map(|pu| preunit_to_full_unit(pu, 0))
-            .collect();
-
-        let mut units_collector = UnitsCollector::new(n_members);
-        new_units
-            .iter()
-            .for_each(|unit| units_collector.add_unit(unit));
-
-        let parents = units_collector.prospective_parents(NodeIndex(0));
+    fn creates_unit_with_ancient_parents() {
+        let n_members = NodeCount(7);
+        let mut creator = Creator::new(NodeIndex(0), n_members);
+        let units = random_full_parent_units_up_to(1, n_members, 43);
+        creator.add_units(&units[0]);
+        for unit in units[1].iter().take(5) {
+            creator.add_unit(unit);
+        }
+        let round = 2;
+        let preunit = creator
+            .create_unit(round)
+            .expect("Creation should succeed.");
+        assert_eq!(preunit.round(), round);
+        for coord in preunit.control_hash().parents().take(5) {
+            assert_eq!(coord.round(), round - 1);
+        }
         assert_eq!(
-            parents.expect_err("should be an error"),
-            ConstraintError::NotEnoughParents
+            preunit
+                .control_hash()
+                .parents()
+                .nth(5)
+                .expect("there is a fourth parent")
+                .round(),
+            round - 2
         );
     }
 
     #[test]
-    fn units_collector_returns_err_when_missing_own_parent() {
-        let n_members = NodeCount(4);
-        let creators = creator_set(n_members);
-        let new_units = create_preunits(creators.iter().take(3), 0);
-        let new_units: Vec<_> = new_units
-            .into_iter()
-            .map(|pu| preunit_to_full_unit(pu, 0))
-            .collect();
-
-        let mut units_collector = UnitsCollector::new(n_members);
-        new_units
-            .iter()
-            .for_each(|unit| units_collector.add_unit(unit));
-
-        let parents = units_collector.prospective_parents(NodeIndex(3));
+    fn creates_unit_with_ancient_parents_weird_order() {
+        let n_members = NodeCount(7);
+        let mut creator = Creator::new(NodeIndex(0), n_members);
+        let units = random_full_parent_units_up_to(1, n_members, 43);
+        for unit in units[1].iter().take(5) {
+            creator.add_unit(unit);
+        }
+        creator.add_units(&units[0]);
+        let round = 2;
+        let preunit = creator
+            .create_unit(round)
+            .expect("Creation should succeed.");
+        assert_eq!(preunit.round(), round);
+        for coord in preunit.control_hash().parents().take(5) {
+            assert_eq!(coord.round(), round - 1);
+        }
         assert_eq!(
-            parents.expect_err("should be an error"),
-            ConstraintError::MissingOwnParent
+            preunit
+                .control_hash()
+                .parents()
+                .nth(5)
+                .expect("there is a fourth parent")
+                .round(),
+            round - 2
         );
     }
 }
diff --git a/consensus/src/creation/mod.rs b/consensus/src/creation/mod.rs
index 9515820d..186aa849 100644
--- a/consensus/src/creation/mod.rs
+++ b/consensus/src/creation/mod.rs
@@ -13,6 +13,7 @@ use futures::{
 use futures_timer::Delay;
 use log::{debug, error, trace, warn};
 
+mod collector;
 mod creator;
 mod packer;