diff --git a/crates/examples/infra/mod.rs b/crates/examples/infra/mod.rs
index 80ef89bb02..944456f917 100644
--- a/crates/examples/infra/mod.rs
+++ b/crates/examples/infra/mod.rs
@@ -324,7 +324,7 @@ pub trait RunDA<
     /// get the anchored view
     /// Note: sequencing leaf does not have state, so does not return state
     async fn initialize_state_and_hotshot(&self) -> SystemContextHandle<TYPES, NODE> {
-        let initializer = hotshot::HotShotInitializer::<TYPES>::from_genesis(&TestInstanceState {})
+        let initializer = hotshot::HotShotInitializer::<TYPES>::from_genesis(TestInstanceState {})
             .expect("Couldn't generate genesis block");
 
         let config = self.get_config();
diff --git a/crates/hotshot/src/lib.rs b/crates/hotshot/src/lib.rs
index 0fe820fb32..26db8d482e 100644
--- a/crates/hotshot/src/lib.rs
+++ b/crates/hotshot/src/lib.rs
@@ -102,6 +102,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> Networks<TYPES, I> {
 }
 
 /// Bundle of all the memberships a consensus instance uses
+#[derive(Clone)]
 pub struct Memberships<TYPES: NodeType> {
     /// Quorum Membership
     pub quorum_membership: TYPES::Membership,
@@ -155,8 +156,7 @@ pub struct SystemContext<TYPES: NodeType, I: NodeImplementation<TYPES>> {
 }
 
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>> SystemContext<TYPES, I> {
-    /// Creates a new [`Arc<SystemContext>`] with the given configuration options and sets it up with the given
-    /// genesis block
+    /// Creates a new [`Arc<SystemContext>`] with the given configuration options.
     ///
     /// To do a full initialization, use `fn init` instead, which will set up background tasks as
     /// well.
@@ -187,7 +187,9 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> SystemContext<TYPES, I> {
 
         // insert genesis (or latest block) to state map
         let mut validated_state_map = BTreeMap::default();
-        let validated_state = Arc::new(TYPES::ValidatedState::genesis(&instance_state));
+        let validated_state = Arc::new(TYPES::ValidatedState::from_header(
+            &anchored_leaf.block_header,
+        ));
         validated_state_map.insert(
             anchored_leaf.get_view_number(),
             View {
@@ -211,10 +213,12 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> SystemContext<TYPES, I> {
                 }
             };
             saved_payloads.insert(anchored_leaf.get_view_number(), encoded_txns.clone());
+            // View 1 doesn't have DA which is responsible for saving the payloads, so we store the
+            // payload for view 1 manually during the intialization.
             saved_payloads.insert(TYPES::Time::new(1), encoded_txns);
         }
 
-        let start_view = anchored_leaf.get_view_number();
+        let start_view = initializer.start_view;
 
         let consensus = Consensus {
             instance_state,
@@ -279,7 +283,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> SystemContext<TYPES, I> {
         broadcast_event(event, &self.output_event_stream.0).await;
     }
 
-    /// Publishes a transaction asynchronously to the network
+    /// Publishes a transaction asynchronously to the network.
     ///
     /// # Errors
     ///
@@ -290,11 +294,12 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> SystemContext<TYPES, I> {
         transaction: TYPES::Transaction,
     ) -> Result<(), HotShotError<TYPES>> {
         trace!("Adding transaction to our own queue");
-        // Wrap up a message
-        // TODO place a view number here that makes sense
-        // we haven't worked out how this will work yet
-        let message = DataMessage::SubmitTransaction(transaction.clone(), TYPES::Time::new(0));
+
         let api = self.clone();
+        let view_number = api.consensus.read().await.cur_view;
+
+        // Wrap up a message
+        let message = DataMessage::SubmitTransaction(transaction.clone(), view_number);
 
         async_spawn(async move {
             let da_membership = &api.memberships.da_membership.clone();
@@ -312,11 +317,11 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> SystemContext<TYPES, I> {
                             sender: api.public_key.clone(),
                             kind: MessageKind::from(message),
                         },
-                        da_membership.get_committee(TYPES::Time::new(0)),
+                        da_membership.get_committee(view_number),
                     ),
                 api
                     .send_external_event(Event {
-                        view_number: api.consensus.read().await.cur_view,
+                        view_number,
                         event: EventType::Transactions {
                             transactions: vec![transaction],
                         },
@@ -606,26 +611,37 @@ pub struct HotShotInitializer<TYPES: NodeType> {
 
     /// Instance-level state.
     instance_state: TYPES::InstanceState,
+
+    /// Starting view number that we are confident won't lead to a double vote after restart.
+    start_view: TYPES::Time,
 }
 
 impl<TYPES: NodeType> HotShotInitializer<TYPES> {
     /// initialize from genesis
     /// # Errors
     /// If we are unable to apply the genesis block to the default state
-    pub fn from_genesis(
-        instance_state: &TYPES::InstanceState,
-    ) -> Result<Self, HotShotError<TYPES>> {
+    pub fn from_genesis(instance_state: TYPES::InstanceState) -> Result<Self, HotShotError<TYPES>> {
         Ok(Self {
-            inner: Leaf::genesis(instance_state),
-            instance_state: instance_state.clone(),
+            inner: Leaf::genesis(&instance_state),
+            instance_state,
+            start_view: TYPES::Time::new(0),
         })
     }
 
-    /// reload previous state based on most recent leaf and the instance-level state.
-    pub fn from_reload(anchor_leaf: Leaf<TYPES>, instance_state: TYPES::InstanceState) -> Self {
+    /// Reload previous state based on most recent leaf and the instance-level state.
+    ///
+    /// # Arguments
+    /// *  `start_view` - The minimum view number that we are confident won't lead to a double vote
+    /// after restart.
+    pub fn from_reload(
+        anchor_leaf: Leaf<TYPES>,
+        instance_state: TYPES::InstanceState,
+        start_view: TYPES::Time,
+    ) -> Self {
         Self {
             inner: anchor_leaf,
             instance_state,
+            start_view,
         }
     }
 }
diff --git a/crates/hotshot/src/tasks/mod.rs b/crates/hotshot/src/tasks/mod.rs
index d250f549d8..cefe1ecd4a 100644
--- a/crates/hotshot/src/tasks/mod.rs
+++ b/crates/hotshot/src/tasks/mod.rs
@@ -172,7 +172,7 @@ pub async fn add_consensus_task<TYPES: NodeType, I: NodeImplementation<TYPES>>(
     rx: Receiver<HotShotEvent<TYPES>>,
     handle: &SystemContextHandle<TYPES, I>,
 ) {
-    let consensus_state = ConsensusTaskState::create_from(handle);
+    let consensus_state = ConsensusTaskState::create_from(handle).await;
 
     inject_consensus_polls(&consensus_state).await;
 
@@ -187,7 +187,7 @@ pub async fn add_vid_task<TYPES: NodeType, I: NodeImplementation<TYPES>>(
     rx: Receiver<HotShotEvent<TYPES>>,
     handle: &SystemContextHandle<TYPES, I>,
 ) {
-    let vid_state = VIDTaskState::create_from(handle);
+    let vid_state = VIDTaskState::create_from(handle).await;
     let task = Task::new(tx, rx, task_reg.clone(), vid_state);
     task_reg.run_task(task).await;
 }
@@ -199,7 +199,7 @@ pub async fn add_upgrade_task<TYPES: NodeType, I: NodeImplementation<TYPES>>(
     rx: Receiver<HotShotEvent<TYPES>>,
     handle: &SystemContextHandle<TYPES, I>,
 ) {
-    let upgrade_state = UpgradeTaskState::create_from(handle);
+    let upgrade_state = UpgradeTaskState::create_from(handle).await;
 
     let task = Task::new(tx, rx, task_reg.clone(), upgrade_state);
     task_reg.run_task(task).await;
@@ -212,7 +212,7 @@ pub async fn add_da_task<TYPES: NodeType, I: NodeImplementation<TYPES>>(
     handle: &SystemContextHandle<TYPES, I>,
 ) {
     // build the da task
-    let da_state = DATaskState::create_from(handle);
+    let da_state = DATaskState::create_from(handle).await;
 
     let task = Task::new(tx, rx, task_reg.clone(), da_state);
     task_reg.run_task(task).await;
@@ -225,7 +225,7 @@ pub async fn add_transaction_task<TYPES: NodeType, I: NodeImplementation<TYPES>>
     rx: Receiver<HotShotEvent<TYPES>>,
     handle: &SystemContextHandle<TYPES, I>,
 ) {
-    let transactions_state = TransactionTaskState::create_from(handle);
+    let transactions_state = TransactionTaskState::create_from(handle).await;
 
     let task = Task::new(tx, rx, task_reg.clone(), transactions_state);
     task_reg.run_task(task).await;
@@ -238,7 +238,7 @@ pub async fn add_view_sync_task<TYPES: NodeType, I: NodeImplementation<TYPES>>(
     rx: Receiver<HotShotEvent<TYPES>>,
     handle: &SystemContextHandle<TYPES, I>,
 ) {
-    let view_sync_state = ViewSyncTaskState::create_from(handle);
+    let view_sync_state = ViewSyncTaskState::create_from(handle).await;
 
     let task = Task::new(tx, rx, task_reg.clone(), view_sync_state);
     task_reg.run_task(task).await;
diff --git a/crates/hotshot/src/tasks/task_state.rs b/crates/hotshot/src/tasks/task_state.rs
index 19f6e026dd..6e8d023113 100644
--- a/crates/hotshot/src/tasks/task_state.rs
+++ b/crates/hotshot/src/tasks/task_state.rs
@@ -1,5 +1,6 @@
 use crate::types::SystemContextHandle;
 
+use async_trait::async_trait;
 use hotshot_constants::VERSION_0_1;
 use hotshot_task_impls::{
     consensus::{CommitmentAndMetadata, ConsensusTaskState},
@@ -24,24 +25,26 @@ use std::{
 };
 
 /// Trait for creating task states.
+#[async_trait]
 pub trait CreateTaskState<TYPES, I>
 where
     TYPES: NodeType,
     I: NodeImplementation<TYPES>,
 {
     /// Function to create the task state from a given `SystemContextHandle`.
-    fn create_from(handle: &SystemContextHandle<TYPES, I>) -> Self;
+    async fn create_from(handle: &SystemContextHandle<TYPES, I>) -> Self;
 }
 
+#[async_trait]
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     for UpgradeTaskState<TYPES, I, SystemContextHandle<TYPES, I>>
 {
-    fn create_from(
+    async fn create_from(
         handle: &SystemContextHandle<TYPES, I>,
     ) -> UpgradeTaskState<TYPES, I, SystemContextHandle<TYPES, I>> {
         UpgradeTaskState {
             api: handle.clone(),
-            cur_view: TYPES::Time::new(0),
+            cur_view: handle.get_cur_view().await,
             quorum_membership: handle.hotshot.memberships.quorum_membership.clone().into(),
             quorum_network: handle.hotshot.networks.quorum_network.clone(),
             should_vote: |_upgrade_proposal| false,
@@ -53,16 +56,17 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     }
 }
 
+#[async_trait]
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     for VIDTaskState<TYPES, I, SystemContextHandle<TYPES, I>>
 {
-    fn create_from(
+    async fn create_from(
         handle: &SystemContextHandle<TYPES, I>,
     ) -> VIDTaskState<TYPES, I, SystemContextHandle<TYPES, I>> {
         VIDTaskState {
             api: handle.clone(),
             consensus: handle.hotshot.get_consensus(),
-            cur_view: TYPES::Time::new(0),
+            cur_view: handle.get_cur_view().await,
             vote_collector: None,
             network: handle.hotshot.networks.quorum_network.clone(),
             membership: handle.hotshot.memberships.vid_membership.clone().into(),
@@ -73,10 +77,11 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     }
 }
 
+#[async_trait]
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     for DATaskState<TYPES, I, SystemContextHandle<TYPES, I>>
 {
-    fn create_from(
+    async fn create_from(
         handle: &SystemContextHandle<TYPES, I>,
     ) -> DATaskState<TYPES, I, SystemContextHandle<TYPES, I>> {
         DATaskState {
@@ -85,7 +90,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
             da_membership: handle.hotshot.memberships.da_membership.clone().into(),
             da_network: handle.hotshot.networks.da_network.clone(),
             quorum_membership: handle.hotshot.memberships.quorum_membership.clone().into(),
-            cur_view: TYPES::Time::new(0),
+            cur_view: handle.get_cur_view().await,
             vote_collector: None.into(),
             public_key: handle.public_key().clone(),
             private_key: handle.private_key().clone(),
@@ -94,15 +99,17 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     }
 }
 
+#[async_trait]
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     for ViewSyncTaskState<TYPES, I, SystemContextHandle<TYPES, I>>
 {
-    fn create_from(
+    async fn create_from(
         handle: &SystemContextHandle<TYPES, I>,
     ) -> ViewSyncTaskState<TYPES, I, SystemContextHandle<TYPES, I>> {
+        let cur_view = handle.get_cur_view().await;
         ViewSyncTaskState {
-            current_view: TYPES::Time::new(0),
-            next_view: TYPES::Time::new(0),
+            current_view: cur_view,
+            next_view: cur_view,
             network: handle.hotshot.networks.quorum_network.clone(),
             membership: handle
                 .hotshot
@@ -125,10 +132,11 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     }
 }
 
+#[async_trait]
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     for TransactionTaskState<TYPES, I, SystemContextHandle<TYPES, I>>
 {
-    fn create_from(
+    async fn create_from(
         handle: &SystemContextHandle<TYPES, I>,
     ) -> TransactionTaskState<TYPES, I, SystemContextHandle<TYPES, I>> {
         TransactionTaskState {
@@ -136,7 +144,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
             consensus: handle.hotshot.get_consensus(),
             transactions: Arc::default(),
             seen_transactions: HashSet::new(),
-            cur_view: TYPES::Time::new(0),
+            cur_view: handle.get_cur_view().await,
             network: handle.hotshot.networks.quorum_network.clone(),
             membership: handle.hotshot.memberships.quorum_membership.clone().into(),
             public_key: handle.public_key().clone(),
@@ -146,10 +154,11 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     }
 }
 
+#[async_trait]
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
     for ConsensusTaskState<TYPES, I, SystemContextHandle<TYPES, I>>
 {
-    fn create_from(
+    async fn create_from(
         handle: &SystemContextHandle<TYPES, I>,
     ) -> ConsensusTaskState<TYPES, I, SystemContextHandle<TYPES, I>> {
         let consensus = handle.hotshot.get_consensus();
@@ -163,7 +172,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>> CreateTaskState<TYPES, I>
         ConsensusTaskState {
             consensus,
             timeout: handle.hotshot.config.next_view_timeout,
-            cur_view: TYPES::Time::new(0),
+            cur_view: handle.get_cur_view().await,
             payload_commitment_and_metadata: Some(CommitmentAndMetadata {
                 commitment: payload_commitment,
                 metadata,
diff --git a/crates/hotshot/src/types/handle.rs b/crates/hotshot/src/types/handle.rs
index 3cd40e4fa8..b6e1ded7f8 100644
--- a/crates/hotshot/src/types/handle.rs
+++ b/crates/hotshot/src/types/handle.rs
@@ -171,9 +171,8 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES> + 'static> SystemContextHandl
         self.hotshot.public_key.clone()
     }
 
-    /// Wrapper to get this node's current view
-    #[cfg(feature = "hotshot-testing")]
-    pub async fn get_current_view(&self) -> TYPES::Time {
+    /// Wrapper to get the view number this node is on.
+    pub async fn get_cur_view(&self) -> TYPES::Time {
         self.hotshot.consensus.read().await.cur_view
     }
 }
diff --git a/crates/task-impls/src/consensus.rs b/crates/task-impls/src/consensus.rs
index db8df961a6..cda79cd3bf 100644
--- a/crates/task-impls/src/consensus.rs
+++ b/crates/task-impls/src/consensus.rs
@@ -395,15 +395,24 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, A: ConsensusApi<TYPES, I> +
                     .await;
                 }
             }));
-            let consensus = self.consensus.read().await;
+            let consensus = self.consensus.upgradable_read().await;
             consensus
                 .metrics
                 .current_view
                 .set(usize::try_from(self.cur_view.get_u64()).unwrap());
-            consensus.metrics.number_of_views_since_last_decide.set(
-                usize::try_from(self.cur_view.get_u64()).unwrap()
-                    - usize::try_from(consensus.last_decided_view.get_u64()).unwrap(),
-            );
+            // Do the comparison before the substraction to avoid potential overflow, since
+            // `last_decided_view` may be greater than `cur_view` if the node is catching up.
+            if usize::try_from(self.cur_view.get_u64()).unwrap()
+                > usize::try_from(consensus.last_decided_view.get_u64()).unwrap()
+            {
+                consensus.metrics.number_of_views_since_last_decide.set(
+                    usize::try_from(self.cur_view.get_u64()).unwrap()
+                        - usize::try_from(consensus.last_decided_view.get_u64()).unwrap(),
+                );
+            }
+            let mut consensus = RwLockUpgradableReadGuard::upgrade(consensus).await;
+            consensus.update_view(new_view);
+            drop(consensus);
 
             return true;
         }
diff --git a/crates/task-impls/src/da.rs b/crates/task-impls/src/da.rs
index aa65a1d26d..67850d5059 100644
--- a/crates/task-impls/src/da.rs
+++ b/crates/task-impls/src/da.rs
@@ -228,6 +228,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, A: ConsensusApi<TYPES, I> +
                     warn!("View changed by more than 1 going to view {:?}", view);
                 }
                 self.cur_view = view;
+                self.consensus.write().await.update_view(view);
 
                 // Inject view info into network
                 let is_da = self
diff --git a/crates/task-impls/src/transactions.rs b/crates/task-impls/src/transactions.rs
index 994870d414..293842ee15 100644
--- a/crates/task-impls/src/transactions.rs
+++ b/crates/task-impls/src/transactions.rs
@@ -184,6 +184,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, A: ConsensusApi<TYPES, I> +
                     make_block = self.membership.get_leader(view) == self.public_key;
                 }
                 self.cur_view = view;
+                self.consensus.write().await.update_view(view);
 
                 // return if we aren't the next leader or we skipped last view and aren't the current leader.
                 if !make_block && self.membership.get_leader(self.cur_view + 1) != self.public_key {
diff --git a/crates/task-impls/src/vid.rs b/crates/task-impls/src/vid.rs
index d27dc04d19..9d603a8cc2 100644
--- a/crates/task-impls/src/vid.rs
+++ b/crates/task-impls/src/vid.rs
@@ -146,6 +146,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, A: ConsensusApi<TYPES, I> +
                     warn!("View changed by more than 1 going to view {:?}", view);
                 }
                 self.cur_view = view;
+                self.consensus.write().await.update_view(view);
 
                 // Start polling for VID disperse for the new view
                 self.network
diff --git a/crates/testing-macros/tests/tests.rs b/crates/testing-macros/tests/tests.rs
index 844d3dcafd..cdd48b6b32 100644
--- a/crates/testing-macros/tests/tests.rs
+++ b/crates/testing-macros/tests/tests.rs
@@ -48,7 +48,7 @@ cross_tests!(
         }];
 
         metadata.spinning_properties = SpinningTaskDescription {
-            node_changes: vec![(5, dead_nodes)],
+            node_changes: vec![(5, dead_nodes)]
         };
         metadata.overall_safety_properties.num_failed_views = 3;
         metadata.overall_safety_properties.num_successful_views = 25;
@@ -86,7 +86,7 @@ cross_tests!(
         ];
 
         metadata.spinning_properties = SpinningTaskDescription {
-            node_changes: vec![(5, dead_nodes)],
+            node_changes: vec![(5, dead_nodes)]
         };
 
         metadata.overall_safety_properties.num_failed_views = 3;
@@ -142,7 +142,7 @@ cross_tests!(
             ];
 
         metadata.spinning_properties = SpinningTaskDescription {
-            node_changes: vec![(5, dead_nodes)],
+            node_changes: vec![(5, dead_nodes)]
         };
 
         metadata
@@ -178,7 +178,7 @@ cross_tests!(
         ];
 
         metadata.spinning_properties = SpinningTaskDescription {
-            node_changes: vec![(5, dead_nodes)],
+            node_changes: vec![(5, dead_nodes)]
         };
 
         // 2 nodes fail triggering view sync, expect no other timeouts
diff --git a/crates/testing/src/spinning_task.rs b/crates/testing/src/spinning_task.rs
index c7e73fbde8..8f37ec0dba 100644
--- a/crates/testing/src/spinning_task.rs
+++ b/crates/testing/src/spinning_task.rs
@@ -1,13 +1,20 @@
 use std::collections::HashMap;
 
-use hotshot::traits::TestableNodeImplementation;
-
 use crate::test_runner::HotShotTaskCompleted;
-use crate::test_runner::LateStartNode;
-use crate::test_runner::Node;
+use crate::test_runner::{LateStartNode, Node, TestRunner};
+use either::{Left, Right};
+use hotshot::{traits::TestableNodeImplementation, HotShotInitializer};
+use hotshot_example_types::state_types::TestInstanceState;
 use hotshot_task::task::{Task, TaskState, TestTaskState};
-use hotshot_types::traits::network::ConnectedNetwork;
-use hotshot_types::{event::Event, traits::node_implementation::NodeType};
+use hotshot_types::{data::Leaf, ValidatorConfig};
+use hotshot_types::{
+    event::Event,
+    message::Message,
+    traits::{
+        network::ConnectedNetwork,
+        node_implementation::{NodeImplementation, NodeType},
+    },
+};
 use snafu::Snafu;
 use std::collections::BTreeMap;
 /// convience type for state and block
@@ -29,6 +36,8 @@ pub struct SpinningTask<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> {
     pub(crate) changes: BTreeMap<TYPES::Time, Vec<ChangeNode>>,
     /// most recent view seen by spinning task
     pub(crate) latest_view: Option<TYPES::Time>,
+    /// Last decided leaf that can be used as the anchor leaf to initialize the node.
+    pub(crate) last_decided_leaf: Leaf<TYPES>,
 }
 
 impl<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> TaskState for SpinningTask<TYPES, I> {
@@ -48,8 +57,14 @@ impl<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> TaskState for Spinni
     }
 }
 
-impl<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> TestTaskState
-    for SpinningTask<TYPES, I>
+impl<
+        TYPES: NodeType<InstanceState = TestInstanceState>,
+        I: TestableNodeImplementation<TYPES>,
+        N: ConnectedNetwork<Message<TYPES>, TYPES::SignatureKey>,
+    > TestTaskState for SpinningTask<TYPES, I>
+where
+    I: TestableNodeImplementation<TYPES, CommitteeElectionConfig = TYPES::ElectionConfigType>,
+    I: NodeImplementation<TYPES, QuorumNetwork = N, CommitteeNetwork = N>,
 {
     type Message = Event<TYPES>;
 
@@ -79,6 +94,34 @@ impl<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> TestTaskState
                             let node_id = idx.try_into().unwrap();
                             if let Some(node) = state.late_start.remove(&node_id) {
                                 tracing::error!("Node {} spinning up late", idx);
+                                let node_id = idx.try_into().unwrap();
+                                let context = match node.context {
+                                    Left(context) => context,
+                                    // Node not initialized. Initialize it
+                                    // based on the received leaf.
+                                    Right((storage, memberships, config)) => {
+                                        let initializer = HotShotInitializer::<TYPES>::from_reload(
+                                            state.last_decided_leaf.clone(),
+                                            TestInstanceState {},
+                                            view_number,
+                                        );
+                                        // We assign node's public key and stake value rather than read from config file since it's a test
+                                        let validator_config =
+                                            ValidatorConfig::generated_from_seed_indexed(
+                                                [0u8; 32], node_id, 1,
+                                            );
+                                        TestRunner::add_node_with_config(
+                                            node_id,
+                                            node.networks.clone(),
+                                            storage,
+                                            memberships,
+                                            initializer,
+                                            config,
+                                            validator_config,
+                                        )
+                                        .await
+                                    }
+                                };
 
                                 // Create the node and add it to the state, so we can shut them
                                 // down properly later to avoid the overflow error in the overall
@@ -86,7 +129,7 @@ impl<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> TestTaskState
                                 let node = Node {
                                     node_id,
                                     networks: node.networks,
-                                    handle: node.context.run_tasks().await,
+                                    handle: context.run_tasks().await,
                                 };
                                 state.handles.push(node.clone());
 
diff --git a/crates/testing/src/task_helpers.rs b/crates/testing/src/task_helpers.rs
index 4afec99b74..faf0b620a8 100644
--- a/crates/testing/src/task_helpers.rs
+++ b/crates/testing/src/task_helpers.rs
@@ -63,7 +63,7 @@ pub async fn build_system_handle(
     let storage = (launcher.resource_generator.storage)(node_id);
     let config = launcher.resource_generator.config.clone();
 
-    let initializer = HotShotInitializer::<TestTypes>::from_genesis(&TestInstanceState {}).unwrap();
+    let initializer = HotShotInitializer::<TestTypes>::from_genesis(TestInstanceState {}).unwrap();
 
     let known_nodes_with_stake = config.known_nodes_with_stake.clone();
     let private_key = config.my_own_validator_config.private_key.clone();
diff --git a/crates/testing/src/test_builder.rs b/crates/testing/src/test_builder.rs
index 0617d18d26..4fb60d2b90 100644
--- a/crates/testing/src/test_builder.rs
+++ b/crates/testing/src/test_builder.rs
@@ -10,15 +10,15 @@ use hotshot_types::{
 };
 
 use super::completion_task::{CompletionTaskDescription, TimeBasedCompletionTaskDescription};
+use super::{
+    overall_safety_task::OverallSafetyPropertiesDescription, txn_task::TxnTaskDescription,
+};
 use crate::{
     spinning_task::SpinningTaskDescription,
     test_launcher::{ResourceGenerators, TestLauncher},
     view_sync_task::ViewSyncTaskDescription,
 };
-
-use super::{
-    overall_safety_task::OverallSafetyPropertiesDescription, txn_task::TxnTaskDescription,
-};
+use hotshot_example_types::state_types::TestInstanceState;
 /// data describing how a round should be timed.
 #[derive(Clone, Debug, Copy)]
 pub struct TimingData {
@@ -43,6 +43,9 @@ pub struct TestMetadata {
     pub total_nodes: usize,
     /// nodes available at start
     pub start_nodes: usize,
+    /// Whether to skip initializing nodes that will start late, which will catch up later with
+    /// `HotShotInitializer::from_reload` in the spinning task.
+    pub skip_late: bool,
     /// number of bootstrap nodes (libp2p usage only)
     pub num_bootstrap_nodes: usize,
     /// Size of the DA committee for the test
@@ -177,6 +180,7 @@ impl Default for TestMetadata {
             min_transactions: 0,
             total_nodes: num_nodes,
             start_nodes: num_nodes,
+            skip_late: false,
             num_bootstrap_nodes: num_nodes,
             da_committee_size: num_nodes,
             spinning_properties: SpinningTaskDescription {
@@ -203,7 +207,10 @@ impl TestMetadata {
     /// # Panics
     /// if some of the the configuration values are zero
     #[must_use]
-    pub fn gen_launcher<TYPES: NodeType, I: TestableNodeImplementation<TYPES>>(
+    pub fn gen_launcher<
+        TYPES: NodeType<InstanceState = TestInstanceState>,
+        I: TestableNodeImplementation<TYPES, CommitteeElectionConfig = TYPES::ElectionConfigType>,
+    >(
         self,
         node_id: u64,
     ) -> TestLauncher<TYPES, I>
diff --git a/crates/testing/src/test_runner.rs b/crates/testing/src/test_runner.rs
index 662997d144..3148f23089 100644
--- a/crates/testing/src/test_runner.rs
+++ b/crates/testing/src/test_runner.rs
@@ -12,6 +12,7 @@ use crate::{
     view_sync_task::ViewSyncTask,
 };
 use async_broadcast::broadcast;
+use either::Either::{self, Left, Right};
 use futures::future::join_all;
 use hotshot::{types::SystemContextHandle, Memberships};
 use hotshot_example_types::state_types::TestInstanceState;
@@ -22,6 +23,7 @@ use hotshot_constants::EVENT_CHANNEL_SIZE;
 use hotshot_task::task::{Task, TaskRegistry, TestTask};
 use hotshot_types::{
     consensus::ConsensusMetricsValue,
+    data::Leaf,
     traits::{
         election::Membership,
         node_implementation::{ConsensusTime, NodeType},
@@ -52,13 +54,24 @@ pub struct Node<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> {
     pub handle: SystemContextHandle<TYPES, I>,
 }
 
+/// Either the node context or the parameters to construct the context for nodes that start late.
+pub type LateNodeContext<TYPES, I> = Either<
+    Arc<SystemContext<TYPES, I>>,
+    (
+        <I as NodeImplementation<TYPES>>::Storage,
+        Memberships<TYPES>,
+        HotShotConfig<<TYPES as NodeType>::SignatureKey, <TYPES as NodeType>::ElectionConfigType>,
+    ),
+>;
+
 /// A yet-to-be-started node that participates in tests
 #[derive(Clone)]
 pub struct LateStartNode<TYPES: NodeType, I: TestableNodeImplementation<TYPES>> {
     /// The underlying networks belonging to the node
     pub networks: Networks<TYPES, I>,
-    /// The context to which we will use to launch HotShot when it's time
-    pub context: Arc<SystemContext<TYPES, I>>,
+    /// Either the context to which we will use to launch HotShot for initialized node when it's
+    /// time, or the parameters that will be used to initialize the node and launch HotShot.
+    pub context: LateNodeContext<TYPES, I>,
 }
 
 /// The runner of a test network
@@ -110,6 +123,7 @@ where
     I: NodeImplementation<TYPES, QuorumNetwork = N, CommitteeNetwork = N>,
 {
     /// excecute test
+    ///
     /// # Panics
     /// if the test fails
     #[allow(clippy::too_many_lines)]
@@ -196,6 +210,7 @@ where
             late_start,
             latest_view: None,
             changes,
+            last_decided_leaf: Leaf::genesis(&TestInstanceState {}),
         };
         let spinning_task = TestTask::<SpinningTask<TYPES, I>, SpinningTask<TYPES, I>>::new(
             Task::new(tx.clone(), rx.clone(), reg.clone(), spinning_task_state),
@@ -299,45 +314,82 @@ where
         );
     }
 
-    /// add nodes
+    /// Add nodes.
+    ///
     /// # Panics
     /// Panics if unable to create a [`HotShotInitializer`]
     pub async fn add_nodes(&mut self, total: usize, late_start: &HashSet<u64>) -> Vec<u64> {
         let mut results = vec![];
         for i in 0..total {
-            tracing::debug!("launch node {}", i);
             let node_id = self.next_node_id;
+            self.next_node_id += 1;
+            tracing::debug!("launch node {}", i);
             let storage = (self.launcher.resource_generator.storage)(node_id);
             let config = self.launcher.resource_generator.config.clone();
-            let initializer =
-                HotShotInitializer::<TYPES>::from_genesis(&TestInstanceState {}).unwrap();
+            let known_nodes_with_stake = config.known_nodes_with_stake.clone();
+            let quorum_election_config = config.election_config.clone().unwrap_or_else(|| {
+                TYPES::Membership::default_election_config(config.total_nodes.get() as u64)
+            });
+            let committee_election_config = I::committee_election_config_generator();
+            let memberships = Memberships {
+                quorum_membership: <TYPES as NodeType>::Membership::create_election(
+                    known_nodes_with_stake.clone(),
+                    quorum_election_config.clone(),
+                ),
+                da_membership: <TYPES as NodeType>::Membership::create_election(
+                    known_nodes_with_stake.clone(),
+                    committee_election_config(config.da_committee_size as u64),
+                ),
+                vid_membership: <TYPES as NodeType>::Membership::create_election(
+                    known_nodes_with_stake.clone(),
+                    quorum_election_config.clone(),
+                ),
+                view_sync_membership: <TYPES as NodeType>::Membership::create_election(
+                    known_nodes_with_stake.clone(),
+                    quorum_election_config,
+                ),
+            };
             let networks = (self.launcher.resource_generator.channel_generator)(node_id);
-            // We assign node's public key and stake value rather than read from config file since it's a test
-            let validator_config =
-                ValidatorConfig::generated_from_seed_indexed([0u8; 32], node_id, 1);
-            let hotshot = self
-                .add_node_with_config(
-                    networks.clone(),
-                    storage,
-                    initializer,
-                    config,
-                    validator_config,
-                )
-                .await;
-            if late_start.contains(&node_id) {
+
+            if self.launcher.metadata.skip_late && late_start.contains(&node_id) {
                 self.late_start.insert(
                     node_id,
                     LateStartNode {
                         networks,
-                        context: hotshot,
+                        context: Right((storage, memberships, config)),
                     },
                 );
             } else {
-                self.nodes.push(Node {
+                let initializer =
+                    HotShotInitializer::<TYPES>::from_genesis(TestInstanceState {}).unwrap();
+                // We assign node's public key and stake value rather than read from config file since it's a test
+                let validator_config =
+                    ValidatorConfig::generated_from_seed_indexed([0u8; 32], node_id, 1);
+                let hotshot = Self::add_node_with_config(
                     node_id,
-                    networks,
-                    handle: hotshot.run_tasks().await,
-                });
+                    networks.clone(),
+                    storage,
+                    memberships,
+                    initializer,
+                    config,
+                    validator_config,
+                )
+                .await;
+                if late_start.contains(&node_id) {
+                    self.late_start.insert(
+                        node_id,
+                        LateStartNode {
+                            networks,
+                            context: Left(hotshot),
+                        },
+                    );
+                } else {
+                    self.nodes.push(Node {
+                        node_id,
+                        networks,
+                        handle: hotshot.run_tasks().await,
+                    });
+                }
             }
             results.push(node_id);
         }
@@ -349,48 +401,24 @@ where
     /// # Panics
     /// if unable to initialize the node's `SystemContext` based on the config
     pub async fn add_node_with_config(
-        &mut self,
+        node_id: u64,
         networks: Networks<TYPES, I>,
         storage: I::Storage,
+        memberships: Memberships<TYPES>,
         initializer: HotShotInitializer<TYPES>,
         config: HotShotConfig<TYPES::SignatureKey, TYPES::ElectionConfigType>,
         validator_config: ValidatorConfig<TYPES::SignatureKey>,
     ) -> Arc<SystemContext<TYPES, I>> {
-        let node_id = self.next_node_id;
-        self.next_node_id += 1;
-        let known_nodes_with_stake = config.known_nodes_with_stake.clone();
         // Get key pair for certificate aggregation
         let private_key = validator_config.private_key.clone();
         let public_key = validator_config.public_key.clone();
-        let quorum_election_config = config.election_config.clone().unwrap_or_else(|| {
-            TYPES::Membership::default_election_config(config.total_nodes.get() as u64)
-        });
-        let committee_election_config = I::committee_election_config_generator();
+
         let network_bundle = hotshot::Networks {
             quorum_network: networks.0.clone(),
             da_network: networks.1.clone(),
             _pd: PhantomData,
         };
 
-        let memberships = Memberships {
-            quorum_membership: <TYPES as NodeType>::Membership::create_election(
-                known_nodes_with_stake.clone(),
-                quorum_election_config.clone(),
-            ),
-            da_membership: <TYPES as NodeType>::Membership::create_election(
-                known_nodes_with_stake.clone(),
-                committee_election_config(config.da_committee_size as u64),
-            ),
-            vid_membership: <TYPES as NodeType>::Membership::create_election(
-                known_nodes_with_stake.clone(),
-                quorum_election_config.clone(),
-            ),
-            view_sync_membership: <TYPES as NodeType>::Membership::create_election(
-                known_nodes_with_stake.clone(),
-                quorum_election_config,
-            ),
-        };
-
         SystemContext::new(
             public_key,
             private_key,
diff --git a/crates/testing/tests/catchup.rs b/crates/testing/tests/catchup.rs
index eae668139a..339103fb1b 100644
--- a/crates/testing/tests/catchup.rs
+++ b/crates/testing/tests/catchup.rs
@@ -236,3 +236,67 @@ async fn test_catchup_in_view_sync() {
         .run_test()
         .await;
 }
+
+// Almost the same as `test_catchup`, but with catchup nodes reloaded from anchor leaf rather than
+// initialized from genesis.
+#[cfg(test)]
+#[cfg_attr(
+    async_executor_impl = "tokio",
+    tokio::test(flavor = "multi_thread", worker_threads = 2)
+)]
+#[cfg_attr(async_executor_impl = "async-std", async_std::test)]
+async fn test_catchup_reload() {
+    use std::time::Duration;
+
+    use hotshot_example_types::node_types::{MemoryImpl, TestTypes};
+    use hotshot_testing::{
+        completion_task::{CompletionTaskDescription, TimeBasedCompletionTaskDescription},
+        overall_safety_task::OverallSafetyPropertiesDescription,
+        spinning_task::{ChangeNode, SpinningTaskDescription, UpDown},
+        test_builder::{TestMetadata, TimingData},
+    };
+
+    async_compatibility_layer::logging::setup_logging();
+    async_compatibility_layer::logging::setup_backtrace();
+    let timing_data = TimingData {
+        next_view_timeout: 2000,
+        ..Default::default()
+    };
+    let mut metadata = TestMetadata::default();
+    let catchup_node = vec![ChangeNode {
+        idx: 19,
+        updown: UpDown::Up,
+    }];
+
+    metadata.timing_data = timing_data;
+    metadata.start_nodes = 19;
+    metadata.skip_late = true;
+    metadata.total_nodes = 20;
+
+    metadata.view_sync_properties =
+        hotshot_testing::view_sync_task::ViewSyncTaskDescription::Threshold(0, 20);
+
+    metadata.spinning_properties = SpinningTaskDescription {
+        // Start the nodes before their leadership.
+        node_changes: vec![(13, catchup_node)],
+    };
+
+    metadata.completion_task_description =
+        CompletionTaskDescription::TimeBasedCompletionTaskBuilder(
+            TimeBasedCompletionTaskDescription {
+                duration: Duration::from_secs(60),
+            },
+        );
+    metadata.overall_safety_properties = OverallSafetyPropertiesDescription {
+        // Make sure we keep commiting rounds after the catchup, but not the full 50.
+        num_successful_views: 22,
+        check_leaf: true,
+        ..Default::default()
+    };
+
+    metadata
+        .gen_launcher::<TestTypes, MemoryImpl>(0)
+        .launch()
+        .run_test()
+        .await;
+}
diff --git a/crates/testing/tests/consensus_task.rs b/crates/testing/tests/consensus_task.rs
index 464e5ba073..ff960e667f 100644
--- a/crates/testing/tests/consensus_task.rs
+++ b/crates/testing/tests/consensus_task.rs
@@ -119,7 +119,8 @@ async fn test_consensus_task() {
         TestTypes,
         MemoryImpl,
         SystemContextHandle<TestTypes, MemoryImpl>,
-    >::create_from(&handle);
+    >::create_from(&handle)
+    .await;
 
     inject_consensus_polls(&consensus_state).await;
 
@@ -169,7 +170,8 @@ async fn test_consensus_vote() {
         TestTypes,
         MemoryImpl,
         SystemContextHandle<TestTypes, MemoryImpl>,
-    >::create_from(&handle);
+    >::create_from(&handle)
+    .await;
 
     inject_consensus_polls(&consensus_state).await;
 
@@ -289,7 +291,8 @@ async fn test_consensus_with_vid() {
         TestTypes,
         MemoryImpl,
         SystemContextHandle<TestTypes, MemoryImpl>,
-    >::create_from(&handle);
+    >::create_from(&handle)
+    .await;
 
     inject_consensus_polls(&consensus_state).await;
 
diff --git a/crates/testing/tests/da_task.rs b/crates/testing/tests/da_task.rs
index 757ac926a5..fe06a34543 100644
--- a/crates/testing/tests/da_task.rs
+++ b/crates/testing/tests/da_task.rs
@@ -88,6 +88,6 @@ async fn test_da_task() {
     .expect("Failed to sign DAData");
     output.insert(HotShotEvent::DAVoteSend(da_vote), 1);
 
-    let da_state = DATaskState::<TestTypes, MemoryImpl, SystemContextHandle<TestTypes, MemoryImpl>>::create_from(&handle);
+    let da_state = DATaskState::<TestTypes, MemoryImpl, SystemContextHandle<TestTypes, MemoryImpl>>::create_from(&handle).await;
     run_harness(input, output, da_state, false).await;
 }
diff --git a/crates/testing/tests/view_sync_task.rs b/crates/testing/tests/view_sync_task.rs
index f32c99f09b..f0d5679273 100644
--- a/crates/testing/tests/view_sync_task.rs
+++ b/crates/testing/tests/view_sync_task.rs
@@ -52,6 +52,7 @@ async fn test_view_sync_task() {
         TestTypes,
         MemoryImpl,
         SystemContextHandle<TestTypes, MemoryImpl>,
-    >::create_from(&handle);
+    >::create_from(&handle)
+    .await;
     run_harness(input, output, view_sync_state, false).await;
 }
diff --git a/crates/types/src/consensus.rs b/crates/types/src/consensus.rs
index 02e51df512..3a1c1e06d5 100644
--- a/crates/types/src/consensus.rs
+++ b/crates/types/src/consensus.rs
@@ -39,7 +39,7 @@ pub struct Consensus<TYPES: NodeType> {
     /// view -> DA cert
     pub saved_da_certs: HashMap<TYPES::Time, DACertificate<TYPES>>,
 
-    /// cur_view from pseudocode
+    /// View number that is currently on.
     pub cur_view: TYPES::Time,
 
     /// last view had a successful decide event
@@ -235,11 +235,9 @@ impl Default for ConsensusMetricsValue {
 }
 
 impl<TYPES: NodeType> Consensus<TYPES> {
-    /// increment the current view
-    /// NOTE may need to do gc here
-    pub fn increment_view(&mut self) -> TYPES::Time {
-        self.cur_view += 1;
-        self.cur_view
+    /// Update the current view.
+    pub fn update_view(&mut self, view_number: TYPES::Time) {
+        self.cur_view = view_number;
     }
 
     /// gather information from the parent chain of leafs