Add support for handling billing process for the service offered by the operator

Cargo.toml

@@ -20,9 +20,10 @@ actix-web = "4"
 anyhow = "1.0.75"
 clap = { version = "4.4.7", features = ["derive"] }
 data-encoding = "2.5.0"
+ethers = "2.0.11"
 hex = "0.4.3"
 k256 = { version = "0.13.2", features = ["ecdsa", "ecdsa-core"] }
-openssl = { version = "0.10", features = ["vendored"] }
+openssl = { version = "0.10.60", features = ["vendored"] }
 rand = "0.8.5"
 reqwest = { version = "0.11.9", features = ["json"] }
 serde = { version = "1.0", features = ["derive"] }

src/billing_handler.rs

@@ -0,0 +1,138 @@
+use crate::model::AppState;
+
+use actix_web::web::{Data, Json};
+use actix_web::{get, post, HttpResponse, Responder};
+use k256::ecdsa::SigningKey;
+use k256::elliptic_curve::generic_array::sequence::Lengthen;
+use serde_json::json;
+use tiny_keccak::{Hasher, Keccak};
+
+#[derive(Debug, serde::Deserialize)]
+pub struct SigningData {
+    nonce: String,
+    tx_hashes: Vec<String>,
+}
+
+#[get("/billing/inspect")]
+pub async fn inspect_bill(appstate: Data<AppState>) -> impl Responder {
+    HttpResponse::Ok().json(json!({
+        "bill": appstate.execution_costs.lock().unwrap().clone(),
+    }))
+}
+
+#[get("/billing/latest")]
+pub async fn get_last_bill_claim(appstate: Data<AppState>) -> impl Responder {
+    let mut last_bill_claim_guard = appstate.last_bill_claim.lock().unwrap();
+    if last_bill_claim_guard.0.is_none() {
+        return HttpResponse::BadRequest().body("No bill claimed yet!");
+    }
+
+    if last_bill_claim_guard.1.is_some() {
+        return HttpResponse::Ok().json(json!({
+            "bill_claim_data": last_bill_claim_guard.0.clone().unwrap(),
+            "signature": last_bill_claim_guard.1.clone().unwrap(),
+        }));
+    }
+
+    let bill_claim_data = hex::decode(last_bill_claim_guard.0.clone().unwrap());
+    if bill_claim_data.is_err() {
+        return HttpResponse::InternalServerError().body(format!(
+            "Failed to decode claimed bill data: {}",
+            bill_claim_data.unwrap_err()
+        ));
+    }
+
+    let bill_claim_data = bill_claim_data.unwrap();
+    let signature = sign_data(bill_claim_data.as_slice(), &appstate.signer).await;
+    if signature.is_err() {
+        return HttpResponse::InternalServerError().body(format!(
+            "Failed to sign billing data: {}",
+            signature.unwrap_err()
+        ));
+    }
+
+    let bill_claim_data = hex::encode(bill_claim_data.as_slice());
+    let signature = signature.unwrap();
+    last_bill_claim_guard.1 = Some(signature.clone());
+
+    HttpResponse::Ok().json(json!({
+        "bill_claim_data": bill_claim_data,
+        "signature": signature,
+    }))
+}
+
+#[post("/billing/export")]
+pub async fn export_bill(appstate: Data<AppState>, data: Json<SigningData>) -> impl Responder {
+    let signing_data = data.into_inner();
+    if signing_data.nonce.is_empty() {
+        return HttpResponse::BadRequest().body("Nonce must not be empty!");
+    }
+
+    if signing_data.tx_hashes.is_empty() {
+        return HttpResponse::BadRequest().body("List of transaction hashes must not be empty!");
+    }
+
+    let mut bytes32_nonce = [0u8; 32];
+    if let Err(err) = hex::decode_to_slice(&signing_data.nonce, &mut bytes32_nonce) {
+        return HttpResponse::BadRequest()
+            .body(format!("Failed to decode nonce into 32 bytes: {}", err));
+    }
+
+    let mut bill_claim_data = bytes32_nonce.to_vec();
+    for tx_hash in signing_data.tx_hashes {
+        if let Some(cost) = appstate.execution_costs.lock().unwrap().remove(&tx_hash) {
+            let mut bytes32_tx_hash = [0u8; 32];
+            if let Err(_) = hex::decode_to_slice(&tx_hash[2..], &mut bytes32_tx_hash) {
+                continue;
+            }
+
+            bill_claim_data.append(&mut bytes32_tx_hash.to_vec());
+            bill_claim_data.append(&mut cost.to_be_bytes().to_vec());
+        }
+    }
+
+    if bill_claim_data.len() == 32 {
+        return HttpResponse::BadRequest()
+            .body("Given transaction hashes are not present in billing data");
+    }
+
+    let signature = sign_data(bill_claim_data.as_slice(), &appstate.signer).await;
+    if signature.is_err() {
+        appstate
+            .last_bill_claim
+            .lock()
+            .unwrap()
+            .0
+            .clone_from(&Some(hex::encode(bill_claim_data.as_slice())));
+
+        return HttpResponse::InternalServerError().body(format!(
+            "Failed to sign billing data: {}",
+            signature.unwrap_err()
+        ));
+    }
+
+    let signature = signature.unwrap();
+    let bill_claim_data = hex::encode(bill_claim_data.as_slice());
+
+    let mut last_bill_claim_guard = appstate.last_bill_claim.lock().unwrap();
+    last_bill_claim_guard.0 = Some(bill_claim_data.clone());
+    last_bill_claim_guard.1 = Some(signature.clone());
+
+    HttpResponse::Ok().json(json!({
+        "bill_claim_data": bill_claim_data,
+        "signature": signature,
+    }))
+}
+
+async fn sign_data(data: &[u8], signer: &SigningKey) -> Result<String, anyhow::Error> {
+    let mut hasher = Keccak::v256();
+    hasher.update(data);
+
+    let mut hash = [0u8; 32];
+    hasher.finalize(&mut hash);
+
+    let (rs, v) = signer.sign_prehash_recoverable(&hash)?;
+    let signature = hex::encode(rs.to_bytes().append(27 + v.to_byte()).as_slice());
+
+    return Ok(signature);
+}

