snarkos_aot/ledger/
query.rs

1use std::{
2    fs::File,
3    io::Write,
4    net::{IpAddr, SocketAddr},
5    ops::Deref,
6    path::PathBuf,
7    sync::Arc,
8};
9
10use anyhow::Result;
11use axum::{
12    extract::{self, Query, State},
13    response::IntoResponse,
14    routing::{get, post},
15    Json, Router,
16};
17use clap::Args;
18use reqwest::StatusCode;
19use serde_json::json;
20use tracing_appender::non_blocking::NonBlocking;
21
22use crate::{
23    cli::{make_env_filter, ReloadHandler},
24    Block, DbLedger, Network, Transaction,
25};
26
27/// Receive inquiries on `/<network>/latest/stateRoot`.
28#[derive(Debug, Args, Clone)]
29pub struct LedgerQuery<N: Network> {
30    /// Port to listen on for incoming messages.
31    #[arg(long, default_value = "3030")]
32    pub port: u16,
33
34    // IP address to bind to.
35    #[arg(long, default_value = "0.0.0.0")]
36    pub bind: IpAddr,
37
38    /// When true, the POST `/block` endpoint will not be available.
39    #[arg(long)]
40    pub readonly: bool,
41
42    /// Receive messages from `/<network>/transaction/broadcast` and record them
43    /// to the output.
44    #[arg(long)]
45    pub record: bool,
46
47    /// Path to the directory containing the stored data.
48    #[arg(long, short, default_value = "transactions.json")]
49    pub output: PathBuf,
50
51    #[clap(skip)]
52    phantom: std::marker::PhantomData<N>,
53}
54
55struct LedgerState<N: Network> {
56    readonly: bool,
57    ledger: DbLedger<N>,
58    appender: Option<NonBlocking>,
59    log_level_handler: ReloadHandler,
60}
61
62type AppState<N> = Arc<LedgerState<N>>;
63
64impl<N: Network> LedgerQuery<N> {
65    #[tokio::main]
66    pub async fn parse(self, ledger: &DbLedger<N>, log_level_handler: ReloadHandler) -> Result<()> {
67        let (appender, _guard) = if self.record {
68            let (appender, guard) = tracing_appender::non_blocking(
69                File::options()
70                    .create(true)
71                    .append(true)
72                    .open(self.output.clone())
73                    .expect("Failed to open the file for writing transactions"),
74            );
75            (Some(appender), Some(guard))
76        } else {
77            (None, None)
78        };
79
80        let state = LedgerState {
81            readonly: self.readonly,
82            ledger: ledger.clone(),
83            appender,
84            log_level_handler,
85        };
86
87        let network = N::str_id();
88
89        let app = Router::new()
90            .route(
91                &format!("/{network}/latest/stateRoot"),
92                get(Self::latest_state_root),
93            )
94            .route(
95                &format!("/{network}/block/height/latest"),
96                get(Self::latest_height),
97            )
98            .route(
99                &format!("/{network}/block/hash/latest"),
100                get(Self::latest_hash),
101            )
102            .route(
103                &format!("/{network}/transaction/broadcast"),
104                post(Self::broadcast_tx),
105            )
106            .route("/block", post(Self::add_block))
107            .route("/log", post(Self::set_log_level))
108            // TODO: for ahead of time ledger generation, support a /beacon_block endpoint to write
109            // beacon block TODO: api to get and decrypt records for a private key
110            .with_state(Arc::new(state));
111
112        let listener = tokio::net::TcpListener::bind(SocketAddr::new(self.bind, self.port)).await?;
113        tracing::info!("listening on: {:?}", listener.local_addr().unwrap());
114        axum::serve(listener, app).await?;
115
116        Ok(())
117    }
118
119    async fn latest_state_root(state: State<AppState<N>>) -> impl IntoResponse {
120        Json(json!(state.ledger.latest_state_root()))
121    }
122
123    async fn latest_height(state: State<AppState<N>>) -> impl IntoResponse {
124        Json(json!(state.ledger.latest_height()))
125    }
126
127    async fn latest_hash(state: State<AppState<N>>) -> impl IntoResponse {
128        Json(json!(state.ledger.latest_hash()))
129    }
130
131    async fn broadcast_tx(
132        state: State<AppState<N>>,
133        payload: extract::Json<Transaction<N>>,
134    ) -> impl IntoResponse {
135        let Ok(tx_json) = serde_json::to_string(payload.deref()) else {
136            return StatusCode::BAD_REQUEST;
137        };
138
139        if let Some(mut a) = state.appender.clone() {
140            match write!(a, "{}", tx_json) {
141                Ok(_) => StatusCode::OK,
142                Err(_) => StatusCode::INTERNAL_SERVER_ERROR,
143            }
144        } else {
145            println!("{}", tx_json);
146            StatusCode::OK
147        }
148    }
149
150    async fn add_block(
151        state: State<AppState<N>>,
152        payload: extract::Json<Block<N>>,
153    ) -> impl IntoResponse {
154        if state.readonly {
155            return (StatusCode::FORBIDDEN, Json(json!({"error": "readonly"})));
156        }
157
158        if state.ledger.latest_hash() != payload.previous_hash()
159            || state.ledger.latest_state_root() != payload.previous_state_root()
160            || state.ledger.latest_height() + 1 != payload.height()
161        {
162            return (
163                StatusCode::BAD_REQUEST,
164                Json(json!({"error": "invalid block"})),
165            );
166        }
167
168        if let Err(e) = state
169            .ledger
170            .check_next_block(&payload, &mut rand::thread_rng())
171        {
172            return (
173                StatusCode::INTERNAL_SERVER_ERROR,
174                Json(json!({"error": format!("failed to validate block: {e}")})),
175            );
176        }
177
178        match state.ledger.advance_to_next_block(&payload) {
179            Ok(_) => (StatusCode::OK, Json(json!({"status": "ok"}))),
180            Err(e) => (
181                StatusCode::INTERNAL_SERVER_ERROR,
182                Json(json!({"error": format!("failed to advance block: {e}")})),
183            ),
184        }
185    }
186
187    async fn set_log_level(
188        state: State<AppState<N>>,
189        Query(verbosity): Query<u8>,
190    ) -> impl IntoResponse {
191        let Ok(_) = state
192            .log_level_handler
193            .modify(|filter| *filter = make_env_filter(verbosity))
194        else {
195            return (
196                StatusCode::INTERNAL_SERVER_ERROR,
197                Json(json!({"error": "failed to set log level"})),
198            );
199        };
200
201        (StatusCode::OK, Json(json!({"status": "ok"})))
202    }
203}
snarkos_aot/ledger/query.rs

snarkos_aot/ledger/
query.rs
