vertigo_cli/serve/
server_state.rs

1use parking_lot::RwLock;
2use std::{
3    collections::HashMap,
4    sync::{Arc, OnceLock},
5    time::{Duration, Instant, SystemTime, UNIX_EPOCH},
6};
7use tokio::sync::mpsc::{error::TryRecvError, unbounded_channel};
8use vertigo::{
9    JsJson, JsJsonSerialize,
10    dev::{
11        command::{CommandForBrowser, ConsoleLogLevel, browser_response},
12        command_wire::decode_dom_commands,
13    },
14};
15use wasmtime::{Engine, InstancePre, Module};
16
17use crate::{
18    commons::{ErrorCode, spawn::SpawnOwner},
19    serve::html::FetchCache,
20};
21
22use super::{
23    html::HtmlResponse,
24    mount_path::MountConfig,
25    request_state::RequestState,
26    response_state::ResponseState,
27    timings::SsrProbe,
28    wasm::{HostState, Message, WasmInstance, build_linker},
29};
30
31#[cfg(feature = "ssr-timings")]
32use super::timings::SsrTimings;
33
34pub fn get_now() -> Duration {
35    let start = SystemTime::now();
36    match start.duration_since(UNIX_EPOCH) {
37        Ok(duration) => duration,
38        Err(err) => {
39            log::error!("Time went backwards: {err}");
40            Duration::from_secs(0)
41        }
42    }
43}
44
45pub type ServerStateMap = HashMap<String, Arc<ServerState>>;
46
47static STATE: OnceLock<Arc<RwLock<ServerStateMap>>> = OnceLock::new();
48
49#[derive(Clone)]
50pub struct ServerState {
51    engine: Engine,
52    /// The module with its imports already resolved by name, once, at startup. Holds the
53    /// [`Module`] internally, so there is no separate field for it.
54    instance_pre: InstancePre<HostState>,
55    /// What `Module::from_binary` cost at startup.
56    module_compile: Duration,
57    pub mount_config: MountConfig,
58    pub port_watch: Option<u16>,
59}
60
61impl ServerState {
62    pub fn init(mount_config: &MountConfig) -> Result<(), ErrorCode> {
63        Self::init_with_watch(mount_config, None)
64    }
65
66    pub fn init_with_watch(
67        mount_config: &MountConfig,
68        port_watch: Option<u16>,
69    ) -> Result<(), ErrorCode> {
70        let engine = Engine::default();
71
72        let (module, module_compile) = build_module_wasm(&engine, mount_config)?;
73
74        // Import resolution happens here, once - a missing or mistyped import is a startup
75        // failure naming the offending import, not a per-request one.
76        let instance_pre = build_linker(&engine)?
77            .instantiate_pre(&module)
78            .map_err(|err| {
79                log::error!("WASM import resolution failed: {err:?}");
80                ErrorCode::ServeWasmInstanceFailed
81            })?;
82
83        let mutex = STATE.get_or_init(|| Arc::new(RwLock::new(ServerStateMap::new())));
84
85        let mut guard = mutex.write();
86        guard.insert(
87            mount_config.mount_point().to_string(),
88            Arc::new(Self {
89                engine,
90                instance_pre,
91                module_compile,
92                mount_config: mount_config.clone(),
93                port_watch,
94            }),
95        );
96
97        Ok(())
98    }
99
100    /// How long compiling this mount point's wasm module took, at startup.
101    pub fn module_compile_time(&self) -> Duration {
102        self.module_compile
103    }
104
105    pub fn global(mount_point: &str) -> Arc<ServerState> {
106        let mutex = STATE.get_or_init(|| Arc::new(RwLock::new(ServerStateMap::new())));
107
108        let guard = mutex.read();
109
110        if let Some(state) = guard.get(mount_point) {
111            return state.clone();
112        }
113
114        unreachable!();
115    }
116
117    pub async fn request(&self, url: &str) -> ResponseState {
118        self.request_inner(url, &SsrProbe::new()).await
119    }
120
121    /// [`ServerState::request`], with the per-phase breakdown of how the render was spent.
122    #[cfg(feature = "ssr-timings")]
123    pub async fn request_timed(&self, url: &str) -> (ResponseState, SsrTimings) {
124        let probe = SsrProbe::new();
125        let mark = probe.start();
126
127        let response = self.request_inner(url, &probe).await;
128
129        let timings = probe.finish(mark, response.body.len());
130        (response, timings)
131    }
132
133    async fn request_inner(&self, url: &str, probe: &SsrProbe) -> ResponseState {
134        let (sender, mut receiver) = unbounded_channel::<Message>();
135
136        let request = RequestState {
137            url: url.to_string(),
138            env: self.mount_config.env.clone(),
139        };
140
141        let fetch = FetchCache::new();
142
143        let instantiate_mark = probe.start();
144        let handle_command = Arc::new({
145            let sender = sender.clone();
146            let probe = probe.clone();
147
148            move |request: RequestState, command| match command {
149                CommandForBrowser::FetchCacheGet => {
150                    browser_response::FetchCacheGet { data: None }.to_json()
151                }
152                CommandForBrowser::FetchExec { request, callback } => {
153                    sender
154                        .send(Message::FetchRequest { callback, request })
155                        .inspect_err(|err| log::error!("Error sending FetchRequest: {err}"))
156                        .unwrap_or_default();
157
158                    JsJson::Null
159                }
160                CommandForBrowser::SetStatus { status } => {
161                    sender
162                        .send(Message::SetStatus(status))
163                        .inspect_err(|err| log::error!("Error sending FetchRequest: {err}"))
164                        .unwrap_or_default();
165
166                    JsJson::Null
167                }
168                CommandForBrowser::IsBrowser => {
169                    let response = browser_response::IsBrowser { value: false };
170
171                    response.to_json()
172                }
173                CommandForBrowser::GetDateNow => {
174                    let time = get_now().as_millis();
175
176                    let response = browser_response::GetDateNow { value: time as u64 };
177
178                    response.to_json()
179                }
180                CommandForBrowser::WebsocketRegister {
181                    host: _,
182                    callback: _,
183                } => JsJson::Null,
184                CommandForBrowser::WebsocketUnregister { callback: _ } => JsJson::Null,
185                CommandForBrowser::WebsocketSendMessage {
186                    callback: _,
187                    message: _,
188                } => JsJson::Null,
189                CommandForBrowser::TimerSet {
190                    callback,
191                    duration,
192                    kind: _,
193                } => {
194                    if duration == 0 {
195                        sender
196                            .send(Message::SetTimeoutZero { callback })
197                            .inspect_err(|err| log::error!("Error sending SetTimeoutZero: {err}"))
198                            .unwrap_or_default();
199                    }
200
201                    JsJson::Null
202                }
203                CommandForBrowser::TimerClear { callback: _ } => JsJson::Null,
204                CommandForBrowser::LocationCallback {
205                    target: _,
206                    mode: _,
207                    callback: _,
208                } => JsJson::Null,
209                CommandForBrowser::LocationSet {
210                    target: _,
211                    mode: _,
212                    value: _,
213                } => JsJson::Null,
214                CommandForBrowser::LocationGet { target: _ } => {
215                    let url = request.url.clone();
216                    browser_response::LocationGet { value: url }.to_json()
217                }
218                CommandForBrowser::CookieGet { name: _ } => {
219                    browser_response::CookieGet { value: "".into() }.to_json()
220                }
221                CommandForBrowser::CookieSet {
222                    name: _,
223                    value: _,
224                    expires_in: _,
225                } => JsJson::Null,
226                CommandForBrowser::CookieJsonGet { name: _ } => browser_response::CookieJsonGet {
227                    value: JsJson::Null,
228                }
229                .to_json(),
230                CommandForBrowser::CookieJsonSet {
231                    name: _,
232                    value: _,
233                    expires_in: _,
234                } => JsJson::Null,
235                CommandForBrowser::GetEnv { name } => {
236                    let env_value = request.env(name);
237
238                    browser_response::GetEnv { value: env_value }.to_json()
239                }
240                CommandForBrowser::Log {
241                    kind,
242                    message,
243                    arg2: _,
244                    arg3: _,
245                    arg4: _,
246                } => {
247                    if kind == ConsoleLogLevel::Error {
248                        log::warn!("{message}");
249                    } else {
250                        log::info!("{message}");
251                    }
252
253                    JsJson::Null
254                }
255                CommandForBrowser::TimezoneOffset => {
256                    browser_response::TimezoneOffset { value: 0 }.to_json()
257                }
258                CommandForBrowser::HistoryBack => JsJson::Null,
259                CommandForBrowser::GetRandom { min, max: _ } => {
260                    browser_response::GetRandom { value: min }.to_json()
261                }
262                CommandForBrowser::JsApiCall { commands: _ } => JsJson::Null,
263                CommandForBrowser::DomBulkUpdate { commands } => {
264                    // Host work, but reached from inside a wasm call - so this is
265                    // phase-3 time measured within a phase-2 region, and `SsrTimings`
266                    // subtracts it back out. See the module docs in `timings.rs`.
267                    let blob_bytes = commands.len();
268                    let decode_mark = probe.start();
269
270                    match decode_dom_commands(&commands) {
271                        Ok(list) => {
272                            // Before the send, so the channel push is not counted as
273                            // decoding.
274                            probe.decoded(decode_mark, blob_bytes, list.len());
275
276                            sender
277                                .send(Message::DomUpdate(list))
278                                .inspect_err(|err| log::error!("Error sending DomUpdate: {err}"))
279                                .unwrap_or_default();
280                        }
281                        Err(err) => log::error!("Error decoding DomBulkUpdate: {err}"),
282                    }
283
284                    JsJson::Null
285                }
286            }
287        });
288
289        let mut inst = WasmInstance::new(
290            &self.engine,
291            &self.instance_pre,
292            HostState {
293                request,
294                sender: sender.clone(),
295                probe: probe.clone(),
296                handle_command,
297            },
298        );
299        probe.instantiate(instantiate_mark);
300
301        // -- !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
302        //TODO - ultimately, do not call call_vertigo_entry_function if something is returned by handle_url
303        // -- !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
304
305        inst.call_vertigo_entry_function();
306
307        if let Some(result) = inst.handle_url(url) {
308            return result;
309        }
310
311        let spawn_resource = SpawnOwner::new({
312            let sender = sender.clone();
313
314            async move {
315                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
316                let _ = sender.send(Message::TimeoutAndSendResponse);
317            }
318        });
319
320        let mut html_response = HtmlResponse::new(
321            sender.clone(),
322            &self.mount_config,
323            inst,
324            self.mount_config.env.clone(),
325            fetch,
326            probe.clone(),
327        );
328
329        loop {
330            let message = receiver.try_recv();
331
332            match message {
333                Ok(message) => {
334                    if let Some(response) = html_response.process_message(message) {
335                        return response;
336                    };
337                    continue;
338                }
339                Err(TryRecvError::Empty) => {} // continue this iteration
340                Err(TryRecvError::Disconnected) => {
341                    break; // send response to browser
342                }
343            }
344
345            if html_response.awaiting_response() {
346                // Parked on an SSR fetch: time the request spent, but not time it spent
347                // working. Kept in its own bucket so it cannot be read as either.
348                let wait_mark = probe.start();
349                let message = receiver.recv().await;
350                probe.fetch_wait(wait_mark);
351
352                if let Some(message) = message
353                    && let Some(response) = html_response.process_message(message)
354                {
355                    return response;
356                };
357            } else {
358                break; // send response to browser
359            }
360        }
361
362        spawn_resource.off();
363        html_response.build_response()
364    }
365}
366
367fn build_module_wasm(
368    engine: &Engine,
369    mount_path: &MountConfig,
370) -> Result<(Module, Duration), ErrorCode> {
371    let full_wasm_path = mount_path.get_wasm_fs_path();
372
373    log::info!("Mounting {} -> {full_wasm_path}", mount_path.mount_point());
374
375    let wasm_content = match std::fs::read(&full_wasm_path) {
376        Ok(wasm_content) => wasm_content,
377        Err(error) => {
378            log::error!("Problem reading the path: wasm_path={full_wasm_path}, error={error}");
379            return Err(ErrorCode::ServeWasmReadFailed);
380        }
381    };
382
383    let now = Instant::now();
384
385    let module = match Module::from_binary(engine, &wasm_content) {
386        Ok(module) => module,
387        Err(err) => {
388            log::error!("Wasm compilation error: error={err}");
389            return Err(ErrorCode::ServeWasmCompileFailed);
390        }
391    };
392
393    let elapsed = now.elapsed();
394    log::info!("WASM module compiled in {} ms.", elapsed.as_millis());
395    Ok((module, elapsed))
396}
vertigo_cli/serve/server_state.rs

vertigo_cli/serve/
server_state.rs
