Allow empty cert when adding tx to pending execution

crates/sui-core/src/authority.rs

@@ -392,7 +392,7 @@ impl AuthorityState {
             }
         );
 
-        let tx_guard = self.acquire_tx_guard(&digest, &certificate).await?;
+        let tx_guard = self.acquire_tx_guard(&certificate).await?;
 
         if certificate.contains_shared_object() {
             self.database
@@ -422,16 +422,13 @@ impl AuthorityState {
         // to do this, since the false contention can be made arbitrarily low (no cost for 1.0 -
         // epsilon of txes) while solutions without false contention have slightly higher cost
         // for every tx.
-        let tx_guard = self.acquire_tx_guard(&digest, &certificate).await?;
+        let tx_guard = self.acquire_tx_guard(&certificate).await?;
 
         self.process_certificate(tx_guard, certificate).await
     }
 
-    async fn acquire_tx_guard<'a>(
-        &'a self,
-        digest: &TransactionDigest,
-        cert: &CertifiedTransaction,
-    ) -> SuiResult<CertTxGuard<'a>> {
+    async fn acquire_tx_guard(&self, cert: &CertifiedTransaction) -> SuiResult<CertTxGuard> {
+        let digest = cert.digest();
         match self.database.wal.begin_tx(digest, cert).await? {
             Some(g) => Ok(g),
             None => {
@@ -790,7 +787,7 @@ impl AuthorityState {
             // the transaction (ie. this transaction is the next to be executed).
             debug!("Shared-locks already assigned to {digest:?} - executing now");
 
-            let tx_guard = self.acquire_tx_guard(digest, &certificate).await?;
+            let tx_guard = self.acquire_tx_guard(&certificate).await?;
 
             return self
                 .process_certificate(tx_guard, certificate)

crates/sui-core/src/authority/authority_store.rs

@@ -290,6 +290,8 @@ impl<S: Eq + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
 
     /// Add a number of certificates to the pending transactions as well as the
     /// certificates structure if they are not already executed.
+    /// Certificates are optional, and if not provided, they will be eventually
+    /// downloaded in the execution driver.
     ///
     /// This function may be run concurrently: it increases atomically an internal index
     /// by the number of certificates passed, and then records the certificates and their
@@ -298,11 +300,11 @@ impl<S: Eq + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
     /// the same certificate may be written twice (but that is OK since it is valid.)
     pub fn add_pending_certificates(
         &self,
-        certs: Vec<(TransactionDigest, CertifiedTransaction)>,
+        certs: Vec<(TransactionDigest, Option<CertifiedTransaction>)>,
     ) -> SuiResult<()> {
         let first_index = self
             .next_pending_seq
-            .fetch_add(certs.len() as u64, std::sync::atomic::Ordering::Relaxed);
+            .fetch_add(certs.len() as u64, Ordering::Relaxed);
 
         let batch = self.pending_execution.batch();
         let batch = batch.insert_batch(
@@ -314,7 +316,9 @@ impl<S: Eq + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
         )?;
         let batch = batch.insert_batch(
             &self.certificates,
-            certs.iter().map(|(digest, cert)| (digest, cert)),
+            certs
+                .into_iter()
+                .filter_map(|(digest, cert_opt)| cert_opt.map(|cert| (digest, cert))),
         )?;
         batch.write()?;
 
@@ -1148,7 +1152,7 @@ impl<S: Eq + Serialize + for<'de> Deserialize<'de>> SuiDataStore<S> {
         let index_to_write = std::iter::once((LAST_CONSENSUS_INDEX_ADDR, consensus_index));
 
         // Schedule the certificate for execution
-        self.add_pending_certificates(vec![(transaction_digest, certificate.clone())])?;
+        self.add_pending_certificates(vec![(transaction_digest, Some(certificate.clone()))])?;
         // Note: if we crash here we are not in an inconsistent state since
         //       it is ok to just update the pending list without updating the sequence.
 

crates/sui-core/src/authority_active/execution_driver/mod.rs

@@ -24,7 +24,12 @@ impl PendCertificateForExecution for AuthorityState {
         &self,
         certs: Vec<(TransactionDigest, CertifiedTransaction)>,
     ) -> SuiResult<()> {
-        self.database.add_pending_certificates(certs)
+        self.database.add_pending_certificates(
+            certs
+                .into_iter()
+                .map(|(digest, cert)| (digest, Some(cert)))
+                .collect(),
+        )
     }
 }
 

crates/sui-core/src/authority_active/execution_driver/tests.rs

@@ -77,7 +77,7 @@ async fn pending_exec_storage_notify() {
         .add_pending_certificates(
             certs
                 .into_iter()
-                .map(|cert| (*cert.digest(), cert))
+                .map(|cert| (*cert.digest(), Some(cert)))
                 .collect(),
         )
         .expect("Storage is ok");
@@ -163,7 +163,7 @@ async fn pending_exec_full() {
         .add_pending_certificates(
             certs
                 .into_iter()
-                .map(|cert| (*cert.digest(), cert))
+                .map(|cert| (*cert.digest(), Some(cert)))
                 .collect(),
         )
         .expect("Storage is ok");