updated

Author: Rachit2323

offchain/_aggregator/Cargo.toml

@@ -0,0 +1,26 @@
+[package]
+name = "square-number-dss-aggregator"
+description = "Square Number DSS Aggregator"
+version = { workspace = true }
+authors = { workspace = true }
+repository = { workspace = true }
+edition = { workspace = true }
+
+[dependencies]
+alloy = { version = "0.2.1", features = ["full"] }
+axum = { version = "0.7.5", features = ["macros"] }
+chrono = { version = "0.4", features = ["serde"] }
+dotenvy = "0.15.7"
+envy = "0.4.2"
+eyre = "0.6.12"
+karak-rs = "=0.1.1"
+serde = { version = "1.0.207", features = ["derive"] }
+serde_json = "1.0.124"
+thiserror = "1.0"
+tokio = { version = "1.39.2", features = ["full"] }
+tower = { version = "0.5.0", features = ["buffer", "limit", "util"] }
+tower-http = { version = "0.5.2", features = ["full"] }
+tower_governor = { version = "0.4.2" }
+tracing = "0.1.40"
+tracing-subscriber = { version = "0.3.18", features = ["env-filter"] }
+url = { version = "2.5.2", features = ["serde"] }

offchain/_aggregator/Dockerfile

@@ -0,0 +1,37 @@
+FROM rust:1.83-slim AS builder
+
+WORKDIR /app
+
+COPY . .
+
+WORKDIR /app/aggregator
+
+# Install necessary dependencies in a single step
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends \
+        pkg-config \
+        libssl-dev \
+        ca-certificates \
+        openssl \
+        libterm-readline-perl-perl && \
+    rm -rf /var/lib/apt/lists/*
+
+RUN cargo build --release --bin square-number-dss-aggregator
+
+# ---- Runtime stage ----
+FROM debian:bookworm-slim
+
+WORKDIR /app
+
+# Set environment to avoid interactive prompts
+ENV DEBIAN_FRONTEND=noninteractive
+
+# Install necessary runtime dependencies
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends curl ca-certificates && \
+    rm -rf /var/lib/apt/lists/*
+
+COPY --from=builder /app/target/release/square-number-dss-aggregator /app/square-number-dss-aggregator
+COPY --from=builder /app/contracts /app/contracts
+
+ENTRYPOINT ["/app/square-number-dss-aggregator"]

offchain/_aggregator/README.md

@@ -0,0 +1,3 @@
+# Aggregator
+
+The aggregator, functioning as an offline entity within the DSS, acts as a trusted central figure. Its primary role is to monitor the DSS contract for any new task requests. As soon as a task request is detected, the aggregator disseminates this request to all [Operators](../operators/README.md) registered in the DSS. After the operators finish executing the requests, the aggregator collects all responses from the operators and verifies their signatures to confirm that the responses are genuinely from the registered operators. Once verified, the aggregator calculates the median of all the received responses. The median is chosen to mitigate the impact of any outliers or erroneous calculations.

offchain/aggregator/src/.env.xx offchain/_aggregator/src/.env.xxoffchain/aggregator/src/.env.xx renamed to offchain/_aggregator/src/.env.xx

@@ -0,0 +1,104 @@
+use alloy::primitives::Address;
+use axum::{extract::State, routing::post, Json, Router};
+use eyre::Result;
+use serde::{Deserialize, Serialize};
+use std::{
+    collections::HashSet,
+    sync::{Arc, RwLock},
+};
+use tracing::info;
+use url::Url;
+
+use crate::error::AppError;
+
+#[derive(Clone, Debug, PartialEq, Eq, Hash, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Operator {
+    public_key: Address,
+    url: Url,
+}
+
+impl Operator {
+    pub fn new(public_key: Address, url: Url) -> Self {
+        Self { public_key, url }
+    }
+
+    pub fn url(&self) -> &Url {
+        &self.url
+    }
+
+    pub fn public_key(&self) -> &Address {
+        &self.public_key
+    }
+}
+
+#[derive(Clone, Debug, Default)]
+pub struct OperatorState {
+    pub operators: Arc<RwLock<HashSet<Operator>>>,
+}
+
+impl OperatorState {
+    pub fn new() -> Self {
+        Self {
+            operators: Arc::new(RwLock::new(HashSet::new())),
+        }
+    }
+
+    pub fn register_operator(&self, operator: Operator) -> Result<()> {
+
+
+        info!("register_operator calledc : {:?}", operator);
+
+        let mut operators = self
+            .operators
+            .write()
+            .map_err(|_| eyre::eyre!("Could not lock"))?;
+        if operators.insert(operator.clone()) {
+            info!("Operator registered: {}", serde_json::to_string(&operator)?);
+        } else {
+            info!(
+                "Operator already registered: {}",
+                serde_json::to_string(&operator)?
+            );
+        }
+        Ok(())
+    }
+
+    pub fn is_operator_registered(&self, operator: Operator) -> Result<bool> {
+        Ok(self
+            .operators
+            .read()
+            .map_err(|_| eyre::eyre!("Could not lock"))?
+            .contains(&operator))
+    }
+}
+
+pub async fn register_operator(
+    State(operators): State<Arc<OperatorState>>,
+    Json(operator): Json<Operator>,
+) -> Result<Json<bool>, AppError> {
+
+    info!("register_operator {:?}", operators);
+
+    operators.register_operator(operator)?;
+    Ok(Json(true))
+}
+
+pub async fn is_operator_registered(
+    State(operators): State<Arc<OperatorState>>,
+    Json(operator): Json<Operator>,
+) -> Result<Json<bool>, AppError> {
+    info!("is_operator_registered  {:?}", operators);
+
+    let registered = operators.is_operator_registered(operator)?;
+    info!("is_operator_registered_after  {:?}", registered);
+
+    Ok(Json(registered))
+}
+
+pub fn aggregator_router(operator_state: Arc<OperatorState>) -> Router {
+    Router::new()
+        .route("/registerOperator", post(register_operator))
+        .route("/isOperatorRegistered", post(is_operator_registered))
+        .with_state(operator_state)
+}

offchain/_aggregator/src/aggregator.rs

@@ -0,0 +1,134 @@
+use alloy::network::{Ethereum, EthereumWallet};
+use alloy::primitives::Address;
+use alloy::providers::fillers::{FillProvider, JoinFill, RecommendedFiller, WalletFiller};
+use alloy::providers::{ProviderBuilder, ReqwestProvider};
+use alloy::sol;
+use alloy::transports::http::ReqwestTransport;
+use karak_rs::contracts::Core::CoreInstance;
+use url::Url;
+use SquareNumberDSS::{TaskRequest, TaskResponse};
+use TxnVerifier::{Task,OperatorResponse};
+use tracing::info;
+
+use crate::Config;
+use crate::TaskError;
+
+sol!(
+    #[sol(rpc)]
+    SquareNumberDSS,
+    "../abi/SquareNumberDSS.json",
+);
+
+
+sol!(
+    #[sol(rpc)]
+    TxnVerifier,
+    "../abi/TxnVerifier.json",
+);
+
+sol!(
+    #[sol(rpc)]
+    #[allow(clippy::too_many_arguments)]
+    VaultAbi,
+    "../abi/Vault.json",
+);
+
+type RecommendedProvider = FillProvider<
+    JoinFill<RecommendedFiller, WalletFiller<EthereumWallet>>,
+    ReqwestProvider,
+    ReqwestTransport,
+    Ethereum,
+>;
+
+pub struct ContractManager {
+    pub dss_instance:
+    TxnVerifier::TxnVerifierInstance<ReqwestTransport, RecommendedProvider>,
+    pub core_instance: CoreInstance<ReqwestTransport, RecommendedProvider>,
+    pub provider: RecommendedProvider,
+}
+
+impl ContractManager {
+    pub fn new(config: &Config) -> Result<Self, TaskError> {
+        let rpc_url = config
+            .get_rpc_url()
+            .map_err(|e| TaskError::CustomUrlError(e.to_string()))?;
+        let private_key = config.get_private_key()?;
+
+        let provider = ProviderBuilder::new()
+            .with_recommended_fillers()
+            .wallet(EthereumWallet::from(private_key))
+            .on_http(rpc_url);
+
+        let txn_verifier_address = config.txn_verifier_address;
+        let dss_instance = TxnVerifier::new(txn_verifier_address, provider.clone());
+
+        let core_address = config.core_address;
+        let core_instance = CoreInstance::new(core_address, provider.clone());
+
+        Ok(Self {
+            dss_instance,
+            core_instance,
+            provider,
+        })
+    }
+
+    pub async fn fetch_vaults_staked_in_dss(
+        &self,
+        operator: Address,
+        dss_address: Address,
+    ) -> Result<Vec<Address>, TaskError> {
+        let result = self
+            .core_instance
+            .fetchVaultsStakedInDSS(operator, dss_address)
+            .call()
+            .await
+            .map_err(|_| TaskError::ContractCallError)?;
+
+        Ok(result.vaults)
+    }
+
+    pub async fn submit_task_response(
+        &self,
+        dss_task_request: Task,
+        task_response: OperatorResponse,
+    ) -> Result<(), TaskError> {
+
+        // info!("submit_task_response  {:?}",dss_task_request);
+        info!("submit_task_responseafter  {:?}",dss_task_request.transaction_hash);
+
+
+        let _ = self
+            .dss_instance
+            .submitTaskResponse(dss_task_request.transaction_hash, task_response)
+            .send()
+            .await
+            .map_err(|_| TaskError::ContractCallError)?;
+
+        Ok(())
+    }
+}
+
+pub struct VaultContract {
+    pub vault_instance: VaultAbi::VaultAbiInstance<ReqwestTransport, RecommendedProvider>,
+    pub provider: RecommendedProvider,
+}
+
+impl VaultContract {
+    pub fn new(
+        rpc_url: Url,
+        private_key: alloy::signers::local::PrivateKeySigner,
+        vault_address: Address,
+    ) -> Result<Self, TaskError> {
+        let provider = ProviderBuilder::new()
+            .with_recommended_fillers()
+            .wallet(EthereumWallet::from(private_key))
+            .on_http(rpc_url);
+
+        let vault_instance = VaultAbi::new(vault_address, provider.clone());
+
+        Ok(Self {
+            vault_instance,
+            provider,
+        })
+    }
+}

offchain/_aggregator/src/contract.rs

@@ -0,0 +1,24 @@
+use axum::{
+    http::StatusCode,
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde_json::json;
+
+pub struct AppError(eyre::Report);
+
+impl IntoResponse for AppError {
+    fn into_response(self) -> Response {
+        tracing::error!(error = %self.0);
+        let status = StatusCode::INTERNAL_SERVER_ERROR;
+        (status, Json(json!({ "error": self.0.to_string() }))).into_response()
+    }
+}
+
+impl From<eyre::Report> for AppError {
+    fn from(error: eyre::Report) -> Self {
+        Self(error)
+    }
+}
+
+pub type Result<T> = std::result::Result<T, AppError>;

offchain/_aggregator/src/error.rs

@@ -0,0 +1,32 @@
+use axum::{
+    http::StatusCode,
+    response::{IntoResponse, Response},
+    Json,
+};
+use serde::{Deserialize, Serialize};
+
+#[derive(Debug, Serialize, Deserialize)]
+pub enum Status {
+    Ok,
+    Warn,
+    Fail,
+}
+
+#[derive(Debug, Serialize, Deserialize)]
+pub struct HealthCheck {
+    status: Status,
+}
+
+impl IntoResponse for HealthCheck {
+    fn into_response(self) -> Response {
+        let code = match self.status {
+            Status::Ok | Status::Warn => StatusCode::OK,
+            Status::Fail => StatusCode::INTERNAL_SERVER_ERROR,
+        };
+        (code, Json(self)).into_response()
+    }
+}
+
+pub async fn health_check() -> HealthCheck {
+    HealthCheck { status: Status::Ok }
+}