diff --git a/crates/delegator/src/api.rs b/crates/delegator/src/api.rs index 66e2144..5a81c09 100644 --- a/crates/delegator/src/api.rs +++ b/crates/delegator/src/api.rs @@ -1,6 +1,6 @@ use async_stream::stream; use axum::{ - extract::State, + extract::{Query, State}, response::{sse::Event, Sse}, Json, }; @@ -12,7 +12,7 @@ use std::hash::{DefaultHasher, Hash, Hasher}; use std::{io, time::Duration}; use tokio::sync::{broadcast, mpsc}; use tokio_stream::Stream; -use tracing::{debug, info}; +use tracing::info; use zetina_common::{ hash, job::{Job, JobData}, @@ -40,57 +40,62 @@ impl Clone for ServerState { #[derive(Debug, Deserialize)] pub struct DelegateRequest { - cairo_pie: Vec, + trace: Vec, } #[derive(Debug, Serialize)] pub struct DelegateResponse { - job_hash: u64, + job_hash: String, } pub async fn deletage_handler( State(state): State, Json(input): Json, ) -> Result, StatusCode> { - let cairo_pie = input.cairo_pie; + let cairo_pie = input.trace; let job_data = JobData::new(0, cairo_pie); let job = Job::try_from_job_data(job_data, &state.signing_key); let serialized_job = serde_json::to_string(&job).unwrap(); state.job_topic_tx.send(serialized_job.into()).await.unwrap(); info!("Sent a new job: {}", hash!(&job)); - Ok(Json(DelegateResponse { job_hash: hash!(&job) })) + Ok(Json(DelegateResponse { job_hash: hash!(&job).to_string() })) } #[derive(Debug, Deserialize)] pub struct JobEventsRequest { - job_hash: u64, + job_hash: String, } #[derive(Debug, Serialize)] +#[serde(tag = "type", content = "data")] pub enum JobEventsResponse { - Picked(u64), + Picked(String), Witness(Vec), } pub async fn job_events_handler( State(mut state): State, - Json(input): Json, + Query(input): Query, ) -> Sse>> { - let job_hash = input.job_hash; - let stream = stream! { + let job_hash = input.job_hash.parse::() + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e.to_string()))?; loop { tokio::select! { Ok(job) = state.job_picked_rx.recv() => { - debug!("Received job picked: {}", hash!(job)); + info!("Received job picked: {}", hash!(job)); if hash!(job) == job_hash { - yield Event::default().json_data(JobEventsResponse::Picked(hash!(job))).map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, e.to_string())); + yield Event::default() + .json_data(JobEventsResponse::Picked(hash!(job).to_string())) + .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, e.to_string())); } }, Ok(job_witness) = state.job_witness_rx.recv() => { - debug!("Received job witness: {}", &job_witness.job_hash); - if hash!(job_witness.job_hash) == job_hash { - yield Event::default().json_data(JobEventsResponse::Witness(job_witness.proof)).map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, e.to_string())); + info!("Received job witness: {}", &job_witness.job_hash); + if job_witness.job_hash == job_hash { + yield Event::default() + .json_data(JobEventsResponse::Witness(job_witness.proof)) + .map_err(|e| io::Error::new(io::ErrorKind::BrokenPipe, e.to_string())); } } else => break @@ -101,7 +106,7 @@ pub async fn job_events_handler( Sse::new(stream).keep_alive( axum::response::sse::KeepAlive::new() - .interval(Duration::from_secs(1)) + .interval(Duration::from_secs(5)) .text("keep-alive-text"), ) } diff --git a/dashboard/cairo_pie.zip b/dashboard/cairo_pie.zip new file mode 100644 index 0000000..a53ed08 Binary files /dev/null and b/dashboard/cairo_pie.zip differ diff --git a/dashboard/public/zetina-logo.png b/dashboard/public/zetina-logo.png new file mode 100644 index 0000000..373d23a Binary files /dev/null and b/dashboard/public/zetina-logo.png differ diff --git a/dashboard/src/app/api.ts b/dashboard/src/app/api.ts index f9bddfa..6d39f68 100644 --- a/dashboard/src/app/api.ts +++ b/dashboard/src/app/api.ts @@ -2,34 +2,30 @@ import { z } from "zod"; // Zod for DelegateRequest export const DelegateRequest = z.object({ - cairo_pie: z.instanceof(Uint8Array), + trace: z.array(z.number()), }); export type DelegateRequest = z.infer; // Zod for DelegateResponse export const DelegateResponse = z.object({ - job_hash: z.number(), + job_hash: z.coerce.bigint(), }); export type DelegateResponse = z.infer; // Zod for JobEventsRequest export const JobEventsRequest = z.object({ - job_hash: z.number(), + job_hash: z.coerce.bigint(), }); export type JobEventsRequest = z.infer; -// Zod for JobEventsResponse -export const Picked = z.object({ - type: z.literal("Picked"), - data: z.number(), +export const JobEventsResponse = z.object({ + type: z.literal("Picked").or(z.literal("Witness")), + data: z.any(), }); -export type Picked = z.infer; +export type JobEventsResponse = z.infer; -export const Witness = z.object({ - type: z.literal("Witness"), - data: z.instanceof(Uint8Array), -}); -export type Witness = z.infer; +export const JobHash = z.coerce.bigint(); +export type JobHash = z.infer; -export const JobEventsResponse = z.union([Picked, Witness]); -export type JobEventsResponse = z.infer; +export const Proof = z.array(z.number()); +export type Proof = z.infer; diff --git a/dashboard/src/app/layout.tsx b/dashboard/src/app/layout.tsx index 7252780..2813e35 100644 --- a/dashboard/src/app/layout.tsx +++ b/dashboard/src/app/layout.tsx @@ -16,6 +16,7 @@ export default function RootLayout({ }>) { return ( + {children} ); diff --git a/dashboard/src/app/page.tsx b/dashboard/src/app/page.tsx index a744256..ab1b2ab 100644 --- a/dashboard/src/app/page.tsx +++ b/dashboard/src/app/page.tsx @@ -12,8 +12,10 @@ import Check from "@mui/icons-material/Check"; import StepConnector, { stepConnectorClasses, } from "@mui/material/StepConnector"; -import { DelegateRequest, DelegateResponse } from "./api"; +import { DelegateRequest, DelegateResponse, JobEventsResponse, JobHash, Proof } from "./api"; import { useState } from "react"; +import subscribeEvents from "./subscribeEvents"; +import assert from "assert"; const steps = [ "Job propagated to network", @@ -27,11 +29,6 @@ const QontoConnector = styled(StepConnector)(({ theme }) => ({ left: "calc(-50% + 16px)", right: "calc(50% + 16px)", }, - [`&.${stepConnectorClasses.active}`]: { - [`& .${stepConnectorClasses.line}`]: { - borderColor: "#784af4", - }, - }, [`&.${stepConnectorClasses.completed}`]: { [`& .${stepConnectorClasses.line}`]: { borderColor: "#784af4", @@ -51,9 +48,6 @@ const QontoStepIconRoot = styled("div")<{ ownerState: { active?: boolean } }>( display: "flex", height: 22, alignItems: "center", - ...(ownerState.active && { - color: "#784af4", - }), "& .QontoStepIcon-completedIcon": { color: "#784af4", zIndex: 1, @@ -86,6 +80,7 @@ export default function Home() { const darkTheme = createTheme({ palette: { mode: "dark", + primary: { main: "#784af4", dark: "#784af4" } }, }); @@ -100,11 +95,15 @@ export default function Home() { reader.onload = async (e) => { if (e.target && e.target.result) { const fileBytes = new Uint8Array(e.target.result as ArrayBuffer); - + console.log(Array.from(fileBytes)); const requestBody: DelegateRequest = DelegateRequest.parse({ - cairo_pie: fileBytes, + trace: Array.from(fileBytes), }); + console.log(requestBody); + + let subscriber: EventSource | null = null + try { const response = await fetch("http://localhost:3010/delegate", { method: "POST", @@ -122,10 +121,33 @@ export default function Home() { await response.json(), ); console.log("Job hash:", data.job_hash); - setIsProcessing(data.job_hash) + + setActiveStep(1) + setIsProcessing(data.job_hash); + + subscriber = subscribeEvents( + "http://localhost:3010/job_events", + `job_hash=${data.job_hash.toString()}`, + (event) => { + let job_event = JobEventsResponse.parse(event); + if (job_event.type == "Picked") { + let job_hash = JobHash.parse(job_event.data); + assert(job_hash == data.job_hash) + setActiveStep(2) + } + if (job_event.type == "Witness") { + let proof = Proof.parse(job_event.data); + setActiveStep(3) + setDownloadBlob([new Blob([new Uint8Array(proof)]), `${data.job_hash}_proof.json`]) + setIsProcessing(null) + subscriber?.close() + } + }, + ); } catch (error) { console.error("Failed to upload file:", error); - setIsProcessing(null) + setIsProcessing(null); + subscriber?.close() } } }; @@ -133,7 +155,9 @@ export default function Home() { reader.readAsArrayBuffer(file); }; - const [isProcessing, setIsProcessing] = useState(null); + const [isProcessing, setIsProcessing] = useState(null); + const [activeStep, setActiveStep] = useState(0); + const [downloadBlob, setDownloadBlob] = useState<[Blob, string] | null>(null); const { getRootProps, getInputProps, isDragActive } = useDropzone({ onDrop: ondrop, @@ -144,8 +168,9 @@ export default function Home() {
-
- Supply program Trace +
+
ZK prover network
+
Prove program Trace
{isProcessing != null ? ( -

Processing job hash: {isProcessing}

+

+ Processing job: {isProcessing.toString()} +

) : isDragActive ? (

Drop the Trace here ...

) : (

Drag Trace here, or click to select files

- ) - } + )}
- + } > @@ -178,7 +206,18 @@ export default function Home() { ))}
-
diff --git a/dashboard/src/app/subscribeEvents.ts b/dashboard/src/app/subscribeEvents.ts index 392a489..d23ec5c 100644 --- a/dashboard/src/app/subscribeEvents.ts +++ b/dashboard/src/app/subscribeEvents.ts @@ -1,11 +1,10 @@ -export default async function subscribeEvents( +export default function subscribeEvents( url: string, - fetchparams: string, - sseparams: string, + params: string, callback: (data: any) => void, ) { - const events = new EventSource(`${url}?${sseparams}`, { - withCredentials: true, + const events = new EventSource(`${url}?${params}`, { + withCredentials: false, }); const eventSource = new ReadableStream({ @@ -29,13 +28,6 @@ export default async function subscribeEvents( }, }); - const response = await fetch(`${url}?${fetchparams}`, { - method: "GET", - credentials: "same-origin", - }).then((r) => r.json()); - - callback(response); - eventSource.pipeTo(eventSink); return events; diff --git a/docker-compose.yaml b/docker-compose.yaml index d8614a9..aa89440 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -23,7 +23,7 @@ services: cpus: '8' memory: '10G' ports: - - "50051:50051/tcp" + - "3010:3010/tcp" environment: - RUST_LOG=info networks: @@ -40,7 +40,7 @@ services: cpus: '8' memory: '10G' ports: - - "50052:50052/tcp" + - "3020:3020/tcp" environment: - RUST_LOG=info networks: