Merge remote-tracking branch 'origin/staging' into feat/integrate-agg-mode-with-db

Author: MarcosNicolau

Makefile

@@ -306,10 +306,10 @@ proof_aggregator_write_program_ids: ## Write proof aggregator zkvm programs ids
 	@cd aggregation_mode/proof_aggregator && ./scripts/build_programs.sh
 
 agg_mode_docker_up:
-	@cd aggregation_mode && docker-compose up -d
+	@cd aggregation_mode && docker compose up -d
 
 agg_mode_docker_down:
-	@cd aggregation_mode && docker-compose down
+	@cd aggregation_mode && docker compose down
 
 agg_mode_docker_clean: agg_mode_docker_down
 	docker volume rm aggregation-mode_postgres_data

aggregation_mode/batcher/abi/AggregationModePaymentService.json

@@ -14,6 +14,23 @@ pub enum DbError {
     ConnectError(String),
 }
 
+#[derive(Debug, Clone, sqlx::Type, serde::Serialize)]
+#[sqlx(type_name = "task_status")]
+#[sqlx(rename_all = "lowercase")]
+pub enum TaskStatus {
+    Pending,
+    Processing,
+    Verified,
+}
+
+#[derive(Debug, Clone, sqlx::FromRow, sqlx::Type, serde::Serialize)]
+pub struct Receipt {
+    pub status: TaskStatus,
+    pub merkle_path: Option<Vec<u8>>,
+    pub nonce: i64,
+    pub address: String,
+}
+
 impl Db {
     pub async fn try_new(connection_url: &str) -> Result<Self, DbError> {
         let pool = PgPoolOptions::new()
@@ -45,29 +62,66 @@ impl Db {
             .map(|res| res.flatten())
     }
 
+    pub async fn get_tasks_by_address_and_nonce(
+        &self,
+        address: &str,
+        nonce: i64,
+    ) -> Result<Vec<Receipt>, sqlx::Error> {
+        sqlx::query_as::<_, Receipt>(
+            "SELECT status,merkle_path,nonce,address FROM tasks
+                WHERE address = $1
+                AND nonce = $2
+                ORDER BY nonce DESC",
+        )
+        .bind(address.to_lowercase())
+        .bind(nonce)
+        .fetch_all(&self.pool)
+        .await
+    }
+
+    pub async fn get_tasks_by_address_with_limit(
+        &self,
+        address: &str,
+        limit: i64,
+    ) -> Result<Vec<Receipt>, sqlx::Error> {
+        sqlx::query_as::<_, Receipt>(
+            "SELECT status,merkle_path,nonce,address FROM tasks
+                WHERE address = $1
+                ORDER BY nonce DESC
+                LIMIT $2",
+        )
+        .bind(address.to_lowercase())
+        .bind(limit)
+        .fetch_all(&self.pool)
+        .await
+    }
+
     pub async fn insert_task(
         &self,
         address: &str,
         proving_system_id: i64,
         proof: &[u8],
         program_commitment: &[u8],
         merkle_path: Option<&[u8]>,
+        nonce: i64,
     ) -> Result<Uuid, sqlx::Error> {
         sqlx::query_scalar::<_, Uuid>(
             "INSERT INTO tasks (
                 address,
                 proving_system_id,
                 proof,
                 program_commitment,
-                merkle_path
-            ) VALUES ($1, $2, $3, $4, $5)
+                merkle_path,
+                nonce
+            ) VALUES ($1, $2, $3, $4, $5, $6)
             RETURNING task_id",
         )
         .bind(address.to_lowercase())
         .bind(proving_system_id)
         .bind(proof)
         .bind(program_commitment)
         .bind(merkle_path)
+        .bind(nonce)
         .fetch_one(&self.pool)
         .await
     }

aggregation_mode/batcher/src/db.rs

@@ -14,13 +14,13 @@ use sqlx::types::BigDecimal;
 
 use super::{
     helpers::format_merkle_path,
-    types::{AppResponse, GetProofMerklePathQueryParams},
+    types::{AppResponse, GetReceiptsQueryParams},
 };
 
 use crate::{
     config::Config,
     db::Db,
-    server::types::{SubmitProofRequestRisc0, SubmitProofRequestSP1},
+    server::types::{GetReceiptsResponse, SubmitProofRequestRisc0, SubmitProofRequestSP1},
     verifiers::{verify_sp1_proof, VerificationError},
 };
 
