fix(consensus): include filled inputs in evidence (#930)

Description
---
fix(consensus): include filled inputs in evidence
fix(consensus): only calculate and add leader fee for COMMIT decisions
fix(consensus): ensure evidence is deduped and sorted

Motivation and Context
---
When a transaction contained 0 inputs and outputs (maybe due to ABORT)
an invariant error would trigger
`BUG Invariant error occurred: Distinct shards is zero for transaction`

This PR fixes this by including filled inputs in evidence and not adding
the fee for the transaction for ABORT.

How Has This Been Tested?
---
Manually

What process can a PR reviewer use to test or verify this change?
---
Submit a transaction with only filled inputs

Breaking Changes
---

- [x] None
- [ ] Requires data directory to be deleted
- [ ] Other - Please specify

dan_layer/consensus_tests/src/support/harness.rs

@@ -338,7 +338,7 @@ impl TestBuilder {
     }
 
     pub fn with_message_filter(mut self, message_filter: MessageFilter) -> Self {
-        self.message_filter = Some(message_filter.into());
+        self.message_filter = Some(message_filter);
         self
     }
 

dan_layer/engine/src/runtime/tracker.rs

@@ -327,9 +327,9 @@ impl StateTracker {
 
     pub fn reset_to_fee_checkpoint(&self) -> Result<(), RuntimeError> {
         let mut checkpoint = self.fee_checkpoint.lock().unwrap();
-        let fee_state = self.read_with(|state| state.fee_state().clone());
         if let Some(checkpoint) = checkpoint.take() {
             self.write_with(|state| {
+                let fee_state = state.fee_state().clone();
                 *state = checkpoint;
                 // Preserve fee state across resets
                 *state.fee_state_mut() = fee_state;

dan_layer/storage/src/consensus_models/command.rs

@@ -88,15 +88,9 @@ impl Evidence {
 
 impl FromIterator<(SubstateAddress, ShardEvidence)> for Evidence {
     fn from_iter<T: IntoIterator<Item = (SubstateAddress, ShardEvidence)>>(iter: T) -> Self {
-        Evidence {
-            evidence: iter.into_iter().collect(),
-        }
-    }
-}
-
-impl Extend<(SubstateAddress, ShardEvidence)> for Evidence {
-    fn extend<T: IntoIterator<Item = (SubstateAddress, ShardEvidence)>>(&mut self, iter: T) {
-        self.evidence.extend(iter.into_iter())
+        let mut evidence = iter.into_iter().collect::<IndexMap<_, _>>();
+        evidence.sort_keys();
+        Evidence { evidence }
     }
 }
 

dan_layer/storage/src/consensus_models/executed_transaction.rs

@@ -142,29 +142,36 @@ impl ExecutedTransaction {
     }
 
     pub fn to_initial_evidence(&self) -> Evidence {
-        let mut evidence = Evidence::empty();
-        evidence.extend(self.transaction.inputs().iter().map(|input| {
+        let mut deduped_evidence = HashMap::new();
+        deduped_evidence.extend(self.transaction.inputs().iter().map(|input| {
             (*input, ShardEvidence {
                 qc_ids: IndexSet::new(),
                 lock: LockFlag::Write,
             })
         }));
 
-        evidence.extend(self.transaction.input_refs().iter().map(|input_ref| {
+        deduped_evidence.extend(self.transaction.input_refs().iter().map(|input_ref| {
             (*input_ref, ShardEvidence {
                 qc_ids: IndexSet::new(),
                 lock: LockFlag::Read,
             })
         }));
 
-        evidence.extend(self.resulting_outputs.iter().map(|output| {
+        deduped_evidence.extend(self.transaction.filled_inputs().iter().map(|input_ref| {
+            (*input_ref, ShardEvidence {
+                qc_ids: IndexSet::new(),
+                lock: LockFlag::Write,
+            })
+        }));
+
+        deduped_evidence.extend(self.resulting_outputs.iter().map(|output| {
             (*output, ShardEvidence {
                 qc_ids: IndexSet::new(),
                 lock: LockFlag::Write,
             })
         }));
 
-        evidence
+        deduped_evidence.into_iter().collect()
     }
 
     pub fn is_finalized(&self) -> bool {

dan_layer/storage/src/consensus_models/transaction_pool.rs

@@ -309,6 +309,9 @@ impl TransactionPoolRecord {
     }
 
     pub fn calculate_leader_fee(&self, involved: NonZeroU64, exhaust_divisor: u64) -> u64 {
+        if self.current_decision().is_abort() {
+            return 0;
+        }
         // TODO: We essentially burn a random amount depending on the shards involved in the transaction. This means it
         //       is hard to tell how much is actually in circulation unless we track this in the Resource. Right
         //       now we'll set exhaust to 0, which is just transaction_fee / involved.