src/contracts/billing_contract_abi.json

@@ -0,0 +1,21 @@
+[
+	{
+		"inputs": [
+			{
+				"internalType": "bytes32",
+				"name": "_txhash",
+				"type": "bytes32"
+			}
+		],
+		"name": "balanceOf",
+		"outputs": [
+			{
+				"internalType": "uint256",
+				"name": "",
+				"type": "uint256"
+			}
+		],
+		"stateMutability": "view",
+		"type": "function"
+	}
+]

src/handler.rs

@@ -1,12 +1,13 @@
-use crate::{cgroups, model::AppState, workerd};
+use std::io::{BufRead, BufReader};
+use std::sync::atomic::Ordering;
+use std::time::{Duration, Instant};
+
+use crate::model::AppState;
+use crate::{cgroups, workerd};
 
 use actix_web::http::{header, StatusCode};
 use actix_web::{web, HttpRequest, HttpResponse, Responder};
 use anyhow::{anyhow, Context};
-use std::io::{BufRead, BufReader};
-use std::sync::atomic::Ordering;
-use std::time::Duration;
-// use std::time::Instant;
 use tokio::time::timeout;
 
 pub async fn serverless(
@@ -66,12 +67,14 @@ pub async fn serverless(
         workerd_runtime_path,
         &appstate.rpc,
         &appstate.contract,
+        &appstate.billing_contract,
     )
     .await
     {
         use workerd::ServerlessError::*;
         return match err {
             CalldataRetrieve(_)
+            | TxDepositNotEnough
             | TxNotFound
             | InvalidTxToType
             | InvalidTxToValue(_, _)
@@ -91,7 +94,7 @@ pub async fn serverless(
         };
     }
 
-    // let execution_timer_start = Instant::now();
+    let execution_timer_start = Instant::now();
 
     // reserve cgroup
     let cgroup = appstate.cgroups.lock().unwrap().reserve();
@@ -153,22 +156,13 @@ pub async fn serverless(
 
         use workerd::ServerlessError::*;
         return match err {
-            CalldataRetrieve(_)
-            | TxNotFound
-            | InvalidTxToType
-            | InvalidTxToValue(_, _)
-            | InvalidTxCalldataType
-            | BadCalldata(_) => HttpResponse::BadRequest().body(format!(
+            ConfigFileCreate(_) => HttpResponse::InternalServerError().body(format!(
                 "{:?}",
-                anyhow!(err).context("failed to create code file")
-            )),
-            CodeFileCreate(_) => HttpResponse::InternalServerError().body(format!(
-                "{:?}",
-                anyhow!(err).context("failed to create code file")
+                anyhow!(err).context("failed to create config file")
             )),
             _ => HttpResponse::InternalServerError().body(format!(
                 "{:?}",
-                anyhow!(err).context("unexpected error while trying to create code file")
+                anyhow!(err).context("unexpected error while trying to create config file")
             )),
         };
     }
@@ -250,6 +244,22 @@ pub async fn serverless(
         .context("CRITICAL: failed to clean up code file")
         .unwrap_or_else(|err| println!("{err:?}"));
 
+    let execution_timer_end = Instant::now();
+    let execution_time = execution_timer_end
+        .duration_since(execution_timer_start)
+        .as_millis();
+
+    // TODO: FIX THE VALUE OF FIXED COST AND CONVERSION RATE
+    let execution_cost = 1 + 2 * execution_time;
+
+    appstate
+        .execution_costs
+        .lock()
+        .unwrap()
+        .entry(tx_hash.to_owned())
+        .and_modify(|cost| *cost += execution_cost)
+        .or_insert(execution_cost);
+
     if let Err(err) = response {
         return HttpResponse::RequestTimeout()
             .body(format!("{:?}", anyhow!(err).context("worker timed out")));
@@ -264,12 +274,5 @@ pub async fn serverless(
     }
     let response = response.unwrap();
 
-    // let execution_timer_end = Instant::now();
-    // let execution_time = execution_timer_end
-    //     .duration_since(execution_timer_start)
-    //     .as_millis()
-    //     .to_string();
-    // println!("Execution time: {}ms", execution_time);
-
     return response;
 }

src/lib.rs

@@ -1,5 +1,17 @@
+pub mod billing_handler;
 pub mod cgroups;
 pub mod handler;
 pub mod model;
 mod tests;
 pub mod workerd;
+
+use ethers::contract::abigen;
+use ethers::providers::{Http, Provider};
+
+abigen!(
+    BillingContract,
+    "src/contracts/billing_contract_abi.json",
+    derives(serde::Serialize, serde::Deserialize)
+);
+
+type BillContract = BillingContract<Provider<Http>>;

src/main.rs

@@ -1,10 +1,17 @@
-use actix_web::{web, App, HttpServer};
-use anyhow::{anyhow, Context};
-use clap::Parser;
-use tokio::fs;
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::time::Duration;
 
 use serverless::cgroups::Cgroups;
 use serverless::model::AppState;
+use serverless::BillingContract;
+
+use actix_web::{web, App, HttpResponse, HttpServer};
+use anyhow::{anyhow, Context};
+use clap::Parser;
+use ethers::providers::{Http, Provider};
+use ethers::types::Address;
+use tokio::fs;
 
 /// Simple program to greet a person
 #[derive(Parser, Debug)]
@@ -13,6 +20,9 @@ struct Args {
     #[clap(long, value_parser, default_value = "6001")]
     port: u16,
 
+    #[clap(long, value_parser)]
+    billing_port: u16, // TODO: ADD THE DEFAULT PORT
+
     #[clap(long, value_parser, default_value = "./runtime/")]
     runtime_path: String,
 
@@ -33,6 +43,9 @@ struct Args {
     )]
     contract: String,
 
+    #[clap(long, value_parser)]
+    billing_contract: String, // TODO: ADD A DEFAULT ADDRESS
+
     #[clap(long, value_parser)]
     signer: String,
 }
@@ -52,6 +65,7 @@ async fn main() -> anyhow::Result<()> {
     // println!("{:?}", response);
 
     let port: u16 = cli.port;
+    let billing_port: u16 = cli.billing_port;
 
     let cgroups = Cgroups::new().context("failed to construct cgroups")?;
     if cgroups.free.is_empty() {
@@ -66,14 +80,28 @@ async fn main() -> anyhow::Result<()> {
     )
     .context("invalid signer key")?;
 
+    let rpc_provider = Provider::<Http>::try_from(&cli.rpc)
+        .context("Failed to connect to the rpc")?
+        .interval(Duration::from_millis(1000));
+    let billing_contract = BillingContract::new(
+        cli.billing_contract
+            .parse::<Address>()
+            .context("Failed to parse billing contract address")?,
+        Arc::new(rpc_provider),
+    );
+
     let app_data = web::Data::new(AppState {
         cgroups: cgroups.into(),
         running: std::sync::atomic::AtomicBool::new(true),
         runtime_path: cli.runtime_path,
         rpc: cli.rpc,
         contract: cli.contract,
-        signer,
+        signer: signer,
+        billing_contract: billing_contract,
+        execution_costs: HashMap::new().into(),
+        last_bill_claim: (None, None).into(),
     });
+    let app_data_clone = app_data.clone();
 
     let server = HttpServer::new(move || {
         App::new()
@@ -86,7 +114,21 @@ async fn main() -> anyhow::Result<()> {
 
     println!("Server started on port {}", port);
 
-    server.await?;
+    let billing_server = HttpServer::new(move || {
+        App::new()
+            .app_data(app_data_clone.clone())
+            .service(serverless::billing_handler::inspect_bill)
+            .service(serverless::billing_handler::get_last_bill_claim)
+            .service(serverless::billing_handler::export_bill)
+            .default_service(web::to(HttpResponse::NotFound))
+    })
+    .bind(("0.0.0.0", billing_port))
+    .context(format!("could not bind to port {billing_port}"))?
+    .run();
+
+    println!("Billing Server started on port {}", billing_port);
+
+    tokio::try_join!(server, billing_server)?;
 
     Ok(())
 }