feat(code/engine): Store proposed values in WAL

code/crates/core-consensus/src/full_proposal.rs

@@ -54,6 +54,15 @@ impl<Ctx: Context> Entry<Ctx> {
     fn full(value: Ctx::Value, validity: Validity, proposal: SignedProposal<Ctx>) -> Self {
         Entry::Full(FullProposal::new(value, validity, proposal))
     }
+
+    fn id(&self) -> Option<<Ctx::Value as Value>::Id> {
+        match self {
+            Entry::Full(p) => Some(p.builder_value.id()),
+            Entry::ProposalOnly(p) => Some(p.value().id()),
+            Entry::ValueOnly(v, _) => Some(v.id()),
+            Entry::Empty => None,
+        }
+    }
 }
 
 #[allow(clippy::derivable_impls)]
@@ -289,6 +298,15 @@ impl<Ctx: Context> FullProposalKeeper<Ctx> {
         self.store_value_at_pol_round(new_value);
     }
 
+    pub fn value_exists(&self, value: &ProposedValue<Ctx>) -> bool {
+        match self.keeper.get(&(value.height, value.round)) {
+            None => false,
+            Some(entries) => entries
+                .iter()
+                .any(|entry| entry.id() == Some(value.value.id())),
+        }
+    }
+
     fn store_value_at_value_round(&mut self, new_value: &ProposedValue<Ctx>) {
         let key = (new_value.height, new_value.round);
         let entries = self.keeper.get_mut(&key);

code/crates/core-consensus/src/handle/propose.rs

@@ -41,15 +41,21 @@ where
         validity: Validity::Valid,
     };
 
-    state.store_value(&proposed_value);
-
     #[cfg(feature = "metrics")]
     metrics.consensus_start();
 
-    perform!(
-        co,
-        Effect::WalAppend(WalEntry::ProposedValue(proposed_value), Default::default())
-    );
+    // If this is the first time we see this value, append it to the WAL
+    if !state.value_exists(&proposed_value) {
+        perform!(
+            co,
+            Effect::WalAppend(
+                WalEntry::ProposedValue(proposed_value.clone()),
+                Default::default()
+            )
+        );
+    }
+
+    state.store_value(&proposed_value);
 
     apply_driver_input(
         co,

code/crates/core-consensus/src/handle/proposed_value.rs

@@ -32,8 +32,6 @@ where
         return Ok(());
     }
 
-    state.store_value(&proposed_value);
-
     // There are two cases where we need to generate an internal Proposal message for consensus to process the full proposal:
     // a) In parts-only mode, where we do not get a Proposal message but only the proposal parts
     // b) In any mode if the proposed value was provided by Sync, where we do net get a Proposal message but only the full value and the certificate
@@ -57,11 +55,18 @@ where
     let validity = proposed_value.validity;
     let proposals = state.proposals_for_value(&proposed_value);
 
-    // Append the proposed value to the WAL, so it can be used for recovery.
-    perform!(
-        co,
-        Effect::WalAppend(WalEntry::ProposedValue(proposed_value), Default::default())
-    );
+    // If this is the first time we see this value, append it to the WAL, so it can be used for recovery.
+    if !state.value_exists(&proposed_value) {
+        perform!(
+            co,
+            Effect::WalAppend(
+                WalEntry::ProposedValue(proposed_value.clone()),
+                Default::default()
+            )
+        );
+    }
+
+    state.store_value(&proposed_value);
 
     for signed_proposal in proposals {
         debug!(

code/crates/core-consensus/src/state.rs

@@ -196,6 +196,10 @@ where
         self.full_proposal_keeper.store_proposal(new_proposal)
     }
 
+    pub fn value_exists(&mut self, new_value: &ProposedValue<Ctx>) -> bool {
+        self.full_proposal_keeper.value_exists(new_value)
+    }
+
     pub fn store_value(&mut self, new_value: &ProposedValue<Ctx>) {
         // Values for higher height should have been cached for future processing
         assert_eq!(new_value.height, self.driver.height());