diff --git a/backend/app/src/main.rs b/backend/app/src/main.rs index 6a718751f..c6457735c 100644 --- a/backend/app/src/main.rs +++ b/backend/app/src/main.rs @@ -418,6 +418,28 @@ async fn print_info(app_config: &AppConfig, loaded_env: Option<&Path>) { if let Some(metadata_update) = config.metadata_update.as_ref() { info!("Metadata path: {}", metadata_update.cache_path); } + if let Some(recording) = config.recording() { + // Nothing bounds recording disk use unless at least one of the three + // policies is set. Worth a warning, not an error: a dedicated + // filesystem is a legitimate reason to run without any of them. + let has_policy_retention = recording.retention.as_ref().is_some_and(|retention| { + retention.keep_last_per_channel.is_some() || retention.delete_after_days.is_some() + }); + let has_watermarks = recording + .disk + .as_ref() + .is_some_and(|disk| disk.high_water_percent.is_some() && disk.low_water_percent.is_some()); + let has_quota = recording.quota.as_ref().is_some_and(|quota| { + quota.default_private_bytes.is_some() || quota.shared_bytes.is_some() || !quota.per_user_bytes.is_empty() + }); + if recording.enabled && !has_policy_retention && !has_watermarks && !has_quota { + warn!( + "recording is enabled with no retention, no disk watermarks, and no quota; \ + recording disk usage is unbounded" + ); + } + } + config_loader::runtime_config_report::log_runtime_config_report(app_config).await; } diff --git a/frontend/src/app/components/authentication.rs b/frontend/src/app/components/authentication.rs index 7a7d63df9..4ce6d458e 100644 --- a/frontend/src/app/components/authentication.rs +++ b/frontend/src/app/components/authentication.rs @@ -135,6 +135,7 @@ pub fn Authentication(props: &AuthenticationProps) -> Html { let now_secs = js_sys::Date::now() / 1000.0; let delay_ms = compute_session_expiry_delay_ms(exp, now_secs as i64, SESSION_EXPIRY_SKEW_SECS); timeout = Some(Timeout::new(delay_ms, move || { + services_ctx.websocket.disconnect(); services_ctx.auth.logout(); services_ctx.toastr.warning(translate.t("MESSAGES.SESSION.EXPIRED")); })); diff --git a/frontend/src/hooks/use_log_stream.rs b/frontend/src/hooks/use_log_stream.rs index f9cd7c5ac..94f5c0582 100644 --- a/frontend/src/hooks/use_log_stream.rs +++ b/frontend/src/hooks/use_log_stream.rs @@ -1,7 +1,9 @@ use crate::{ - services::{get_base_href, get_token}, - utils::set_timeout, + hooks::use_service_context, + model::EventMessage, + services::{get_base_href, get_token, ActiveSocket, EventService}, }; +use gloo_timers::callback::Timeout; use log::{error, trace, warn}; use shared::{ model::{LogEntry, LogLevel, LogWsMessage}, @@ -11,23 +13,198 @@ use std::{ cell::{Cell, RefCell}, rc::Rc, }; -use wasm_bindgen::{closure::Closure, JsCast}; +use wasm_bindgen::closure::Closure; use web_sys::{CloseEvent, ErrorEvent, Event, MessageEvent, WebSocket}; use yew::prelude::*; -const WS_RECONNECT_BASE_MS: i32 = 300; -const WS_RECONNECT_MAX_MS: i32 = 3000; +const WS_RECONNECT_BASE_MS: u32 = 300; +const WS_RECONNECT_MAX_MS: u32 = 3000; const DEFAULT_MAX_LOG_LINES: usize = 2000; -fn reconnect_delay(attempt: u16) -> i32 { +fn reconnect_delay(attempt: u16) -> u32 { if attempt < 6 { - let d = WS_RECONNECT_BASE_MS * (i32::from(attempt) + 1); + let d = WS_RECONNECT_BASE_MS * (u32::from(attempt) + 1); d.min(WS_RECONNECT_MAX_MS) } else { WS_RECONNECT_MAX_MS } } +/// Owns a log subscription's socket and timer; callbacks only hold weak references. +struct LogStream { + enabled: Cell, + epoch: Cell, + attempt: Cell, + min_level: Cell>, + socket: RefCell>, + reconnect_timer: RefCell>, + connected: Callback, + message: Callback, + events: Rc, +} + +impl LogStream { + fn new(connected: Callback, message: Callback, events: Rc) -> Rc { + Rc::new(Self { + enabled: Cell::new(false), + epoch: Cell::new(0), + attempt: Cell::new(0), + min_level: Cell::new(None), + socket: RefCell::new(None), + reconnect_timer: RefCell::new(None), + connected, + message, + events, + }) + } + + fn disconnect(&self) { + self.enabled.set(false); + self.epoch.set(self.epoch.get().wrapping_add(1)); + self.attempt.set(0); + self.close_socket(); + } + + fn close_socket(&self) { + let timer = self.reconnect_timer.borrow_mut().take(); + drop(timer); + // Release the borrow before freeing any currently executing handler. + let socket = self.socket.borrow_mut().take(); + drop(socket); + } + + fn connect(self: &Rc) { + if !self.enabled.get() { + return; + } + if self + .socket + .borrow() + .as_ref() + .is_some_and(|socket| matches!(socket.websocket().ready_state(), WebSocket::CONNECTING | WebSocket::OPEN)) + { + return; + } + self.close_socket(); + // Logout removes the token before the hook's effect cleanup runs. + let Some(token) = get_token() else { + self.disconnect(); + self.connected.emit(false); + return; + }; + let epoch = self.epoch.get().wrapping_add(1); + self.epoch.set(epoch); + let path = concat_path_leading_slash(&get_base_href(), "ws/logs"); + let ws = match WebSocket::new(&format!("{path}?token={token}")) { + Ok(ws) => ws, + Err(err) => { + error!("Failed to create log websocket: {err:?}"); + return; + } + }; + let socket = ActiveSocket::new( + ws, + Closure::new(self.handler(epoch, Self::on_message)), + Closure::new(self.handler(epoch, Self::on_open)), + Closure::new(self.handler(epoch, Self::on_close)), + Closure::new(self.handler(epoch, Self::on_error)), + ); + *self.socket.borrow_mut() = Some(socket); + } + + fn handler(self: &Rc, epoch: u64, handle: fn(&Rc, E)) -> impl FnMut(E) + 'static { + let stream = Rc::downgrade(self); + move |event| { + if let Some(stream) = stream.upgrade() { + if stream.enabled.get() && stream.epoch.get() == epoch { + handle(&stream, event); + } + } + } + } + + fn send(&self, message: &LogWsMessage) { + if let Some(socket) = self.socket.borrow().as_ref() { + if socket.websocket().ready_state() == WebSocket::OPEN { + if let Ok(json) = serde_json::to_string(message) { + let _ = socket.websocket().send_with_str(&json); + } + } + } + } + + fn on_open(self: &Rc, _: Event) { + trace!("Log WebSocket connected"); + if let Some(token) = get_token() { + self.send(&LogWsMessage::Auth(token)); + self.send(&LogWsMessage::Filter { min_level: self.min_level.get() }); + } else { + self.disconnect(); + self.connected.emit(false); + } + } + + fn on_message(self: &Rc, event: MessageEvent) { + let Some(text) = event.data().as_string() else { + return; + }; + let Ok(message) = serde_json::from_str::(&text) else { + return; + }; + match &message { + LogWsMessage::Unauthorized => { + warn!("Log WebSocket unauthorized"); + self.disconnect(); + self.connected.emit(false); + self.events.broadcast(EventMessage::Unauthorized); + return; + } + LogWsMessage::Authorized | LogWsMessage::History(_) | LogWsMessage::Entry(_) => { + // Only successful authorization resets backoff, not a rejected upgrade. + self.attempt.set(0); + self.connected.emit(true); + } + _ => {} + } + self.message.emit(message); + } + + fn on_close(self: &Rc, _: CloseEvent) { + trace!("Log WebSocket closed"); + let epoch = self.epoch.get(); + self.close_socket(); + self.connected.emit(false); + if self.epoch.get() == epoch { + self.schedule_reconnect(); + } + } + + fn on_error(self: &Rc, event: ErrorEvent) { + trace!("Log WebSocket error: {event:?}"); + self.connected.emit(false); + } + + fn schedule_reconnect(self: &Rc) { + if !self.enabled.get() || self.reconnect_timer.borrow().is_some() { + return; + } + let attempt = self.attempt.get(); + self.attempt.set(attempt.saturating_add(1)); + let epoch = self.epoch.get(); + let stream = Rc::downgrade(self); + *self.reconnect_timer.borrow_mut() = Some(Timeout::new(reconnect_delay(attempt), move || { + let Some(stream) = stream.upgrade() else { + return; + }; + let timer = stream.reconnect_timer.borrow_mut().take(); + drop(timer); + if stream.enabled.get() && stream.epoch.get() == epoch { + stream.connect(); + } + })); + } +} + pub struct UseLogStreamOptions { pub active: bool, pub max_lines: usize, @@ -48,261 +225,234 @@ pub struct UseLogStreamHandle { #[hook] pub fn use_log_stream(options: UseLogStreamOptions) -> UseLogStreamHandle { + let services = use_service_context(); let connected = use_state(|| false); let logs = use_state(|| Rc::new(Vec::::new())); let min_level = use_state(|| None::); + let max_lines = use_mut_ref(|| options.max_lines); + *max_lines.borrow_mut() = options.max_lines; + // WebSocket events can arrive before Yew renders the previous state update. + let log_buffer = use_mut_ref(|| Rc::new(Vec::::new())); + let stream = use_mut_ref({ + let connected = connected.setter(); + let logs = logs.setter(); + let buffer = Rc::clone(&log_buffer); + let max_lines = Rc::clone(&max_lines); + move || { + LogStream::new( + Callback::from(move |value| connected.set(value)), + Callback::from(move |message| { + let updated = { + let mut buffer = buffer.borrow_mut(); + match message { + LogWsMessage::History(entries) => *buffer = Rc::new(entries), + LogWsMessage::Entry(entry) => Rc::make_mut(&mut buffer).push(entry), + _ => return, + } + let max_lines = *max_lines.borrow(); + let list = Rc::make_mut(&mut buffer); + let excess = list.len().saturating_sub(max_lines); + list.drain(..excess); + Rc::clone(&buffer) + }; + logs.set(updated); + }), + Rc::clone(&services.event), + ) + } + }); + let stream = Rc::clone(&stream.borrow()); - let ws_cell = use_mut_ref(|| None::); - let attempt_cell = use_mut_ref(|| 0u16); - let active_cell = use_mut_ref(|| options.active); - let max_lines = options.max_lines; - - *active_cell.borrow_mut() = options.active; - - // Clear callback let clear = { - let logs = logs.clone(); + let logs = logs.setter(); Callback::from(move |()| { - logs.set(Rc::new(Vec::new())); + let empty = Rc::new(Vec::new()); + *log_buffer.borrow_mut() = Rc::clone(&empty); + logs.set(empty); }) }; - // Set min level callback let set_min_level = { - let min_level_state = min_level.clone(); - let ws_cell = ws_cell.clone(); + let min_level_state = min_level.setter(); + let stream = Rc::clone(&stream); Callback::from(move |lvl: Option| { min_level_state.set(lvl); - if let Some(ws) = ws_cell.borrow().as_ref() { - if ws.ready_state() == WebSocket::OPEN { - let msg = LogWsMessage::Filter { min_level: lvl }; - if let Ok(json) = serde_json::to_string(&msg) { - let _ = ws.send_with_str(&json); - } - } - } + stream.min_level.set(lvl); + stream.send(&LogWsMessage::Filter { min_level: lvl }); }) }; { - let connected = connected.clone(); - let logs = logs.clone(); - let min_level = min_level.clone(); - let ws_cell = ws_cell.clone(); - let attempt_cell = attempt_cell.clone(); - let active = options.active; - - use_effect_with(active, move |&active| { - if !active { - if let Some(ws) = ws_cell.borrow_mut().take() { - ws.set_onmessage(None); - ws.set_onopen(None); - ws.set_onerror(None); - ws.set_onclose(None); - let _ = ws.close(); - } + let connected = connected.setter(); + use_effect_with(options.active, move |&active| { + if active { + stream.enabled.set(true); + stream.connect(); + } else { + stream.disconnect(); connected.set(false); - return Box::new(|| {}) as Box; } - - let is_alive = Rc::new(Cell::new(true)); - - type ConnectFn = Rc; - type ConnectFnCell = Rc>>; - - let connect_fn: ConnectFnCell = Rc::new(RefCell::new(None)); - let connect_fn_clone = connect_fn.clone(); - - let do_connect = { - let connected = connected.clone(); - let logs = logs.clone(); - let min_level = min_level.clone(); - let ws_cell = ws_cell.clone(); - let attempt_cell = attempt_cell.clone(); - let is_alive = is_alive.clone(); - let connect_fn = connect_fn_clone.clone(); - - Rc::new(move || { - if !is_alive.get() { - return; - } - - if let Some(ws) = ws_cell.borrow().as_ref() { - let state = ws.ready_state(); - if state == WebSocket::CONNECTING || state == WebSocket::OPEN { - return; - } - } - - let base_href = get_base_href(); - let path = concat_path_leading_slash(&base_href, "ws/logs"); - let token_opt = get_token(); - let ws_url = if let Some(ref token) = token_opt { format!("{path}?token={token}") } else { path }; - - match WebSocket::new(&ws_url) { - Err(err) => { - error!("Failed to create log websocket: {err:?}"); - let attempt = *attempt_cell.borrow(); - *attempt_cell.borrow_mut() = attempt.saturating_add(1); - let delay = reconnect_delay(attempt); - let connect_fn_inner = connect_fn.clone(); - set_timeout( - move || { - if let Some(f) = connect_fn_inner.borrow().as_ref() { - f(); - } - }, - delay, - ); - } - Ok(socket) => { - *ws_cell.borrow_mut() = Some(socket.clone()); - - // onopen - { - let socket_clone = socket.clone(); - let token_opt = token_opt.clone(); - let min_level_val = *min_level; - let attempt_cell = attempt_cell.clone(); - let is_alive = is_alive.clone(); - - let onopen = Closure::::wrap(Box::new(move |_| { - if !is_alive.get() { - return; - } - trace!("Log WebSocket connected"); - *attempt_cell.borrow_mut() = 0; - - if let Some(ref token) = token_opt { - let auth_msg = LogWsMessage::Auth(token.clone()); - if let Ok(json) = serde_json::to_string(&auth_msg) { - let _ = socket_clone.send_with_str(&json); - } - } - - if let Some(lvl) = min_level_val { - let filter_msg = LogWsMessage::Filter { min_level: Some(lvl) }; - if let Ok(json) = serde_json::to_string(&filter_msg) { - let _ = socket_clone.send_with_str(&json); - } - } - })); - socket.set_onopen(Some(onopen.as_ref().unchecked_ref())); - onopen.forget(); - } - - // onmessage - { - let connected = connected.clone(); - let logs = logs.clone(); - let is_alive = is_alive.clone(); - - let onmessage = - Closure::::wrap(Box::new(move |e: MessageEvent| { - if !is_alive.get() { - return; - } - if let Ok(text) = e.data().dyn_into::() { - let text_str: String = text.into(); - if let Ok(msg) = serde_json::from_str::(&text_str) { - match msg { - LogWsMessage::Authorized => { - connected.set(true); - } - LogWsMessage::Unauthorized => { - connected.set(false); - warn!("Log WebSocket unauthorized"); - } - LogWsMessage::History(entries) => { - connected.set(true); - let mut new_logs = entries; - if new_logs.len() > max_lines { - let skip = new_logs.len() - max_lines; - new_logs.drain(0..skip); - } - logs.set(Rc::new(new_logs)); - } - LogWsMessage::Entry(entry) => { - connected.set(true); - logs.set({ - let mut list = (**logs).clone(); - list.push(entry); - if list.len() > max_lines { - list.remove(0); - } - Rc::new(list) - }); - } - _ => {} - } - } - } - })); - socket.set_onmessage(Some(onmessage.as_ref().unchecked_ref())); - onmessage.forget(); - } - - // onclose - { - let connected = connected.clone(); - let ws_cell = ws_cell.clone(); - let attempt_cell = attempt_cell.clone(); - let is_alive = is_alive.clone(); - let connect_fn_inner = connect_fn.clone(); - - let onclose = Closure::::wrap(Box::new(move |_| { - if !is_alive.get() { - return; - } - trace!("Log WebSocket closed"); - connected.set(false); - *ws_cell.borrow_mut() = None; - - let attempt = *attempt_cell.borrow(); - *attempt_cell.borrow_mut() = attempt.saturating_add(1); - let delay = reconnect_delay(attempt); - let connect_fn_reconnect = connect_fn_inner.clone(); - set_timeout( - move || { - if let Some(f) = connect_fn_reconnect.borrow().as_ref() { - f(); - } - }, - delay, - ); - })); - socket.set_onclose(Some(onclose.as_ref().unchecked_ref())); - onclose.forget(); - } - - // onerror - { - let onerror = Closure::::wrap(Box::new(move |e: ErrorEvent| { - trace!("Log WebSocket error: {e:?}"); - })); - socket.set_onerror(Some(onerror.as_ref().unchecked_ref())); - onerror.forget(); - } - } - } - }) - }; - - *connect_fn.borrow_mut() = Some(do_connect.clone()); - do_connect(); - - Box::new(move || { - is_alive.set(false); - if let Some(ws) = ws_cell.borrow_mut().take() { - ws.set_onmessage(None); - ws.set_onopen(None); - ws.set_onerror(None); - ws.set_onclose(None); - let _ = ws.close(); - } - }) as Box + move || stream.disconnect() }); } UseLogStreamHandle { connected: *connected, logs: (*logs).clone(), min_level: *min_level, set_min_level, clear } } + +#[cfg(all(test, target_arch = "wasm32"))] +mod browser_tests { + use super::{reconnect_delay, LogStream}; + use crate::{ + model::EventMessage, + services::{get_token, set_token, EventService}, + }; + use gloo_timers::future::TimeoutFuture; + use shared::model::{LogWsMessage, TOKEN_NO_AUTH}; + use std::{cell::Cell, rc::Rc}; + use wasm_bindgen::JsValue; + use wasm_bindgen_test::wasm_bindgen_test; + use web_sys::{CloseEvent, MessageEvent, MessageEventInit, WebSocket}; + use yew::Callback; + + struct SavedToken(Option); + + impl SavedToken { + fn new() -> Self { + let saved = Self(get_token()); + set_token(Some(TOKEN_NO_AUTH)); + saved + } + } + + impl Drop for SavedToken { + fn drop(&mut self) { set_token(self.0.as_deref()); } + } + + fn stream(events: Rc) -> Rc { LogStream::new(Callback::noop(), Callback::noop(), events) } + + fn socket(stream: &LogStream) -> Result { + stream + .socket + .borrow() + .as_ref() + .map(|socket| socket.websocket().clone()) + .ok_or_else(|| JsValue::from_str("missing log socket")) + } + + fn assert_detached(socket: &WebSocket) { + assert!(socket.onmessage().is_none()); + assert!(socket.onopen().is_none()); + assert!(socket.onclose().is_none()); + assert!(socket.onerror().is_none()); + } + + #[wasm_bindgen_test] + fn unauthorized_callback_detaches_itself_and_stops_reconnecting() -> Result<(), JsValue> { + let _saved = SavedToken::new(); + let events = Rc::new(EventService::new()); + let unauthorized = Rc::new(Cell::new(0)); + let observed = Rc::clone(&unauthorized); + events.subscribe(move |event| { + if event == EventMessage::Unauthorized { + observed.set(observed.get() + 1); + } + }); + let stream = stream(events); + stream.enabled.set(true); + stream.connect(); + let socket = socket(&stream)?; + let callback = socket.onmessage().ok_or_else(|| JsValue::from_str("missing message callback"))?; + let init = MessageEventInit::new(); + let json = + serde_json::to_string(&LogWsMessage::Unauthorized).map_err(|err| JsValue::from_str(&err.to_string()))?; + init.set_data(&JsValue::from_str(&json)); + callback.call1(&socket, MessageEvent::new_with_event_init_dict("message", &init)?.as_ref())?; + + assert_eq!(unauthorized.get(), 1); + assert!(!stream.enabled.get()); + assert!(stream.socket.borrow().is_none()); + assert!(stream.reconnect_timer.borrow().is_none()); + assert_detached(&socket); + Ok(()) + } + + #[wasm_bindgen_test] + async fn close_callback_reconnect_is_canceled_when_the_hook_unmounts() -> Result<(), JsValue> { + let _saved = SavedToken::new(); + let stream = stream(Rc::new(EventService::new())); + stream.enabled.set(true); + stream.connect(); + let socket = socket(&stream)?; + let callback = socket.onclose().ok_or_else(|| JsValue::from_str("missing close callback"))?; + callback.call1(&socket, CloseEvent::new("close")?.as_ref())?; + assert_detached(&socket); + assert!(stream.socket.borrow().is_none()); + assert!(stream.reconnect_timer.borrow().is_some()); + + stream.disconnect(); + let epoch = stream.epoch.get(); + assert!(stream.reconnect_timer.borrow().is_none()); + TimeoutFuture::new(reconnect_delay(0) + 50).await; + assert!(stream.socket.borrow().is_none()); + assert_eq!(stream.epoch.get(), epoch); + Ok(()) + } + + #[wasm_bindgen_test] + async fn pending_reconnect_does_not_open_a_socket_after_logout() { + let _saved = SavedToken::new(); + let stream = stream(Rc::new(EventService::new())); + stream.enabled.set(true); + stream.schedule_reconnect(); + set_token(None); + TimeoutFuture::new(reconnect_delay(0) + 50).await; + assert!(!stream.enabled.get()); + assert!(stream.socket.borrow().is_none()); + assert!(stream.reconnect_timer.borrow().is_none()); + } + + #[wasm_bindgen_test] + fn socket_and_timer_do_not_keep_an_unmounted_stream_alive() -> Result<(), JsValue> { + let _saved = SavedToken::new(); + let stream = stream(Rc::new(EventService::new())); + let weak = Rc::downgrade(&stream); + stream.enabled.set(true); + stream.connect(); + let socket = socket(&stream)?; + stream.schedule_reconnect(); + stream.schedule_reconnect(); + assert_eq!(stream.attempt.get(), 1); + drop(stream); + assert!(weak.upgrade().is_none()); + assert_detached(&socket); + Ok(()) + } + + #[wasm_bindgen_test] + fn callbacks_from_an_old_subscription_cannot_change_a_new_subscription() -> Result<(), JsValue> { + let _saved = SavedToken::new(); + let calls = Rc::new(Cell::new(0)); + let observed = Rc::clone(&calls); + let stream = LogStream::new( + Callback::from(move |_| observed.set(observed.get() + 1)), + Callback::noop(), + Rc::new(EventService::new()), + ); + stream.enabled.set(true); + stream.connect(); + let old_socket = socket(&stream)?; + let mut old_handler = stream.handler(stream.epoch.get(), LogStream::on_close); + stream.disconnect(); + stream.enabled.set(true); + stream.connect(); + let epoch = stream.epoch.get(); + old_handler(CloseEvent::new("close")?); + assert_eq!(calls.get(), 0); + assert_eq!(stream.epoch.get(), epoch); + assert!(stream.socket.borrow().is_some()); + assert!(stream.reconnect_timer.borrow().is_none()); + assert_detached(&old_socket); + stream.disconnect(); + Ok(()) + } +} diff --git a/frontend/src/services/websocket_service.rs b/frontend/src/services/websocket_service.rs index cc782bdc7..d978941cb 100644 --- a/frontend/src/services/websocket_service.rs +++ b/frontend/src/services/websocket_service.rs @@ -1,8 +1,8 @@ use crate::{ model::EventMessage, services::{get_base_href, get_token, EventService, StatusService}, - utils::set_timeout, }; +use gloo_timers::callback::Timeout; use log::{error, trace, warn}; use shared::{ model::{ProtocolMessage, PROTOCOL_VERSION}, @@ -84,12 +84,36 @@ fn enqueue_pending(pending: &mut VecDeque>, bytes: &[u8], idempotent: bo /// An open socket together with the JS handlers attached to it. /// Dropping it detaches the handlers, closes the socket and frees the closures. -struct ActiveSocket { +pub(crate) struct ActiveSocket { ws: WebSocket, - on_message_handler: Closure, - on_open_handler: Closure, - on_close_handler: Closure, - on_error_handler: Closure, + _on_message_handler: Closure, + _on_open_handler: Closure, + _on_close_handler: Closure, + _on_error_handler: Closure, +} + +impl ActiveSocket { + pub(crate) fn new( + ws: WebSocket, + on_message_handler: Closure, + on_open_handler: Closure, + on_close_handler: Closure, + on_error_handler: Closure, + ) -> Self { + ws.set_onmessage(Some(on_message_handler.as_ref().unchecked_ref())); + ws.set_onopen(Some(on_open_handler.as_ref().unchecked_ref())); + ws.set_onclose(Some(on_close_handler.as_ref().unchecked_ref())); + ws.set_onerror(Some(on_error_handler.as_ref().unchecked_ref())); + Self { + ws, + _on_message_handler: on_message_handler, + _on_open_handler: on_open_handler, + _on_close_handler: on_close_handler, + _on_error_handler: on_error_handler, + } + } + + pub(crate) fn websocket(&self) -> &WebSocket { &self.ws } } impl Drop for ActiveSocket { @@ -110,6 +134,7 @@ pub struct WebSocketInner { connection_epoch: Cell, attempt_counter: Cell, socket: RefCell>, + reconnect_timer: RefCell>, pending_messages: RefCell>>, status_service: Rc, event_service: Rc, @@ -135,6 +160,7 @@ impl WebSocketService { connection_epoch: Cell::new(0), attempt_counter: Cell::new(0), socket: RefCell::new(None), + reconnect_timer: RefCell::new(None), pending_messages: RefCell::new(VecDeque::new()), status_service, event_service, @@ -172,6 +198,8 @@ impl WebSocketService { } fn close_socket(&self) { + let timer = self.reconnect_timer.borrow_mut().take(); + drop(timer); // Drop outside the borrow; this may free the handler that is currently running, // which wasm-bindgen defers until that invocation returns. let socket = self.socket.borrow_mut().take(); @@ -201,17 +229,13 @@ impl WebSocketService { self.connection_epoch.set(socket_epoch); ws.set_binary_type(web_sys::BinaryType::Arraybuffer); - let socket = ActiveSocket { - on_message_handler: Closure::new(self.socket_handler(socket_epoch, Self::on_message)), - on_open_handler: Closure::new(self.socket_handler(socket_epoch, Self::on_open)), - on_close_handler: Closure::new(self.socket_handler(socket_epoch, Self::on_close)), - on_error_handler: Closure::new(self.socket_handler(socket_epoch, Self::on_error)), + let socket = ActiveSocket::new( ws, - }; - socket.ws.set_onmessage(Some(socket.on_message_handler.as_ref().unchecked_ref())); - socket.ws.set_onopen(Some(socket.on_open_handler.as_ref().unchecked_ref())); - socket.ws.set_onclose(Some(socket.on_close_handler.as_ref().unchecked_ref())); - socket.ws.set_onerror(Some(socket.on_error_handler.as_ref().unchecked_ref())); + Closure::new(self.socket_handler(socket_epoch, Self::on_message)), + Closure::new(self.socket_handler(socket_epoch, Self::on_open)), + Closure::new(self.socket_handler(socket_epoch, Self::on_close)), + Closure::new(self.socket_handler(socket_epoch, Self::on_error)), + ); *self.socket.borrow_mut() = Some(socket); } @@ -223,7 +247,7 @@ impl WebSocketService { let Some(service) = service.upgrade().map(Self) else { return; }; - if socket_epoch_is_current(service.connection_epoch.get(), socket_epoch) { + if service.enabled.get() && socket_epoch_is_current(service.connection_epoch.get(), socket_epoch) { handle(&service, event); } } @@ -245,10 +269,14 @@ impl WebSocketService { fn on_close(&self, event: CloseEvent) { trace!("WebSocket closed (Code {}, Reason: {}, Clean: {})", event.code(), event.reason(), event.was_clean()); + let socket_epoch = self.connection_epoch.get(); self.close_socket(); self.connected.set(false); self.event_service.broadcast(EventMessage::WebSocketStatus(false)); - self.schedule_reconnect(); + // A subscriber may log out or start another connection during the broadcast. + if socket_epoch_is_current(self.connection_epoch.get(), socket_epoch) { + self.schedule_reconnect(); + } } fn on_error(&self, event: ErrorEvent) { @@ -258,7 +286,7 @@ impl WebSocketService { } fn schedule_reconnect(&self) { - if !self.enabled.get() { + if !self.enabled.get() || self.reconnect_timer.borrow().is_some() { return; } let attempt = self.attempt_counter.get().saturating_add(1); @@ -273,15 +301,17 @@ impl WebSocketService { warn!("WebSocket reconnect attempt #{attempt} scheduled in {delay} ms"); let socket_epoch = self.connection_epoch.get(); - let service = self.clone(); - set_timeout( - move || { - if socket_epoch_is_current(service.connection_epoch.get(), socket_epoch) { - service.connect_ws(); - } - }, - delay as i32, - ); + let service = Rc::downgrade(&self.0); + *self.reconnect_timer.borrow_mut() = Some(Timeout::new(delay, move || { + let Some(service) = service.upgrade().map(Self) else { + return; + }; + let timer = service.reconnect_timer.borrow_mut().take(); + drop(timer); + if service.enabled.get() && socket_epoch_is_current(service.connection_epoch.get(), socket_epoch) { + service.connect_ws(); + } + })); } fn try_send_message(&self, msg: &ProtocolMessage) -> bool { @@ -381,8 +411,7 @@ impl WebSocketService { let event_service = &self.event_service; match message { ProtocolMessage::Unauthorized => { - self.connected.set(false); - self.pending_messages.borrow_mut().clear(); + self.disconnect(); event_service.broadcast(EventMessage::Unauthorized); } ProtocolMessage::Error(err) => { @@ -543,7 +572,7 @@ mod browser_tests { use std::{cell::Cell, rc::Rc}; use wasm_bindgen::{JsCast, JsValue}; use wasm_bindgen_test::wasm_bindgen_test; - use web_sys::{js_sys::Uint8Array, MessageEvent, MessageEventInit, WebSocket}; + use web_sys::{js_sys::Uint8Array, CloseEvent, MessageEvent, MessageEventInit, WebSocket}; fn message_event(message: &ProtocolMessage) -> Result { let bytes = message.to_bytes().map_err(|err| JsValue::from_str(&err.to_string()))?; @@ -657,4 +686,45 @@ mod browser_tests { assert_eq!(websocket.connection_epoch.get(), epoch); websocket.disconnect(); } + + #[wasm_bindgen_test] + fn close_callback_cannot_schedule_reconnect_for_a_new_login() -> Result<(), JsValue> { + let events = Rc::new(EventService::new()); + let websocket = WebSocketService::new(Rc::new(StatusService::new()), Rc::clone(&events)); + websocket.connect_ws_with_backoff(); + let socket = current_socket(&websocket).ok_or_else(|| JsValue::from_str("missing websocket"))?; + let callback = socket.onclose().ok_or_else(|| JsValue::from_str("missing close callback"))?; + let reconnected = Rc::new(Cell::new(false)); + let observed = Rc::clone(&reconnected); + let service = websocket.clone(); + let subscription = events.subscribe(move |msg| { + if msg == EventMessage::WebSocketStatus(false) && !observed.replace(true) { + service.disconnect(); + service.connect_ws_with_backoff(); + } + }); + + callback.call1(&socket, CloseEvent::new("close")?.as_ref())?; + assert!(reconnected.get()); + assert!(websocket.socket.borrow().is_some()); + assert!(websocket.reconnect_timer.borrow().is_none()); + assert_eq!(websocket.attempt_counter.get(), 0); + assert!(socket.onclose().is_none()); + events.unsubscribe(subscription); + websocket.disconnect(); + Ok(()) + } + + #[wasm_bindgen_test] + fn reconnect_timer_is_unique_and_does_not_keep_service_alive() { + let websocket = WebSocketService::new(Rc::new(StatusService::new()), Rc::new(EventService::new())); + let weak = Rc::downgrade(&websocket.0); + websocket.enabled.set(true); + websocket.schedule_reconnect(); + websocket.schedule_reconnect(); + assert_eq!(websocket.attempt_counter.get(), 1); + assert!(websocket.reconnect_timer.borrow().is_some()); + drop(websocket); + assert!(weak.upgrade().is_none()); + } } diff --git a/shared/src/model/config/recording.rs b/shared/src/model/config/recording.rs index 2b9da09bb..ddd1081f1 100644 --- a/shared/src/model/config/recording.rs +++ b/shared/src/model/config/recording.rs @@ -683,26 +683,5 @@ pub(crate) fn prepare_recording_config(recording: &mut RecordingConfigDto) -> Re return Err(TuliproxError::ConfigRecording("recording.fallback_bytes_per_minute must be > 0".to_string())); } - // Nothing bounds recording disk use unless at least one of the three - // policies is set. Worth a warning, not an error: a dedicated - // filesystem is a legitimate reason to run without any of them. - let has_policy_retention = recording - .retention - .as_ref() - .is_some_and(|retention| retention.keep_last_per_channel.is_some() || retention.delete_after_days.is_some()); - let has_watermarks = recording - .disk - .as_ref() - .is_some_and(|disk| disk.high_water_percent.is_some() && disk.low_water_percent.is_some()); - let has_quota = recording.quota.as_ref().is_some_and(|quota| { - quota.default_private_bytes.is_some() || quota.shared_bytes.is_some() || !quota.per_user_bytes.is_empty() - }); - if recording.enabled && !has_policy_retention && !has_watermarks && !has_quota { - log::warn!( - "recording is enabled with no retention, no disk watermarks, and no quota; \ - recording disk usage is unbounded" - ); - } - Ok(()) }