[aptos-debugger] Implement the CLI interface (aptos-labs#6289)

* [aptos-debugger] Implement the CLI interface

* fixup! [aptos-debugger] Implement the CLI interface

aptos-move/aptos-debugger/Cargo.toml

@@ -12,12 +12,14 @@ edition = "2021"
 [dependencies]
 anyhow = { workspace = true }
 aptos-gas = { workspace = true }
+aptos-logger = { workspace = true }
 aptos-resource-viewer = { workspace = true }
 aptos-rest-client = { workspace = true }
 aptos-state-view = { workspace = true }
 aptos-types = { workspace = true }
 aptos-validator-interface = { workspace = true }
 aptos-vm = { workspace = true }
+clap = { workspace = true }
 move-binary-format = { workspace = true }
 move-cli = { workspace = true }
 move-compiler = { workspace = true }
@@ -26,3 +28,5 @@ move-resource-viewer = { workspace = true }
 move-table-extension = { workspace = true }
 move-vm-runtime = { workspace = true }
 move-vm-test-utils = { workspace = true }
+tokio = { workspace = true }
+url = { workspace = true }

aptos-move/aptos-debugger/src/lib.rs

@@ -63,6 +63,7 @@ impl AptosDebugger {
             .debugger
             .get_committed_transactions(begin, limit)
             .await?;
+
         let mut ret = vec![];
         while limit != 0 {
             println!(

aptos-move/aptos-debugger/src/main.rs

@@ -0,0 +1,55 @@
+// Copyright (c) Aptos
+// SPDX-License-Identifier: Apache-2.0
+
+use anyhow::Result;
+use aptos_debugger::AptosDebugger;
+use aptos_rest_client::Client;
+use aptos_vm::AptosVM;
+use clap::{Parser, Subcommand};
+use std::path::PathBuf;
+use url::Url;
+
+#[derive(Subcommand)]
+pub enum Target {
+    /// Use full node's rest api as query endpoint.
+    Rest { endpoint: String },
+    /// Use a local db instance to serve as query endpoint.
+    DB { path: PathBuf },
+}
+#[derive(Parser)]
+pub struct Argument {
+    #[clap(subcommand)]
+    target: Target,
+
+    #[clap(long)]
+    begin_version: u64,
+
+    #[clap(long)]
+    limit: u64,
+
+    #[clap(long, default_value = "1")]
+    concurrency_level: usize,
+}
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    aptos_logger::Logger::new().init();
+    let args = Argument::parse();
+    AptosVM::set_concurrency_level_once(args.concurrency_level);
+
+    let debugger = match args.target {
+        Target::Rest { endpoint } => {
+            AptosDebugger::rest_client(Client::new(Url::parse(&endpoint)?))?
+        },
+        Target::DB { path } => AptosDebugger::db(path)?,
+    };
+
+    println!(
+        "{:#?}",
+        debugger
+            .execute_past_transactions(args.begin_version, args.limit)
+            .await?
+    );
+
+    Ok(())
+}

aptos-move/aptos-validator-interface/Cargo.toml

@@ -22,5 +22,6 @@ aptos-state-view = { workspace = true }
 aptos-storage-interface = { workspace = true }
 aptos-types = { workspace = true }
 async-trait = { workspace = true }
+lru = { workspace = true }
 move-binary-format = { workspace = true }
 tokio = { workspace = true }

aptos-move/aptos-validator-interface/src/lib.rs

@@ -18,11 +18,9 @@ use aptos_types::{
     },
     transaction::{Transaction, Version},
 };
+use lru::LruCache;
 use move_binary_format::file_format::CompiledModule;
-use std::sync::{
-    mpsc::{channel, Receiver, Sender},
-    Arc, Mutex,
-};
+use std::sync::{Arc, Mutex};
 use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};
 
 // TODO(skedia) Clean up this interfact to remove account specific logic and move to state store
@@ -115,36 +113,58 @@ pub trait AptosValidatorInterface: Sync {
 }
 
 pub struct DebuggerStateView {
-    query_sender: Mutex<UnboundedSender<(StateKey, Version)>>,
-    query_receiver: Mutex<Receiver<Result<Option<StateValue>>>>,
+    query_sender:
+        Mutex<UnboundedSender<(StateKey, Version, std::sync::mpsc::Sender<Option<Vec<u8>>>)>>,
     version: Version,
 }
 
 async fn handler_thread<'a>(
     db: Arc<dyn AptosValidatorInterface + Send>,
-    mut thread_receiver: UnboundedReceiver<(StateKey, Version)>,
-    thread_sender: Sender<Result<Option<StateValue>>>,
+    mut thread_receiver: UnboundedReceiver<(
+        StateKey,
+        Version,
+        std::sync::mpsc::Sender<Option<Vec<u8>>>,
+    )>,
 ) {
+    const M: usize = 1024 * 1024;
+    let cache = Arc::new(Mutex::new(
+        LruCache::<(StateKey, Version), Option<Vec<u8>>>::new(M),
+    ));
+
     loop {
-        let (key, version) = if let Some((key, version)) = thread_receiver.recv().await {
-            (key, version)
+        let (key, version, sender) =
+            if let Some((key, version, sender)) = thread_receiver.recv().await {
+                (key, version, sender)
+            } else {
+                break;
+            };
+
+        if let Some(val) = cache.lock().unwrap().get(&(key.clone(), version)) {
+            sender.send(val.clone()).unwrap();
         } else {
-            break;
-        };
-        let val = db.get_state_value_by_version(&key, version - 1).await;
-        thread_sender.send(val).unwrap();
+            assert!(version > 0, "Expecting a non-genesis version");
+            let db = db.clone();
+            let cache = cache.clone();
+            tokio::spawn(async move {
+                let val = db
+                    .get_state_value_by_version(&key, version - 1)
+                    .await
+                    .ok()
+                    .and_then(|v| v.map(|s| s.into_bytes()));
+                cache.lock().unwrap().put((key, version), val.clone());
+                sender.send(val)
+            });
+        }
     }
 }
 
 impl DebuggerStateView {
     pub fn new(db: Arc<dyn AptosValidatorInterface + Send>, version: Version) -> Self {
         let (query_sender, thread_receiver) = unbounded_channel();
-        let (thread_sender, query_receiver) = channel();
 
-        tokio::spawn(async move { handler_thread(db, thread_receiver, thread_sender).await });
+        tokio::spawn(async move { handler_thread(db, thread_receiver).await });
         Self {
             query_sender: Mutex::new(query_sender),
-            query_receiver: Mutex::new(query_receiver),
             version,
         }
     }
@@ -154,18 +174,12 @@ impl DebuggerStateView {
         state_key: &StateKey,
         version: Version,
     ) -> Result<Option<Vec<u8>>> {
-        self.query_sender
-            .lock()
-            .unwrap()
-            .send((state_key.clone(), version))
+        let (tx, rx) = std::sync::mpsc::channel();
+        let query_handler_locked = self.query_sender.lock().unwrap();
+        query_handler_locked
+            .send((state_key.clone(), version, tx))
             .unwrap();
-        Ok(self
-            .query_receiver
-            .lock()
-            .unwrap()
-            .recv()?
-            .ok()
-            .and_then(|v| v.map(|s| s.into_bytes())))
+        Ok(rx.recv()?)
     }
 }
 

