feat: resumeable archiving process & cloud indexing

.env.example

@@ -1,2 +1,6 @@
 archiver_pk=...
-network="./networks/your_network.json"
+network="./networks/your_network.json"
+
+DATABASE_HOST=""
+DATABASE_USERNAME=""
+DATABASE_PASSWORD=""

Cargo.toml

@@ -4,13 +4,15 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
+anyhow = "1.0.86"
 borsh = "1.5.1"
 borsh-derive = "1.5.1"
 brotli = "6.0.0"
 dotenv = "0.15.0"
 ethers = "2.0.14"
 ethers-core = "2.0.14"
 ethers-providers = "2.0.14"
+planetscale-driver = "0.5.1"
 serde = "1.0.205"
 serde_json = "1.0.122"
 tokio = { version = "1.39.2", features = ["full"] }

Dockerfile

@@ -0,0 +1,7 @@
+FROM rust:1.73
+
+COPY ./ ./
+
+RUN cargo build --release
+
+CMD ["./target/release/wvm-archiver"]

db_schema.sql

@@ -0,0 +1,7 @@
+DROP TABLE IF EXISTS WeaveVMArchiver;
+
+CREATE TABLE IF NOT EXISTS WeaveVMArchiver (
+    Id INT AUTO_INCREMENT PRIMARY KEY,
+    NetworkBlockId INT UNIQUE,
+    WeaveVMArchiveTxid VARCHAR(68) UNIQUE
+);

src/main.rs

@@ -1,8 +1,6 @@
-use ethers_core::k256::pkcs8::der::asn1::Null;
-
 use crate::utils::archive_block::archive;
 use crate::utils::schema::Network;
-
+use crate::utils::planetscale::{ps_archive_block, ps_get_latest_block_id};
 use std::thread;
 use std::time::Duration;
 
