feat: cost tracking

committer/src/api.rs

@@ -8,7 +8,8 @@ use actix_web::{
     error::InternalError, get, http::StatusCode, web, App, HttpResponse, HttpServer, Responder,
 };
 use ports::storage::Storage;
-use services::{HealthReporter, StatusReporter};
+use serde::Deserialize;
+use services::{CostReporter, HealthReporter, StatusReporter};
 
 use crate::{
     config::Config,
@@ -19,21 +20,24 @@ use crate::{
 pub async fn launch_api_server(
     config: &Config,
     metrics_registry: Registry,
-    storage: impl Storage + 'static,
+    storage: impl Storage + 'static + Clone,
     fuel_health_check: HealthChecker,
     eth_health_check: HealthChecker,
 ) -> Result<()> {
     let metrics_registry = Arc::new(metrics_registry);
-    let status_reporter = Arc::new(StatusReporter::new(storage));
+    let status_reporter = Arc::new(StatusReporter::new(storage.clone()));
     let health_reporter = Arc::new(HealthReporter::new(fuel_health_check, eth_health_check));
+    let cost_reporter = Arc::new(CostReporter::new(storage));
     HttpServer::new(move || {
         App::new()
             .app_data(web::Data::new(Arc::clone(&metrics_registry)))
             .app_data(web::Data::new(Arc::clone(&status_reporter)))
             .app_data(web::Data::new(Arc::clone(&health_reporter)))
+            .app_data(web::Data::new(Arc::clone(&cost_reporter)))
             .service(status)
             .service(metrics)
             .service(health)
+            .service(costs)
     })
     .bind((config.app.host, config.app.port))
     .map_err(|e| Error::Other(e.to_string()))?
@@ -80,6 +84,30 @@ async fn metrics(registry: web::Data<Arc<Registry>>) -> impl Responder {
     std::result::Result::<_, InternalError<_>>::Ok(text)
 }
 
+#[derive(Deserialize)]
+struct CostQueryParams {
+    from_height: u32,
+    limit: Option<usize>,
+}
+
+#[get("/costs")]
+async fn costs(
+    data: web::Data<Arc<CostReporter<Database>>>,
+    query: web::Query<CostQueryParams>,
+) -> impl Responder {
+    let limit = query.limit.unwrap_or(100);
+
+    if limit == 0 || limit > 1000 {
+        return HttpResponse::BadRequest()
+            .body("Invalid 'limit' parameter. Must be between 1 and 1000.");
+    }
+
+    match data.get_costs(query.from_height, limit).await {
+        Ok(bundle_costs) => HttpResponse::Ok().json(bundle_costs),
+        Err(e) => HttpResponse::from_error(map_to_internal_err(e)),
+    }
+}
+
 fn map_to_internal_err(error: impl std::error::Error) -> InternalError<String> {
     InternalError::new(error.to_string(), StatusCode::INTERNAL_SERVER_ERROR)
 }

e2e/src/committer.rs

@@ -1,7 +1,7 @@
 use std::{path::Path, time::Duration};
 
 use anyhow::Context;
-use ports::types::Address;
+use ports::types::{Address, BundleCost};
 use url::Url;
 
 #[derive(Default)]
@@ -116,7 +116,7 @@ impl Committer {
             cmd.env("COMMITTER__ETH__BLOB_POOL_KEY_ARN", blob_wallet_key_arn);
         }
 
-        let sink = if self.show_logs {
+        let sink = if true {
             std::process::Stdio::inherit
         } else {
             std::process::Stdio::null
@@ -127,7 +127,7 @@ impl Committer {
 
         Ok(CommitterProcess {
             _child: child,
-            port: unused_port,
+            port: 8080,
         })
     }
 
@@ -266,4 +266,17 @@ impl CommitterProcess {
             .expect("metric format to be in the format 'NAME VAL'")
             .parse()?)
     }
+
+    pub async fn fetch_costs(&self, from_height: u32, limit: usize) -> anyhow::Result<BundleCost> {
+        let response = reqwest::get(format!(
+            "http://localhost:{}/costs?from_height={}&limit={}",
+            self.port, from_height, limit
+        ))
+        .await?
+        .error_for_status()?
+        .text()
+        .await?;
+
+        Ok(serde_json::from_str(&response)?)
+    }
 }

e2e/src/lib.rs

@@ -100,6 +100,8 @@ mod tests {
             tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
         }
 
+        let _costs = stack.committer.fetch_costs(0, 10).await?;
+
         Ok(())
     }
 

packages/eth/src/websocket/connection.rs

@@ -438,9 +438,16 @@ impl WsConnection {
 
         let block_number = Self::extract_block_number_from_receipt(&tx_receipt)?;
 
+        let fee = tx_receipt
+            .gas_used
+            .saturating_mul(tx_receipt.effective_gas_price);
+        let blob_fee = Self::extract_blob_fee_from_receipt(&tx_receipt);
+
         Ok(Some(TransactionResponse::new(
             block_number,
             tx_receipt.status(),
+            fee,
+            blob_fee,
         )))
     }
 
@@ -449,6 +456,13 @@ impl WsConnection {
             Error::Other("transaction receipt does not contain block number".to_string())
         })
     }
