[state-sync-v2] return the first write set version in db

Closes: aptos-labs#9644

execution/executor/src/fuzzing.rs

@@ -121,6 +121,10 @@ impl DbReader<DpnProto> for FakeDb {
         unimplemented!()
     }
 
+    fn get_first_write_set_version(&self) -> Result<Option<Version>> {
+        unimplemented!()
+    }
+
     fn get_transaction_outputs(
         &self,
         _start_version: Version,

json-rpc/src/tests/utils.rs

@@ -278,6 +278,10 @@ impl DbReader<DpnProto> for MockDiemDB {
         unimplemented!()
     }
 
+    fn get_first_write_set_version(&self) -> Result<Option<Version>> {
+        unimplemented!()
+    }
+
     fn get_transaction_outputs(
         &self,
         _start_version: Version,

state-sync/storage-service/server/src/tests.rs

@@ -452,6 +452,10 @@ impl DbReader<DpnProto> for MockDbReader {
         unimplemented!()
     }
 
+    fn get_first_write_set_version(&self) -> Result<Option<Version>> {
+        unimplemented!()
+    }
+
     fn get_transaction_outputs(
         &self,
         _start_version: Version,

storage/diemdb/src/lib.rs

@@ -794,6 +794,11 @@ impl DbReader<DpnProto> for DiemDB {
         })
     }
 
+    /// Get the first version that write set starts existent.
+    fn get_first_write_set_version(&self) -> Result<Option<Version>> {
+        self.transaction_store.get_first_write_set_version()
+    }
+
     /// Gets a batch of transactions for the purpose of synchronizing state to another node.
     ///
     /// This is used by the State Synchronizer module internally.

storage/diemdb/src/transaction_store/mod.rs

@@ -168,6 +168,13 @@ impl TransactionStore {
             .ok_or_else(|| DiemDbError::NotFound(format!("WriteSet at version {}", version)).into())
     }
 
+    /// Get the first version that write set starts existent.
+    pub fn get_first_write_set_version(&self) -> Result<Option<Version>> {
+        let mut iter = self.db.iter::<WriteSetSchema>(Default::default())?;
+        iter.seek_to_first();
+        iter.next().map(|res| res.map(|(v, _)| v)).transpose()
+    }
+
     /// Save executed transaction vm output given `version`
     pub fn put_write_set(
         &self,

storage/diemdb/src/transaction_store/test.rs

@@ -38,6 +38,8 @@ proptest! {
         }
         store.db.write_schemas(cs.batch).unwrap();
 
+        assert_eq!(store.get_first_write_set_version().unwrap(), Some(0));
+
         let ledger_version = txns.len() as Version - 1;
         for (ver, (txn, write_set)) in itertools::zip_eq(txns.iter(), write_sets.iter()).enumerate() {
             prop_assert_eq!(store.get_transaction(ver as Version).unwrap(), txn.clone());

storage/storage-client/src/lib.rs

@@ -180,6 +180,10 @@ impl DbReader<DpnProto> for StorageClient {
         unimplemented!()
     }
 
+    fn get_first_write_set_version(&self) -> Result<Option<Version>> {
+        unimplemented!()
+    }
+
     fn get_transaction_outputs(
         &self,
         _start_version: Version,

storage/storage-interface/src/lib.rs

@@ -211,6 +211,11 @@ pub trait DbReader<PS: ProtocolSpec>: Send + Sync {
         fetch_events: bool,
     ) -> Result<TransactionWithProof<PS::TransactionInfo>>;
 
+    /// See [`DiemDB::get_first_write_set_version`].
+    ///
+    /// [`DiemDB::get_first_write_set_version`]: ../diemdb/struct.DiemDB.html#method.get_first_write_set_version
+    fn get_first_write_set_version(&self) -> Result<Option<Version>>;
+
     /// See [`DiemDB::get_transaction_outputs`].
     ///
     /// [`DiemDB::get_transaction_outputs`]: ../diemdb/struct.DiemDB.html#method.get_transaction_outputs

storage/storage-interface/src/mock.rs

@@ -69,6 +69,10 @@ impl DbReader<DpnProto> for MockDbReaderWriter {
         unimplemented!()
     }
 
+    fn get_first_write_set_version(&self) -> Result<Option<Version>> {
+        unimplemented!()
+    }
+
     fn get_transaction_outputs(
         &self,
         _start_version: Version,