aptos-move/aptos-validator-interface/src/rest_interface.rs

@@ -51,7 +51,11 @@ impl AptosValidatorInterface for RestDebuggerInterface {
             StateKey::AccessPath(path) => match path.get_path() {
                 Path::Code(module_id) => Ok(Some(StateValue::new(
                     self.0
-                        .get_account_module_bcs(*module_id.address(), module_id.name().as_str())
+                        .get_account_module_bcs_at_version(
+                            *module_id.address(),
+                            module_id.name().as_str(),
+                            version,
+                        )
                         .await
                         .map_err(|err| anyhow!("Failed to get account states: {:?}", err))?
                         .into_inner()
@@ -84,14 +88,24 @@ impl AptosValidatorInterface for RestDebuggerInterface {
         start: Version,
         limit: u64,
     ) -> Result<Vec<Transaction>> {
-        Ok(self
-            .0
-            .get_transactions_bcs(Some(start), Some(limit as u16))
-            .await?
-            .into_inner()
-            .into_iter()
-            .map(|txn| txn.transaction)
-            .collect())
+        let mut ret = vec![];
+
+        while ret.len() < limit as usize {
+            ret.extend(
+                self.0
+                    .get_transactions_bcs(
+                        Some(start + ret.len() as u64),
+                        Some(limit as u16 - ret.len() as u16),
+                    )
+                    .await?
+                    .into_inner()
+                    .into_iter()
+                    .map(|txn| txn.transaction),
+            );
+            println!("Got {}/{} txns from RestApi.", ret.len(), limit);
+        }
+
+        Ok(ret)
     }
 
     async fn get_latest_version(&self) -> Result<Version> {

crates/aptos-rest-client/src/lib.rs

@@ -1099,6 +1099,19 @@ impl Client {
         self.get_bcs(url).await
     }
 
+    pub async fn get_account_module_bcs_at_version(
+        &self,
+        address: AccountAddress,
+        module_name: &str,
+        version: u64,
+    ) -> AptosResult<Response<bytes::Bytes>> {
+        let url = self.build_path(&format!(
+            "accounts/{}/module/{}?ledger_version={}",
+            address, module_name, version
+        ))?;
+        self.get_bcs(url).await
+    }
+
     pub async fn get_account_events(
         &self,
         address: AccountAddress,