+
+    fn extract_blob_fee_from_receipt(receipt: &TransactionReceipt) -> u128 {
+        match (receipt.blob_gas_used, receipt.blob_gas_price) {
+            (Some(gas_used), Some(gas_price)) => gas_used.saturating_mul(gas_price),
+            _ => 0,
+        }
+    }
 }
 
 #[cfg(test)]

packages/ports/Cargo.toml

@@ -19,6 +19,7 @@ itertools = { workspace = true, features = ["use_std"], optional = true }
 mockall = { workspace = true, optional = true }
 nonempty = { workspace = true }
 rand = { workspace = true, optional = true }
+serde = { workspace = true }
 sqlx = { workspace = true, features = ["chrono"] }
 thiserror = { workspace = true, optional = true }
 trait-variant = { workspace = true, optional = true }

packages/ports/src/ports/storage.rs

@@ -12,8 +12,8 @@ use itertools::Itertools;
 pub use sqlx::types::chrono::{DateTime, Utc};
 
 use crate::types::{
-    BlockSubmission, BlockSubmissionTx, CollectNonEmpty, CompressedFuelBlock, Fragment, L1Tx,
-    NonEmpty, NonNegative, TransactionState,
+    BlockSubmission, BlockSubmissionTx, BundleCost, CollectNonEmpty, CompressedFuelBlock, Fragment,
+    L1Tx, NonEmpty, NonNegative, TransactionState,
 };
 
 #[derive(Debug, thiserror::Error)]
@@ -197,6 +197,12 @@ pub trait Storage: Send + Sync {
         selective_changes: Vec<([u8; 32], TransactionState)>,
         noncewide_changes: Vec<([u8; 32], u32, TransactionState)>,
     ) -> Result<()>;
+    async fn update_costs(&self, cost_per_tx: Vec<([u8; 32], u128, u64)>) -> Result<()>;
+    async fn get_finalized_costs(
+        &self,
+        from_block_height: u32,
+        limit: usize,
+    ) -> Result<Vec<BundleCost>>;
 }
 
 impl<T: Storage + Send + Sync> Storage for Arc<T> {
@@ -246,6 +252,8 @@ impl<T: Storage + Send + Sync> Storage for Arc<T> {
                     selective_changes: Vec<([u8; 32], TransactionState)>,
                     noncewide_changes: Vec<([u8; 32], u32, TransactionState)>,
                 ) -> Result<()>;
+                async fn update_costs(&self, cost_per_tx: Vec<([u8; 32], u128, u64)>) -> Result<()>;
+                async fn get_finalized_costs(&self, from_block_height: u32, limit: usize) -> Result<Vec<BundleCost>>;
         }
     }
 }
@@ -297,6 +305,8 @@ impl<T: Storage + Send + Sync> Storage for &T {
                     selective_changes: Vec<([u8; 32], TransactionState)>,
                     noncewide_changes: Vec<([u8; 32], u32, TransactionState)>,
                 ) -> Result<()>;
+                async fn update_costs(&self, cost_per_tx: Vec<([u8; 32], u128, u64)>) -> Result<()>;
+                async fn get_finalized_costs(&self, from_block_height: u32, limit: usize) -> Result<Vec<BundleCost>>;
         }
     }
 }

packages/ports/src/types.rs

@@ -7,6 +7,7 @@ mod non_empty;
 pub use non_empty::*;
 
 mod block_submission;
+mod bundle_cost;
 mod fragment;
 #[cfg(feature = "l1")]
 mod fuel_block_committed_on_l1;
@@ -16,6 +17,7 @@ mod state_submission;
 mod transactions;
 
 pub use block_submission::*;
+pub use bundle_cost::*;
 pub use fragment::*;
 #[cfg(feature = "l1")]
 pub use fuel_block_committed_on_l1::*;

packages/ports/src/types/bundle_cost.rs

@@ -0,0 +1,10 @@
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct BundleCost {
+    pub cost: u128,
+    pub size: u64,
+    pub da_block_height: u64,
+    pub start_height: u64,
+    pub end_height: u64,
+}