@@ -11,15 +9,21 @@ mod utils;
 async fn main() {
     let network = Network::config();
     let block_time = network.block_time;
-    let mut start_block = network.start_block;
+    let ps_latest_archived_block = ps_get_latest_block_id().await;
+    // it defaults to network.start_block if planestcale fails
+    let mut start_block = ps_latest_archived_block;
 
     println!("\n{:#?}\n\n", network);
 
-    // poll block & archive
+    // poll blocks & archive
     loop {
         println!("\n{}", "#".repeat(100));
-        println!("\nARCHIVING BLOCK #{} of Network {} -- ChainId: {}\n", start_block, network.name, network.network_chain_id);
-        archive(Some(start_block)).await;
+        println!(
+            "\nARCHIVING BLOCK #{} of Network {} -- ChainId: {}\n",
+            start_block, network.name, network.network_chain_id
+        );
+        let archive_txid = archive(Some(start_block)).await.unwrap();
+        let _ =  ps_archive_block(&start_block, &archive_txid).await;
         start_block += 1;
         println!("\n{}", "#".repeat(100));
         thread::sleep(Duration::from_secs(block_time.into()));

src/utils/archive_block.rs

@@ -1,13 +1,13 @@
+use crate::utils::get_block::by_number;
 use crate::utils::schema::{Block, Network};
 use crate::utils::transaction::send_wvm_calldata;
-use crate::utils::get_block::by_number;
-
+use anyhow::Error;
 
-pub async fn archive(block_number: Option<u64>) {
+pub async fn archive(block_number: Option<u64>) -> Result<String, Error> {
     let network = Network::config();
     let start_block = network.start_block;
     let block_to_archive = block_number.unwrap_or(start_block);
-    
+
     // fetch block
     let block_data = by_number(block_to_archive).await.unwrap();
     // serialize response into Block struct
@@ -17,10 +17,10 @@ pub async fn archive(block_number: Option<u64>) {
     let borsh_res = Block::borsh_ser(&block_data_struct);
     // brotli compress the borsh serialized block
     let brotli_res = Block::brotli_compress(&borsh_res);
-    
+
     // println!("borsh vec length: {:?}", borsh_res.len());
     // println!("brotli vec length: {:?}", brotli_res.len());
 
-    let _ = send_wvm_calldata(brotli_res).await;
-
-}
+    let txid = send_wvm_calldata(brotli_res).await.unwrap();
+    Ok(txid)
+}

src/utils/env_var.rs

@@ -1,7 +1,7 @@
-use std::env;
 use dotenv::dotenv;
+use std::env;
 
-pub fn get_env_var(key: &str) -> Result<String, env::VarError>{
+pub fn get_env_var(key: &str) -> Result<String, env::VarError> {
     dotenv().ok();
     match env::var(key) {
         Ok(val) => Ok(val),

src/utils/get_block.rs

@@ -1,10 +1,10 @@
-use ethers_providers::{Middleware, ProviderError};
-use ethers_core::types::{Block, H256};
 use crate::utils::schema::Network;
+use ethers_core::types::{Block, H256};
+use ethers_providers::{Middleware, ProviderError};
 
 pub async fn by_number(number: u64) -> Result<Option<Block<H256>>, ProviderError> {
     let network: Network = Network::config();
-    let provider = Network::provider(&network, false).await;    
+    let provider = Network::provider(&network, false).await;
     let block = provider.get_block(number).await;
 
     match block {

src/utils/mod.rs

@@ -1,5 +1,6 @@
-pub mod schema;
+pub mod archive_block;
+pub mod env_var;
 pub mod get_block;
+pub mod schema;
 pub mod transaction;
-pub mod env_var;
-pub mod archive_block;
+pub mod planetscale;

src/utils/planetscale.rs

@@ -0,0 +1,54 @@
+use std::net;
+
+use crate::utils::env_var::get_env_var;
+use crate::utils::schema::Network;
+use anyhow::Error;
+use planetscale_driver::{PSConnection, query};
+
+async fn ps_init() -> PSConnection {
+
+    let host = get_env_var("DATABASE_HOST").unwrap();
+    let username = get_env_var("DATABASE_USERNAME").unwrap();
+    let password = get_env_var("DATABASE_PASSWORD").unwrap();
+
+    let conn: PSConnection = PSConnection::new(
+        &host,
+        &username,
+        &password,
+      );
+
+      conn     
+}
+
+pub async fn ps_archive_block(network_block_id: &u64, wvm_calldata_txid: &str) -> Result<(), Error> {
+    // format to the table VAR(66) limitation
+    let wvm_calldata_txid = wvm_calldata_txid.trim_matches('"');    
+    let conn = ps_init().await;
+
+    let res = query("INSERT INTO WeaveVMArchiver(NetworkBlockId, WeaveVMArchiveTxid) VALUES($0, \"$1\")")
+        .bind(network_block_id)
+        .bind(wvm_calldata_txid)
+        .execute(&conn)
+        .await;
+
+    match res {
+        Ok(result) => {
+            println!("Insert operation was successful: {:?}", result);
+            Ok(result)
+        }
+        Err(e) => {
+            println!("Error occurred during insert operation: {:?}", e);
+            Err(e)
+        }
+    }
+}
+
+pub async fn ps_get_latest_block_id() -> u64 {
+    let network = Network::config();
+    let conn = ps_init().await;
+
+    let latest_archived: u64 = query("SELECT MAX(NetworkBlockId) AS LatestNetworkBlockId FROM WeaveVMArchiver;").fetch_scalar(&conn).await.unwrap_or(network.start_block);
+    // return latest archived block in planetscale + 1
+    // so the process can start archiving from latest_archived + 1
+    latest_archived + 1
+}

src/utils/schema.rs

@@ -1,12 +1,12 @@
+use crate::utils::env_var::get_env_var;
+use borsh::to_vec;
+use borsh_derive::{BorshDeserialize, BorshSerialize};
+use ethers_providers::{Http, Provider};
 use serde::{Deserialize, Serialize};
+use serde_json::Value;
+use std::convert::TryFrom;
 use std::fs::File;
 use std::io::{Read, Write};
-use ethers_providers::{Http, Provider};
-use std::convert::TryFrom;
-use serde_json::Value;
-use borsh_derive::{BorshDeserialize, BorshSerialize};
-use borsh::to_vec;
-use crate::utils::env_var::get_env_var;
 
 #[derive(Serialize, Deserialize, Debug)]
 pub struct Network {
@@ -18,21 +18,21 @@ pub struct Network {
     pub block_time: u32,
     pub start_block: u64, // as per ethers_provider
     pub archiver_address: String,
-    pub archive_pool_address: String
+    pub archive_pool_address: String,
 }
 
 impl Network {
     pub fn config() -> Network {
         let network_config = get_env_var("network").unwrap();
         let mut file = File::open(network_config).unwrap();
         let mut data = String::new();
-        
+
         file.read_to_string(&mut data).unwrap();
-        
-        let network: Network = serde_json::from_str(&data).unwrap(); 
+
+        let network: Network = serde_json::from_str(&data).unwrap();
         // cannot self send data
         assert_ne!(network.archiver_address, network.archive_pool_address);
-        network       
+        network
     }
 
     pub async fn provider(&self, rpc: bool) -> Provider<Http> {
@@ -44,58 +44,51 @@ impl Network {
         } else {
             target_rpc = &network.network_rpc
         }
-        let provider: Provider<Http> = Provider::<Http>::try_from(
-            target_rpc
-        ).expect("could not instantiate HTTP Provider");
-
+        let provider: Provider<Http> =
+            Provider::<Http>::try_from(target_rpc).expect("could not instantiate HTTP Provider");
+
         provider
     }
 }
 
-
 #[derive(Debug, Deserialize, Serialize, BorshSerialize, BorshDeserialize, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub struct Block {
-    pub base_fee_per_gas: Option<String>,           // "baseFeePerGas"
-    pub blob_gas_used: Option<String>,              // "blobGasUsed"
-    pub difficulty: String,                         // "difficulty"
-    pub excess_blob_gas: Option<String>,            // "excessBlobGas"
-    pub extra_data: String,                         // "extraData"
-    pub gas_limit: String,                          // "gasLimit"
-    pub gas_used: String,                           // "gasUsed"
-    pub hash: String,                               // "hash"
-    pub logs_bloom: String,                         // "logsBloom"
-    pub miner: String,                              // "miner"
-    pub mix_hash: String,                           // "mixHash"
-    pub nonce: String,                              // "nonce"
-    pub number: String,                             // "number"
-    pub parent_beacon_block_root: Option<String>,   // "parentBeaconBlockRoot"
-    pub parent_hash: String,                        // "parentHash"
-    pub receipts_root: String,                      // "receiptsRoot"
-    pub seal_fields: Vec<String>,                   // "sealFields" as an array of strings
-    pub sha3_uncles: String,                        // "sha3Uncles"
-    pub size: String,                               // "size"
-    pub state_root: String,                         // "stateRoot"
-    pub timestamp: String,                          // "timestamp"
-    pub total_difficulty: String,                   // "totalDifficulty"
-    pub transactions: Vec<String>,                  // "transactions" as an array of strings
+    pub base_fee_per_gas: Option<String>,         // "baseFeePerGas"
+    pub blob_gas_used: Option<String>,            // "blobGasUsed"
+    pub difficulty: String,                       // "difficulty"
+    pub excess_blob_gas: Option<String>,          // "excessBlobGas"
+    pub extra_data: String,                       // "extraData"
+    pub gas_limit: String,                        // "gasLimit"
+    pub gas_used: String,                         // "gasUsed"
+    pub hash: String,                             // "hash"
+    pub logs_bloom: String,                       // "logsBloom"
+    pub miner: String,                            // "miner"
+    pub mix_hash: String,                         // "mixHash"
+    pub nonce: String,                            // "nonce"
+    pub number: String,                           // "number"
+    pub parent_beacon_block_root: Option<String>, // "parentBeaconBlockRoot"
+    pub parent_hash: String,                      // "parentHash"
+    pub receipts_root: String,                    // "receiptsRoot"
+    pub seal_fields: Vec<String>,                 // "sealFields" as an array of strings
+    pub sha3_uncles: String,                      // "sha3Uncles"
+    pub size: String,                             // "size"
+    pub state_root: String,                       // "stateRoot"
+    pub timestamp: String,                        // "timestamp"
+    pub total_difficulty: String,                 // "totalDifficulty"
+    pub transactions: Vec<String>,                // "transactions" as an array of strings
 }
 
 impl Block {
     pub fn load_block_from_value(value: Value) -> Result<Block, serde_json::Error> {
         serde_json::from_value::<Block>(value)
     }
     pub fn brotli_compress(input: &[u8]) -> Vec<u8> {
-        let mut writer = brotli::CompressorWriter::new(
-            Vec::new(),
-            4096,
-            11,
-            22);
+        let mut writer = brotli::CompressorWriter::new(Vec::new(), 4096, 11, 22);
         writer.write_all(input).unwrap();
         writer.into_inner()
     }
-
     pub fn borsh_ser(input: &Block) -> Vec<u8> {
         to_vec(input).unwrap()
-    } 
-}
+    }
+}