miden_note_transport_node/node/grpc/
mod.rs

1mod streaming;
2
3use std::collections::BTreeSet;
4use std::net::SocketAddr;
5use std::sync::Arc;
6use std::time::Duration;
7
8use chrono::Utc;
9use miden_note_transport_proto::miden_note_transport::miden_note_transport_server::MidenNoteTransportServer;
10use miden_note_transport_proto::miden_note_transport::{
11    FetchNotesRequest,
12    FetchNotesResponse,
13    SendNoteRequest,
14    SendNoteResponse,
15    StatsResponse,
16    StreamNotesRequest,
17    TransportNote,
18};
19use miden_protocol::utils::serde::Deserializable;
20use rand::Rng;
21use tokio::sync::mpsc;
22use tonic::Status;
23use tonic_web::GrpcWebLayer;
24use tower::limit::GlobalConcurrencyLimitLayer;
25use tower::timeout::TimeoutLayer;
26use tower_http::cors::{Any, CorsLayer};
27
28use self::streaming::{NoteStreamer, StreamerMessage, Sub, Subface};
29use crate::database::Database;
30use crate::metrics::MetricsGrpc;
31
32/// Miden Note Transport gRPC server
33pub struct GrpcServer {
34    database: Arc<Database>,
35    config: GrpcServerConfig,
36    streamer: StreamerCtx,
37    metrics: MetricsGrpc,
38}
39
40/// [`GrpcServer`] configuration
41#[derive(Clone, Debug)]
42pub struct GrpcServerConfig {
43    /// Server host
44    pub host: String,
45    /// Server port
46    pub port: u16,
47    /// Maximum note size to be stored
48    pub max_note_size: usize,
49    /// Maximum number of concurrent connections
50    pub max_connections: usize,
51    /// Connection timeout in seconds
52    pub request_timeout: usize,
53}
54
55/// Streaming task interface context
56pub(super) struct StreamerCtx {
57    tx: mpsc::Sender<StreamerMessage>,
58    handle: tokio::task::JoinHandle<()>,
59}
60
61impl Default for GrpcServerConfig {
62    fn default() -> Self {
63        Self {
64            host: "127.0.0.1".to_string(),
65            port: 57292,
66            max_note_size: 512_000,
67            max_connections: 4096,
68            request_timeout: 4,
69        }
70    }
71}
72
73impl GrpcServer {
74    /// gRPC server constructor
75    pub fn new(database: Arc<Database>, config: GrpcServerConfig, metrics: MetricsGrpc) -> Self {
76        let streamer = StreamerCtx::spawn(database.clone());
77        Self { database, config, streamer, metrics }
78    }
79
80    /// Convert into a service
81    pub fn into_service(self) -> MidenNoteTransportServer<Self> {
82        MidenNoteTransportServer::new(self)
83    }
84
85    /// gRPC server running-task
86    pub async fn serve(self) -> crate::Result<()> {
87        let (health_reporter, health_svc) = tonic_health::server::health_reporter();
88        health_reporter.set_serving::<MidenNoteTransportServer<Self>>().await;
89
90        let addr = format!("{}:{}", self.config.host, self.config.port)
91            .parse::<SocketAddr>()
92            .map_err(|e| crate::Error::Internal(format!("Invalid address: {e}")))?;
93
94        let cors = CorsLayer::new().allow_origin(Any).allow_headers(Any).allow_methods(Any);
95
96        tonic::transport::Server::builder()
97            .accept_http1(true)
98            .layer(cors)
99            .layer(GrpcWebLayer::new())
100            .layer(GlobalConcurrencyLimitLayer::new(self.config.max_connections))
101            .layer(TimeoutLayer::new(Duration::from_secs(self.config.request_timeout as u64)))
102            .add_service(health_svc)
103            .add_service(self.into_service())
104            .serve(addr)
105            .await
106            .map_err(|e| crate::Error::Internal(format!("Server error: {e}")))
107    }
108}
109
110impl StreamerCtx {
111    /// Spawn a [`NoteStreamer`] task
112    ///
113    /// Returns related context composed of the handle and `mpsc::Sender` `tx` for control messages.
114    pub(super) fn spawn(database: Arc<Database>) -> Self {
115        let (tx, rx) = mpsc::channel(128);
116        let handle = tokio::spawn(NoteStreamer::new(database, rx).stream());
117        Self { tx, handle }
118    }
119}
120
121#[tonic::async_trait]
122impl miden_note_transport_proto::miden_note_transport::miden_note_transport_server::MidenNoteTransport
123    for GrpcServer
124{
125    #[tracing::instrument(skip(self), fields(operation = "grpc.send_note.request"))]
126    async fn send_note(
127        &self,
128        request: tonic::Request<SendNoteRequest>,
129    ) -> Result<tonic::Response<SendNoteResponse>, tonic::Status> {
130        let request_data = request.into_inner();
131        let pnote = request_data.note.ok_or_else(|| Status::invalid_argument("Missing note"))?;
132
133        let timer = self.metrics.grpc_send_note_request((pnote.header.len() + pnote.details.len()) as u64);
134
135        // Validate note size
136        if pnote.details.len() > self.config.max_note_size {
137            return Err(Status::resource_exhausted(format!("Note too large ({})", pnote.details.len())));
138        }
139
140        // Convert protobuf request to internal types
141        let header = miden_protocol::note::NoteHeader::read_from_bytes(&pnote.header)
142            .map_err(|e| Status::invalid_argument(format!("Invalid header: {e:?}")))?;
143
144        // Create note for database
145        let note_for_db = crate::types::StoredNote {
146            header,
147            details: pnote.details,
148            created_at: Utc::now(),
149        };
150
151        self.database
152            .store_note(&note_for_db)
153            .await.map_err(|e| tonic::Status::internal(format!("Failed to store note: {e:?}")))?;
154
155        timer.finish("ok");
156
157        Ok(tonic::Response::new(SendNoteResponse {}))
158    }
159
160    #[tracing::instrument(skip(self), fields(operation = "grpc.fetch_notes.request"))]
161    async fn fetch_notes(
162        &self,
163        request: tonic::Request<FetchNotesRequest>,
164    ) -> Result<tonic::Response<FetchNotesResponse>, tonic::Status> {
165        let timer = self.metrics.grpc_fetch_notes_request();
166
167        let request_data = request.into_inner();
168        let tags = request_data.tags.into_iter().collect::<BTreeSet<_>>();
169        let cursor = request_data.cursor;
170
171        let mut rcursor = cursor;
172        let mut proto_notes = vec![];
173        for tag in tags {
174            let stored_notes = self
175                .database
176                .fetch_notes(tag.into(), cursor)
177                .await.map_err(|e| tonic::Status::internal(format!("Failed to fetch notes: {e:?}")))?;
178
179            for stored_note in &stored_notes {
180                let ts_cursor: u64 = stored_note
181                    .created_at
182                    .timestamp_micros()
183                    .try_into()
184                    .map_err(|_| tonic::Status::internal("Timestamp too large for cursor"))?;
185                rcursor = rcursor.max(ts_cursor);
186            }
187
188            proto_notes.extend(stored_notes.into_iter().map(TransportNote::from));
189        }
190
191        timer.finish("ok");
192
193        let proto_notes_size = proto_notes.iter().map(|pnote| (pnote.header.len() + pnote.details.len()) as u64).sum();
194        self.metrics.grpc_fetch_notes_response(
195            proto_notes.len() as u64,
196            proto_notes_size,
197        );
198
199        Ok(tonic::Response::new(FetchNotesResponse { notes: proto_notes, cursor: rcursor }))
200    }
201
202    type StreamNotesStream = Sub;
203    #[tracing::instrument(skip(self), fields(operation = "grpc.stream_notes.request"))]
204    async fn stream_notes(
205        &self,
206        request: tonic::Request<StreamNotesRequest>,
207    ) -> Result<tonic::Response<Self::StreamNotesStream>, tonic::Status> {
208        let request_data = request.into_inner();
209        let tag = request_data.tag.into();
210        let id = rand::rng().random();
211        let (sub_tx, sub_rx) = mpsc::channel(32);
212        let sub = Sub::new(id, tag, sub_rx, self.streamer.tx.clone());
213        let subf = Subface::new(id, tag, sub_tx);
214        self.streamer.tx.try_send(StreamerMessage::AddSub(subf))
215                    .map_err(|e| tonic::Status::internal(format!("Failed sending internal streamer message: {e}")))?;
216
217        Ok(tonic::Response::new(sub))
218    }
219
220    #[tracing::instrument(skip(self), fields(operation = "grpc.stats.request"))]
221    async fn stats(
222        &self,
223        _request: tonic::Request<()>,
224    ) -> Result<tonic::Response<StatsResponse>, tonic::Status> {
225        let (total_notes, total_tags) = self
226            .database
227            .get_stats()
228            .await.map_err(|e| tonic::Status::internal(format!("Failed to get stats: {e:?}")))?;
229
230        let response = StatsResponse {
231            total_notes,
232            total_tags,
233            notes_per_tag: Vec::new(), // TODO: Implement notes_per_tag
234        };
235
236        Ok(tonic::Response::new(response))
237    }
238}
239
240impl Drop for StreamerCtx {
241    fn drop(&mut self) {
242        if let Err(e) = self.tx.try_send(StreamerMessage::Shutdown) {
243            tracing::error!("Streamer shutdown message sending failure: {e}");
244            self.handle.abort();
245        }
246    }
247}
miden_note_transport_node/node/grpc/mod.rs

miden_note_transport_node/node/grpc/
mod.rs
