Fix checkpoint catch up for transactions not in extra_transactions (M…

…ystenLabs#4281)

When catching up to checkpoint we want to be able to move to next checkpoint even when transactions aren't in extra_transaction. This was partially done in MystenLabs#4178 and # 4221 but there were few problems introduced previously, and this PR fixes them, namely:

- We now store all transactions in `transactions_to_checkpoint` when catching up to checkpoint right away. To achieve this we remove txn sequence number from transactions_to_checkpoint table as it is not used(and if needed is present in other databases)

- We now can receive transactions from batch service that are already in checkpoint(could not happen before as checkpoint always followed batch service). We add double check to `update_processed_transactions` to exclude those. There is no concurrency issues here since `update_processed_transactions` takes &mut self.

crates/sui-core/src/checkpoints/mod.rs

@@ -94,8 +94,7 @@ pub struct CheckpointStoreTables {
     /// The list of all transaction/effects that are checkpointed mapping to the checkpoint
     /// sequence number they were assigned to.
     #[default_options_override_fn = "transactions_to_checkpoint_table_default_config"]
-    pub transactions_to_checkpoint:
-        DBMap<ExecutionDigests, (CheckpointSequenceNumber, TxSequenceNumber)>,
+    pub transactions_to_checkpoint: DBMap<ExecutionDigests, CheckpointSequenceNumber>,
 
     /// The mapping from checkpoint to transaction/effects contained within the checkpoint.
     /// The checkpoint content should be causally ordered and is consistent among
@@ -975,14 +974,7 @@ impl CheckpointStore {
 
         // Now write the checkpoint data to the database
 
-        let transactions_to_checkpoint: Vec<_> = transactions
-            .iter()
-            .zip(transactions_with_seq.iter())
-            .filter_map(|(tx, opt)| {
-                // If iseq is missing here then batch service will update this index in update_processed_transactions
-                opt.as_ref().map(|iseq| (*tx, (seq, *iseq)))
-            })
-            .collect();
+        let transactions_to_checkpoint: Vec<_> = transactions.iter().map(|tx| (*tx, seq)).collect();
 
         let batch = batch.insert_batch(
             &self.tables.transactions_to_checkpoint,
@@ -1007,9 +999,22 @@ impl CheckpointStore {
         transactions: &[(TxSequenceNumber, ExecutionDigests)],
     ) -> Result<(), SuiError> {
         let batch = self.tables.extra_transactions.batch();
+        let already_in_checkpoint = self
+            .tables
+            .transactions_to_checkpoint
+            .multi_get(transactions.iter().map(|(_seq, digest)| *digest))?;
         let batch = batch.insert_batch(
             &self.tables.extra_transactions,
-            transactions.iter().map(|(seq, digest)| (digest, seq)),
+            transactions
+                .iter()
+                .zip(already_in_checkpoint.iter())
+                .filter_map(|((seq, digest), cpk)| {
+                    if cpk.is_some() {
+                        None
+                    } else {
+                        Some((digest, seq))
+                    }
+                }),
         )?;
 
         // Write to the database.

crates/sui-core/src/checkpoints/tests/checkpoint_tests.rs

@@ -224,13 +224,13 @@ fn make_checkpoint_db() {
     assert_eq!(cps.tables.checkpoint_contents.iter().count(), 1);
     assert_eq!(cps.tables.extra_transactions.iter().count(), 2); // t3 & t6
 
-    let (_cp_seq, tx_seq) = cps
+    let cp_seq = cps
         .tables
         .transactions_to_checkpoint
         .get(&t4)
         .unwrap()
         .unwrap();
-    assert_eq!(tx_seq, 4);
+    assert_eq!(cp_seq, 0);
 }
 
 #[test]