@@ -47,7 +47,7 @@ impl BatcherServer {
                 // Note: this is temporary and should be lowered when we accept proofs via multipart form data instead of json
                 .app_data(web::JsonConfig::default().limit(50 * 1024 * 1024)) // 50mb
                 .route("/nonce/{address}", web::get().to(Self::get_nonce))
-                .route("/proof/merkle", web::get().to(Self::get_proof_merkle_path))
+                .route("/receipts", web::get().to(Self::get_receipts))
                 .route("/proof/sp1", web::post().to(Self::post_proof_sp1))
                 .route("/proof/risc0", web::post().to(Self::post_proof_risc0))
         })
@@ -58,20 +58,28 @@ impl BatcherServer {
         .expect("Server to never end");
     }
 
+    // Returns the nonce (number of submitted tasks) for a given address
     async fn get_nonce(req: HttpRequest) -> impl Responder {
-        let Some(address) = req.match_info().get("address") else {
+        let Some(address_raw) = req.match_info().get("address") else {
             return HttpResponse::BadRequest()
                 .json(AppResponse::new_unsucessfull("Missing address", 400));
         };
 
-        // TODO: validate valid ethereum address
+        // Check that the address is a valid ethereum address
+        if alloy::primitives::Address::from_str(address_raw.trim()).is_err() {
+            return HttpResponse::BadRequest()
+                .json(AppResponse::new_unsucessfull("Invalid address", 400));
+        }
+
+        let address = address_raw.to_lowercase();
+
         let Some(state) = req.app_data::<Data<BatcherServer>>() else {
             return HttpResponse::InternalServerError()
                 .json(AppResponse::new_unsucessfull("Internal server error", 500));
         };
 
         let state = state.get_ref();
-        match state.db.count_tasks_by_address(address).await {
+        match state.db.count_tasks_by_address(&address).await {
             Ok(count) => HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!(
                 {
                     "nonce": count
@@ -82,6 +90,7 @@ impl BatcherServer {
         }
     }
 
+    // Posts an SP1 proof to the batcher, recovering the address from the signature
     async fn post_proof_sp1(
         req: HttpRequest,
         MultipartForm(data): MultipartForm<SubmitProofRequestSP1>,
@@ -174,6 +183,7 @@ impl BatcherServer {
                 &proof_content,
                 &vk_content,
                 None,
+                data.nonce.0 as i64,
             )
             .await
         {
@@ -186,65 +196,89 @@ impl BatcherServer {
     }
 
     /// TODO: complete for risc0 (see `post_proof_sp1`)
+    // Posts a Risc0 proof to the batcher, recovering the address from the signature
     async fn post_proof_risc0(
         _req: HttpRequest,
         MultipartForm(_): MultipartForm<SubmitProofRequestRisc0>,
     ) -> impl Responder {
         HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({})))
     }
 
-    async fn get_proof_merkle_path(
+    // Returns the last 100 receipt merkle proofs for the address received in the URL.
+    // In case of also receiving a nonce on the query param, it returns only the merkle proof for that nonce.
+    async fn get_receipts(
         req: HttpRequest,
-        params: web::Query<GetProofMerklePathQueryParams>,
+        params: web::Query<GetReceiptsQueryParams>,
     ) -> impl Responder {
         let Some(state) = req.app_data::<Data<BatcherServer>>() else {
-            return HttpResponse::InternalServerError()
-                .json(AppResponse::new_unsucessfull("Internal server error", 500));
+            return HttpResponse::InternalServerError().json(AppResponse::new_unsucessfull(
+                "Internal server error: Failed to get app data",
+                500,
+            ));
         };
 
         let state = state.get_ref();
 
-        // TODO: maybe also accept proof commitment in query param
-        let Some(id) = params.id.clone() else {
-            return HttpResponse::BadRequest().json(AppResponse::new_unsucessfull(
-                "Provide task `id` query param",
-                400,
-            ));
-        };
-
-        if id.is_empty() {
-            return HttpResponse::BadRequest().json(AppResponse::new_unsucessfull(
-                "Proof id cannot be empty",
-                400,
-            ));
+        if alloy::primitives::Address::from_str(params.address.clone().trim()).is_err() {
+            return HttpResponse::BadRequest()
+                .json(AppResponse::new_unsucessfull("Invalid address", 400));
         }
 
-        let Ok(proof_id) = sqlx::types::Uuid::parse_str(&id) else {
-            return HttpResponse::BadRequest()
-                .json(AppResponse::new_unsucessfull("Proof id invalid uuid", 400));
+        let limit = match params.limit {
+            Some(received_limit) => received_limit.min(100),
+            None => 100,
         };
 
-        let db_result = state.db.get_merkle_path_by_task_id(proof_id).await;
-        let merkle_path = match db_result {
-            Ok(Some(merkle_path)) => merkle_path,
-            Ok(None) => {
-                return HttpResponse::NotFound().json(AppResponse::new_unsucessfull(
-                    "Proof merkle path not found",
-                    404,
-                ))
-            }
-            Err(_) => {
-                return HttpResponse::InternalServerError()
-                    .json(AppResponse::new_unsucessfull("Internal server error", 500));
-            }
+        let address = params.address.to_lowercase();
+
+        let query = if let Some(nonce) = params.nonce {
+            state
+                .db
+                .get_tasks_by_address_and_nonce(&address, nonce)
+                .await
+        } else {
+            state
+                .db
+                .get_tasks_by_address_with_limit(&address, limit)
+                .await
         };
 
-        match format_merkle_path(&merkle_path) {
-            Ok(merkle_path) => {
-                HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({
-                    "merkle_path": merkle_path
-                })))
-            }
+        let Ok(receipts) = query else {
+            return HttpResponse::InternalServerError().json(AppResponse::new_unsucessfull(
+                "Internal server error: Failed to get tasks by address and nonce",
+                500,
+            ));
+        };
+
+        let responses: Result<Vec<GetReceiptsResponse>, String> = receipts
+            .into_iter()
+            .map(|receipt| {
+                let Some(merkle_path) = receipt.merkle_path else {
+                    return Ok(GetReceiptsResponse {
+                        status: receipt.status,
+                        merkle_path: Vec::new(),
+                        nonce: receipt.nonce,
+                        address: receipt.address,
+                    });
+                };
+
+                let Ok(formatted) = format_merkle_path(&merkle_path) else {
+                    return Err("Error formatting merkle path".into());
+                };
+
+                Ok(GetReceiptsResponse {
+                    status: receipt.status,
+                    merkle_path: formatted,
+                    nonce: receipt.nonce,
+                    address: receipt.address,
+                })
+            })
+            .collect();
+
+        match responses {
+            Ok(resp) => HttpResponse::Ok().json(AppResponse::new_sucessfull(serde_json::json!({
+                "receipts": resp
+            }))),
             Err(_) => HttpResponse::InternalServerError()
                 .json(AppResponse::new_unsucessfull("Internal server error", 500)),
         }

aggregation_mode/batcher/src/server/http.rs

@@ -2,6 +2,8 @@ use actix_multipart::form::{tempfile::TempFile, text::Text, MultipartForm};
 use serde::{Deserialize, Serialize};
 use serde_json::Value;
 
+use crate::db::TaskStatus;
+
 #[derive(Serialize, Deserialize)]
 pub(super) struct AppResponse {
     status: u16,
@@ -27,10 +29,14 @@ impl AppResponse {
     }
 }
 
-/// Query parameters accepted by `GET /proof/merkle`, containing an optional proof id.
+/// Query parameters accepted by `GET /receipts`. Requires an address, and accepts a nonce
+/// and a limit for the amount of tasks included in the query (the maximum value is 100).
+/// Note: The limit value will only be taken into account if nonce is None.
 #[derive(Deserialize, Clone)]
-pub(super) struct GetProofMerklePathQueryParams {
-    pub id: Option<String>,
+pub(super) struct GetReceiptsQueryParams {
+    pub address: String,
+    pub nonce: Option<i64>,
+    pub limit: Option<i64>,
 }
 
 #[derive(Debug, MultipartForm)]
@@ -48,3 +54,11 @@ pub(super) struct SubmitProofRequestRisc0 {
     pub _program_image_id_hex: Text<String>,
     pub _signature_hex: Text<String>,
 }
+
+#[derive(Debug, Clone, sqlx::FromRow, sqlx::Type, serde::Serialize)]
+pub struct GetReceiptsResponse {
+    pub status: TaskStatus,
+    pub merkle_path: Vec<String>,
+    pub nonce: i64,
+    pub address: String,
+}

aggregation_mode/batcher/src/server/types.rs

@@ -7,7 +7,8 @@ CREATE TABLE tasks (
     proof BYTEA,
     program_commitment BYTEA,
     merkle_path BYTEA,
-    status task_status DEFAULT 'pending'
+    status task_status DEFAULT 'pending',
+    nonce BIGINT NOT NULL
 );
 
 CREATE TABLE payment_events (