[authority sync] Provide an interface for explorers to sync with single authority

sui/src/bench.rs

@@ -167,7 +167,7 @@ impl ClientServerBenchmark {
             let state = AuthorityState::new(
                 committee.clone(),
                 public_auth0,
-                Box::pin(secret_auth0),
+                Arc::pin(secret_auth0),
                 store,
                 genesis::clone_genesis_compiled_modules(),
                 &mut genesis::get_genesis_context(),

sui/src/sui_commands.rs

@@ -263,7 +263,7 @@ async fn make_server_with_genesis_ctx(
     let state = AuthorityState::new(
         committee.clone(),
         name,
-        Box::pin(authority.key_pair.copy()),
+        Arc::pin(authority.key_pair.copy()),
         store,
         preload_modules,
         genesis_ctx,

sui_core/src/authority.rs

@@ -27,9 +27,11 @@ use sui_types::{
     MOVE_STDLIB_ADDRESS, SUI_FRAMEWORK_ADDRESS,
 };
 
+use crate::authority_batch::BatchSender;
+
 #[cfg(test)]
 #[path = "unit_tests/authority_tests.rs"]
-mod authority_tests;
+pub mod authority_tests;
 
 mod temporary_store;
 use temporary_store::AuthorityTemporaryStore;
@@ -46,7 +48,8 @@ const MAX_GAS_BUDGET: u64 = 18446744073709551615 / 1000 - 1;
 ///
 /// Typically instantiated with Box::pin(keypair) where keypair is a `KeyPair`
 ///
-type StableSyncAuthoritySigner = Pin<Box<dyn signature::Signer<AuthoritySignature> + Send + Sync>>;
+pub type StableSyncAuthoritySigner =
+    Pin<Arc<dyn signature::Signer<AuthoritySignature> + Send + Sync>>;
 
 pub struct AuthorityState {
     // Fixed size, static, identity of the authority
@@ -63,6 +66,11 @@ pub struct AuthorityState {
 
     /// The database
     _database: Arc<AuthorityStore>,
+
+    /// The sender to notify of new transactions
+    /// and create batches for this authority.
+    /// Keep as None if there is no need for this.
+    batch_sender: Option<BatchSender>,
 }
 
 /// The authority state encapsulates all state, drives execution, and ensures safety.
@@ -72,6 +80,16 @@ pub struct AuthorityState {
 ///
 /// Repeating valid commands should produce no changes and return no error.
 impl AuthorityState {
+    /// Set a listener for transaction certificate updates. Returns an
+    /// error if a listener is already registered.
+    pub fn set_batch_sender(&mut self, batch_sender: BatchSender) -> SuiResult {
+        if self.batch_sender.is_some() {
+            return Err(SuiError::AuthorityUpdateFailure);
+        }
+        self.batch_sender = Some(batch_sender);
+        Ok(())
+    }
+
     /// The logic to check one object against a reference, and return the object if all is well
     /// or an error if not.
     fn check_one_lock(
@@ -381,8 +399,17 @@ impl AuthorityState {
             &gas_object_id,
         );
         // Update the database in an atomic manner
-        self.update_state(temporary_store, certificate, to_signed_effects)
-            .await // Returns the TransactionInfoResponse
+
+        let (seq, resp) = self
+            .update_state(temporary_store, certificate, to_signed_effects)
+            .await?; // Returns the OrderInfoResponse
+
+        // If there is a notifier registered, notify:
+        if let Some(sender) = &self.batch_sender {
+            sender.send_item(seq, transaction_digest).await?;
+        }
+
+        Ok(resp)
     }
 
     fn execute_transaction(
@@ -616,6 +643,7 @@ impl AuthorityState {
             move_vm: adapter::new_move_vm(native_functions)
                 .expect("We defined natives to not fail here"),
             _database: store,
+            batch_sender: None,
         };
 
         for genesis_modules in genesis_packages {
@@ -627,6 +655,11 @@ impl AuthorityState {
         state
     }
 
+    #[cfg(test)]
+    pub fn db(&self) -> Arc<AuthorityStore> {
+        self._database.clone()
+    }
+
     async fn get_object(&self, object_id: &ObjectID) -> Result<Option<Object>, SuiError> {
         self._database.get_object(object_id)
     }
@@ -708,9 +741,10 @@ impl AuthorityState {
     async fn update_state(
         &self,
         temporary_store: AuthorityTemporaryStore,
+
         certificate: CertifiedTransaction,
         signed_effects: SignedTransactionEffects,
-    ) -> Result<TransactionInfoResponse, SuiError> {
+    ) -> Result<(u64, TransactionInfoResponse), SuiError> {
         self._database
             .update_state(temporary_store, certificate, signed_effects)
     }

sui_core/src/authority/authority_store.rs

@@ -6,10 +6,17 @@ use rocksdb::Options;
 use std::collections::BTreeSet;
 use std::convert::TryInto;
 use std::path::Path;
+
+use std::sync::atomic::AtomicU64;
 use sui_types::base_types::SequenceNumber;
 use typed_store::rocks::{open_cf, DBBatch, DBMap};
+
+use std::sync::atomic::Ordering;
 use typed_store::traits::Map;
 
+pub use crate::authority_batch::AuthorityBatch;
+use crate::authority_batch::{SignedBatch, TxSequenceNumber};
+
 pub struct AuthorityStore {
     /// This is a map between the object ID and the latest state of the object, namely the
     /// state that is needed to process new transactions. If an object is deleted its entry is
@@ -61,6 +68,16 @@ pub struct AuthorityStore {
 
     /// Internal vector of locks to manage concurrent writes to the database
     lock_table: Vec<parking_lot::Mutex<()>>,
+
+    // Tables used for authority batch structure
+    /// A sequence on all executed certificates and effects.
+    pub executed_sequence: DBMap<TxSequenceNumber, TransactionDigest>,
+
+    /// A sequence of batches indexing into the sequence of executed transactions.
+    pub batches: DBMap<TxSequenceNumber, SignedBatch>,
+
+    /// The next available sequence number to use in the `executed sequence` table.
+    pub next_sequence_number: AtomicU64,
 }
 
 impl AuthorityStore {
@@ -79,9 +96,27 @@ impl AuthorityStore {
                 "signed_effects",
                 "sequenced",
                 "schedule",
+                "executed_sequence",
+                "batches",
             ],
         )
         .expect("Cannot open DB.");
+
+        let executed_sequence =
+            DBMap::reopen(&db, Some("executed_sequence")).expect("Cannot open CF.");
+
+        // Read the index of the last entry in the sequence of commands
+        // to extract the next sequence number or it is zero.
+        let next_sequence_number = AtomicU64::new(
+            executed_sequence
+                .iter()
+                .skip_prior_to(&TxSequenceNumber::MAX)
+                .expect("Error reading table.")
+                .next()
+                .map(|(v, _)| v + 1u64)
+                .unwrap_or(0),
+        );
+
         AuthorityStore {
             objects: DBMap::reopen(&db, Some("objects")).expect("Cannot open CF."),
             owner_index: DBMap::reopen(&db, Some("owner_index")).expect("Cannot open CF."),
@@ -98,6 +133,9 @@ impl AuthorityStore {
                 .into_iter()
                 .map(|_| parking_lot::Mutex::new(()))
                 .collect(),
+            executed_sequence,
+            batches: DBMap::reopen(&db, Some("batches")).expect("Cannot open CF."),
+            next_sequence_number,
         }
     }
 
@@ -330,7 +368,7 @@ impl AuthorityStore {
         temporary_store: AuthorityTemporaryStore,
         certificate: CertifiedTransaction,
         signed_effects: SignedTransactionEffects,
-    ) -> Result<TransactionInfoResponse, SuiError> {
+    ) -> Result<(TxSequenceNumber, TransactionInfoResponse), SuiError> {
         // Extract the new state from the execution
         // TODO: events are already stored in the TxDigest -> TransactionEffects store. Is that enough?
         let mut write_batch = self.transaction_lock.batch();
@@ -348,13 +386,19 @@ impl AuthorityStore {
             std::iter::once((transaction_digest, &signed_effects)),
         )?;
 
-        self.batch_update_objects(write_batch, temporary_store, transaction_digest)?;
-
-        Ok(TransactionInfoResponse {
-            signed_transaction: self.signed_transactions.get(&transaction_digest)?,
-            certified_transaction: Some(certificate),
-            signed_effects: Some(signed_effects),
-        })
+        // Safe to unwrap since the "true" flag ensures we get a sequence value back.
+        let seq: TxSequenceNumber = self
+            .batch_update_objects(write_batch, temporary_store, transaction_digest, true)?
+            .unwrap();
+
+        Ok((
+            seq,
+            TransactionInfoResponse {
+                signed_transaction: self.signed_transactions.get(&transaction_digest)?,
+                certified_transaction: Some(certificate),
+                signed_effects: Some(signed_effects),
+            },
+        ))
     }
 
     /// Persist temporary storage to DB for genesis modules
@@ -365,7 +409,8 @@ impl AuthorityStore {
     ) -> Result<(), SuiError> {
         debug_assert_eq!(transaction_digest, TransactionDigest::genesis());
         let write_batch = self.transaction_lock.batch();
-        self.batch_update_objects(write_batch, temporary_store, transaction_digest)
+        self.batch_update_objects(write_batch, temporary_store, transaction_digest, false)
+            .map(|_| ())
     }
 
     /// Helper function for updating the objects in the state
@@ -374,7 +419,8 @@ impl AuthorityStore {
         mut write_batch: DBBatch,
         temporary_store: AuthorityTemporaryStore,
         transaction_digest: TransactionDigest,
-    ) -> Result<(), SuiError> {
+        should_sequence: bool,
+    ) -> Result<Option<TxSequenceNumber>, SuiError> {
         let (objects, active_inputs, written, deleted, _events) = temporary_store.into_inner();
 
         // Archive the old lock.
@@ -454,6 +500,7 @@ impl AuthorityStore {
 
         // This is the critical region: testing the locks and writing the
         // new locks must be atomic, and no writes should happen in between.
+        let mut return_seq = None;
         {
             // Acquire the lock to ensure no one else writes when we are in here.
             let _mutexes = self.acquire_locks(&active_inputs[..]);
@@ -465,13 +512,30 @@ impl AuthorityStore {
                 object_lock.ok_or(SuiError::TransactionLockDoesNotExist)?;
             }
 
+            if should_sequence {
+                // Now we are sure we are going to execute, add to the sequence
+                // number and insert into authority sequence.
+                //
+                // NOTE: it is possible that we commit to the database transactions
+                //       out of order with respect to their sequence number. It is also
+                //       possible for the authority to crash without committing the
+                //       full sequence, and the batching logic needs to deal with this.
+                let next_seq = self.next_sequence_number.fetch_add(1, Ordering::SeqCst);
+                write_batch = write_batch.insert_batch(
+                    &self.executed_sequence,
+                    std::iter::once((next_seq, transaction_digest)),
+                )?;
+
+                return_seq = Some(next_seq);
+            }
+
             // Atomic write of all locks & other data
             write_batch.write()?;
 
             // implicit: drop(_mutexes);
         } // End of critical region
 
-        Ok(())
+        Ok(return_seq)
     }
 
     /// Returns the last entry we have for this object in the parents_sync index used