fix(resharding): balance checker for ignored delayed receipts (#12619)

Tinkering with resharding testloop in another PR, I got some errors when
creating / removing an account.
One of these errors is balance checker failing.

For delayed receipt test, we use `call_burn_gas_contract` that does not
attach deposit, so balance checker did not fail.
But if a delayed receipt has deposit and it is skipped (because it
belongs to the sibling shard), then balance checker fails,
because it calculated total deposit independently of what delayed
receipts were actually processed.

---------

Co-authored-by: Andrea <[email protected]>

integration-tests/src/test_loop/tests/resharding_v3.rs

@@ -331,7 +331,7 @@ fn call_burn_gas_contract(
                         signer_id.clone(),
                         receiver_id.clone(),
                         &signer,
-                        0,
+                        1,
                         method_name,
                         args,
                         gas_burnt_per_call + 10 * TGAS,
@@ -401,7 +401,7 @@ fn call_promise_yield(
                             signer_id.clone(),
                             receiver_id.clone(),
                             &signer,
-                            0,
+                            1,
                             "call_yield_resume_read_data_id_from_storage".to_string(),
                             yield_payload.clone(),
                             300 * TGAS,

runtime/runtime/src/balance_checker.rs

@@ -275,6 +275,7 @@ pub(crate) fn check_balance(
     final_state: &TrieUpdate,
     validator_accounts_update: &Option<ValidatorAccountsUpdate>,
     incoming_receipts: &[Receipt],
+    processed_delayed_receipts: &[Receipt],
     yield_timeout_receipts: &[Receipt],
     transactions: &[SignedTransaction],
     outgoing_receipts: &[Receipt],
@@ -288,11 +289,6 @@ pub(crate) fn check_balance(
     let final_delayed_receipt_indices: DelayedReceiptIndices =
         get(final_state, &TrieKey::DelayedReceiptIndices)?.unwrap_or_default();
 
-    // Previously delayed receipts that were processed this time.
-    let processed_delayed_receipts = get_delayed_receipts(
-        initial_state,
-        initial_delayed_receipt_indices.first_index..final_delayed_receipt_indices.first_index,
-    )?;
     // Receipts that were not processed this time and are delayed now.
     let new_delayed_receipts = get_delayed_receipts(
         final_state,
@@ -427,6 +423,7 @@ mod tests {
             &[],
             &[],
             &[],
+            &[],
             &ApplyStats::default(),
         )
         .unwrap();
@@ -445,6 +442,7 @@ mod tests {
             &[],
             &[],
             &[],
+            &[],
             &ApplyStats::default(),
         )
         .unwrap_err();
@@ -509,6 +507,7 @@ mod tests {
             &[],
             &[],
             &[],
+            &[],
             &ApplyStats::default(),
         )
         .unwrap();
@@ -555,6 +554,7 @@ mod tests {
             &None,
             &[],
             &[],
+            &[],
             &[tx],
             &[receipt],
             &ApplyStats {
@@ -627,6 +627,7 @@ mod tests {
                 &None,
                 &[receipt],
                 &[],
+                &[],
                 &[tx],
                 &[],
                 &ApplyStats::default(),
@@ -669,6 +670,7 @@ mod tests {
                 &None,
                 &[receipt],
                 &[],
+                &[],
                 &[tx],
                 &[],
                 &ApplyStats::default(),
@@ -748,6 +750,7 @@ mod tests {
             &None,
             &[],
             &[],
+            &[],
             &[tx],
             &[],
             &ApplyStats {
@@ -819,6 +822,7 @@ mod tests {
             &[],
             &[],
             &[],
+            &[],
             &outgoing_receipts,
             &ApplyStats::default(),
         )
@@ -882,6 +886,7 @@ mod tests {
             &[],
             &[],
             &[],
+            &[],
             &outgoing_receipts,
             &ApplyStats::default(),
         );

runtime/runtime/src/lib.rs

@@ -2071,7 +2071,8 @@ impl Runtime {
         let apply_state = processing_state.apply_state;
         let epoch_info_provider = processing_state.epoch_info_provider;
         let mut state_update = processing_state.state_update;
-        let delayed_receipts = processing_state.delayed_receipts;
+        let pending_delayed_receipts = processing_state.delayed_receipts;
+        let processed_delayed_receipts = process_receipts_result.processed_delayed_receipts;
         let promise_yield_result = process_receipts_result.promise_yield_result;
 
         if promise_yield_result.promise_yield_indices
@@ -2086,10 +2087,10 @@ impl Runtime {
 
         // Congestion info needs a final touch to select an allowed shard if
         // this shard is fully congested.
-        let delayed_receipts_count = delayed_receipts.upper_bound_len();
+        let delayed_receipts_count = pending_delayed_receipts.upper_bound_len();
         let mut own_congestion_info = receipt_sink.own_congestion_info();
         if let Some(congestion_info) = &mut own_congestion_info {
-            delayed_receipts.apply_congestion_changes(congestion_info)?;
+            pending_delayed_receipts.apply_congestion_changes(congestion_info)?;
             let protocol_version = apply_state.current_protocol_version;
 
             let (all_shards, shard_seed) =
@@ -2123,6 +2124,7 @@ impl Runtime {
                 &state_update,
                 validator_accounts_update,
                 processing_state.incoming_receipts,
+                &processed_delayed_receipts,
                 &promise_yield_result.timeout_receipts,
                 processing_state.transactions,
                 &receipt_sink.outgoing_receipts(),
@@ -2187,7 +2189,6 @@ impl Runtime {
             .observe(chunk_recorded_size_upper_bound / f64::max(1.0, chunk_recorded_size));
         metrics::report_recorded_column_sizes(&trie, &apply_state);
         let proof = trie.recorded_storage();
-        let processed_delayed_receipts = process_receipts_result.processed_delayed_receipts;
         let processed_yield_timeouts = promise_yield_result.processed_yield_timeouts;
         let bandwidth_scheduler_state_hash = receipt_sink
             .bandwidth_scheduler_output()