Allow Proposal requesting from any DA node

crates/task-impls/src/consensus/handlers.rs

@@ -137,11 +137,6 @@ pub async fn create_and_send_proposal<TYPES: NodeType, V: Versions>(
         proposed_leaf.view_number(),
     );
 
-    consensus
-        .write()
-        .await
-        .update_last_proposed_view(message.clone())?;
-
     async_sleep(Duration::from_millis(round_start_delay)).await;
 
     broadcast_event(

crates/task-impls/src/consensus/mod.rs

@@ -420,7 +420,7 @@ impl<TYPES: NodeType, I: NodeImplementation<TYPES>, V: Versions> ConsensusTaskSt
                     }
 
                     if let Err(e) = self.consensus.write().await.update_high_qc(qc.clone()) {
-                        tracing::error!("{e:?}");
+                        tracing::trace!("{e:?}");
                     }
                     debug!(
                         "Attempting to publish proposal after forming a QC for view {}",

crates/task-impls/src/helpers.rs

@@ -469,24 +469,26 @@ pub async fn validate_proposal_safety_and_liveness<TYPES: NodeType, V: Versions>
         },
     };
 
-    if let Err(e) = consensus
-        .write()
-        .await
-        .update_validated_state_map(view_number, view.clone())
     {
-        tracing::trace!("{e:?}");
-    }
-    consensus
-        .write()
-        .await
-        .update_saved_leaves(proposed_leaf.clone());
+        let mut consensus_write = consensus.write().await;
+        if let Err(e) = consensus_write.update_validated_state_map(view_number, view.clone()) {
+            tracing::trace!("{e:?}");
+        }
+        consensus_write.update_saved_leaves(proposed_leaf.clone());
 
-    // Broadcast that we've updated our consensus state so that other tasks know it's safe to grab.
-    broadcast_event(
-        Arc::new(HotShotEvent::ValidatedStateUpdated(view_number, view)),
-        &event_stream,
-    )
-    .await;
+        // Update our internal storage of the proposal. The proposal is valid, so
+        // we swallow this error and just log if it occurs.
+        if let Err(e) = consensus_write.update_last_proposed_view(proposal.clone()) {
+            tracing::debug!("Internal proposal update failed; error = {e:#}");
+        };
+
+        // Broadcast that we've updated our consensus state so that other tasks know it's safe to grab.
+        broadcast_event(
+            Arc::new(HotShotEvent::ValidatedStateUpdated(view_number, view)),
+            &event_stream,
+        )
+        .await;
+    }
 
     UpgradeCertificate::validate(
         &proposal.data.upgrade_certificate,
@@ -505,37 +507,39 @@ pub async fn validate_proposal_safety_and_liveness<TYPES: NodeType, V: Versions>
     // passes.
 
     // Liveness check.
-    let read_consensus = consensus.read().await;
-    let liveness_check = justify_qc.view_number() > read_consensus.locked_view();
-
-    // Safety check.
-    // Check if proposal extends from the locked leaf.
-    let outcome = read_consensus.visit_leaf_ancestors(
-        justify_qc.view_number(),
-        Terminator::Inclusive(read_consensus.locked_view()),
-        false,
-        |leaf, _, _| {
-            // if leaf view no == locked view no then we're done, report success by
-            // returning true
-            leaf.view_number() != read_consensus.locked_view()
-        },
-    );
-    let safety_check = outcome.is_ok();
-
-    ensure!(safety_check || liveness_check, {
-        if let Err(e) = outcome {
-            broadcast_event(
-                Event {
-                    view_number,
-                    event: EventType::Error { error: Arc::new(e) },
-                },
-                &event_sender,
-            )
-            .await;
-        }
+    {
+        let read_consensus = consensus.read().await;
+        let liveness_check = justify_qc.view_number() > read_consensus.locked_view();
+
+        // Safety check.
+        // Check if proposal extends from the locked leaf.
+        let outcome = read_consensus.visit_leaf_ancestors(
+            justify_qc.view_number(),
+            Terminator::Inclusive(read_consensus.locked_view()),
+            false,
+            |leaf, _, _| {
+                // if leaf view no == locked view no then we're done, report success by
+                // returning true
+                leaf.view_number() != read_consensus.locked_view()
+            },
+        );
+        let safety_check = outcome.is_ok();
+
+        ensure!(safety_check || liveness_check, {
+            if let Err(e) = outcome {
+                broadcast_event(
+                    Event {
+                        view_number,
+                        event: EventType::Error { error: Arc::new(e) },
+                    },
+                    &event_sender,
+                )
+                .await;
+            }
 
-        format!("Failed safety and liveness check \n High QC is {:?}  Proposal QC is {:?}  Locked view is {:?}", read_consensus.high_qc(), proposal.data.clone(), read_consensus.locked_view())
-    });
+            format!("Failed safety and liveness check \n High QC is {:?}  Proposal QC is {:?}  Locked view is {:?}", read_consensus.high_qc(), proposal.data.clone(), read_consensus.locked_view())
+        });
+    }
 
     // We accept the proposal, notify the application layer
 
@@ -550,6 +554,7 @@ pub async fn validate_proposal_safety_and_liveness<TYPES: NodeType, V: Versions>
         &event_sender,
     )
     .await;
+
     // Notify other tasks
     broadcast_event(
         Arc::new(HotShotEvent::QuorumProposalValidated(

crates/task-impls/src/network.rs

@@ -315,7 +315,7 @@ impl<
                     MessageKind::<TYPES>::from_consensus_message(SequencingMessage::General(
                         GeneralConsensusMessage::ProposalRequested(req.clone(), signature),
                     )),
-                    TransmitType::Direct(membership.leader(req.view_number)),
+                    TransmitType::DaCommitteeBroadcast,
                 ),
                 HotShotEvent::QuorumProposalResponseSend(sender_key, proposal) => (
                     sender_key.clone(),

crates/task-impls/src/quorum_proposal/handlers.rs

@@ -229,10 +229,6 @@ impl<TYPES: NodeType, V: Versions> ProposalDependencyHandle<TYPES, V> {
             proposed_leaf.view_number(),
         );
 
-        self.consensus
-            .write()
-            .await
-            .update_last_proposed_view(message.clone())?;
         async_sleep(Duration::from_millis(self.round_start_delay)).await;
         broadcast_event(
             Arc::new(HotShotEvent::QuorumProposalSend(

crates/task-impls/src/request.rs

@@ -88,6 +88,7 @@ type Signature<TYPES> =
 impl<TYPES: NodeType, I: NodeImplementation<TYPES>> TaskState for NetworkRequestState<TYPES, I> {
     type Event = HotShotEvent<TYPES>;
 
+    #[instrument(skip_all, target = "NetworkRequestState", fields(id = self.id))]
     async fn handle_event(
         &mut self,
         event: Arc<Self::Event>,