diff --git a/crates/connetto-client/src/lib.rs b/crates/connetto-client/src/lib.rs index 8b8b4ad9..b3b47412 100644 --- a/crates/connetto-client/src/lib.rs +++ b/crates/connetto-client/src/lib.rs @@ -242,6 +242,9 @@ pub enum ClientError { /// meets this keeps working and retries when a transport arrives. #[error("not connected: this operation needs a server")] NotConnected, + /// The client's pump stopped on a fault no reconnect can cure, named here. + #[error("the client stopped: {0}")] + Stopped(String), /// Acquiring or refreshing the access token failed. #[error("authentication error: {0}")] Auth(String), @@ -1156,6 +1159,13 @@ pub enum ClientEvent { /// Why the server closed the session. reason: FatalErrorReason, }, + /// The pump stopped on a fault no reconnect can cure, and said which. + /// [`Closed`](Self::Closed) follows, and every live handle's `changed` + /// returns [`ClientError::Stopped`] from then on. + Stopped { + /// The fault, as text. + detail: String, + }, /// The connection closed. Closed, /// The credential was rejected and refresh could not recover it, so the diff --git a/crates/connetto-client/src/live.rs b/crates/connetto-client/src/live.rs index ff505695..8a9c2425 100644 --- a/crates/connetto-client/src/live.rs +++ b/crates/connetto-client/src/live.rs @@ -25,7 +25,7 @@ use core::task::Poll; use core::time::Duration; use std::collections::{HashMap, HashSet}; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, Mutex as StdMutex, RwLock, Weak}; +use std::sync::{Arc, Mutex as StdMutex, OnceLock, RwLock, Weak}; use connetto_core::messages::{BindValue, SubscriptionSpec}; use connetto_core::quote_ident; @@ -654,10 +654,12 @@ fn run_probe(conn: &mut SqliteConnection, probe: &str) -> Result>, wake: Arc, + stopped: OnceLock, } /// What [`LiveQuery`] and [`LiveValue`] both are underneath: a subscription @@ -686,10 +688,15 @@ impl LiveHandleCore { } async fn changed(&mut self) -> Result<(), ClientError> { - self.changed - .changed() - .await - .map_err(|_| ClientError::Transport("live query driver stopped".to_owned())) + if let Some(detail) = self.reaper.stopped.get() { + return Err(ClientError::Stopped(detail.clone())); + } + self.changed.changed().await.map_err(|_| { + self.reaper.stopped.get().map_or_else( + || ClientError::Transport("live query driver stopped".to_owned()), + |detail| ClientError::Stopped(detail.clone()), + ) + }) } } @@ -1389,6 +1396,7 @@ where reaper: Arc::new(Reaper { pending: StdMutex::new(Vec::new()), wake: Arc::clone(&wake), + stopped: OnceLock::new(), }), events, next_live: AtomicU64::new(1), @@ -2733,7 +2741,7 @@ where /// A transport drop ends the pump, unless a reconnect driver is present, in /// which case the pump recovers: backoff, fresh transport, session resume, /// re-declared subscriptions. Local faults (session, apply, protocol) stay -/// terminal either way. +/// terminal either way and are announced as [`ClientEvent::Stopped`]. async fn pump( shared: Arc>, alive: Weak, @@ -2784,12 +2792,16 @@ async fn pump( let mut state = shared.state.lock().await; match drain_and_flush(&mut state, &shared, reconnect.is_some()).await { - PumpFlow::Proceed => {} - PumpFlow::Recover => { + Drained::Step => {} + Drained::Recover => { needs_recovery = true; continue; } - PumpFlow::Exit => return, + Drained::Fail(err) => { + drop(state); + pump_fail(&shared, err).await; + return; + } } // One cancellable pump step. A wake interrupts the idle wait so lock @@ -2830,11 +2842,12 @@ async fn pump( continue; } PumpFlow::Exit => { - // The server ended the session with no driver to resume it, so - // the writes made meanwhile are queued for the next run. - if let Err(err) = state.conn.flush().await { - tracing::warn!(error = %err, "queueing the last writes at exit failed"); - } + end_unresumed(&mut state, &shared).await; + return; + } + PumpFlow::Fail(err) => { + drop(state); + pump_fail(&shared, err).await; return; } } @@ -2969,8 +2982,63 @@ enum PumpFlow { Proceed, /// The transport is gone. Recover at the top of the next iteration. Recover, - /// The pump has ended. + /// The session ended with no driver to resume it. Exit, + /// A fault no reconnect can cure ended the pump. + Fail(ClientError), +} + +/// What the pump does once dropped handles are drained and local writes +/// flushed. A session only ends on a step, so draining never ends the pump. +enum Drained { + /// Take the pump step. + Step, + /// The transport is gone. Recover at the top of the next iteration. + Recover, + /// A fault no reconnect can cure ended the pump. + Fail(ClientError), +} + +/// Queue the writes made meanwhile for the next run, then end the pump, when +/// the session ended with no driver to resume it. +async fn end_unresumed(state: &mut State, shared: &Shared) +where + T: Transport + MaybeSend + 'static, + T::Error: core::fmt::Display, +{ + if let Err(err) = state.conn.flush().await { + tracing::warn!(error = %err, "queueing the last writes at exit failed"); + } + pump_finished(shared); +} + +/// End the pump on a fault, queueing what can be queued, closing the +/// transport, releasing every live handle with the fault, then announcing it +/// and `Closed`. +async fn pump_fail(shared: &Arc>, err: ClientError) +where + T: Transport + MaybeSend + 'static, + T::Error: core::fmt::Display, +{ + let detail = err.to_string(); + tracing::warn!(error = %detail, "the client pump stopped on a fault"); + // Set before the senders drop, so a woken handle reads the fault. + let _ = shared.reaper.stopped.set(detail.clone()); + let mut state = shared.state.lock().await; + if let Err(err) = state.conn.flush().await { + tracing::warn!(error = %err, "queueing the last writes at the fault failed"); + } + let _ = state.conn.close().await; + // The closed socket is gone, so nothing later sends on it or reads the + // connection as live. + state.conn.disconnect(); + state.registry.clear(); + state.values.clear(); + state.computed.clear(); + drop(state); + let _ = shared.events.send(ClientEvent::Stopped { detail }); + let _ = shared.events.send(ClientEvent::Closed); + pump_finished(shared); } /// Queue the last writes, close the transport, announce `Closed`, and end @@ -3030,7 +3098,7 @@ async fn drain_and_flush( state: &mut State, shared: &Arc>, has_driver: bool, -) -> PumpFlow +) -> Drained where T: Transport + MaybeSend + 'static, T::Error: core::fmt::Display, @@ -3041,27 +3109,25 @@ where // nothing re-declares it. if let Err(err) = drain_dropped(&mut *state, &shared.reaper).await { if is_disconnect(&err) { - return PumpFlow::Recover; + return Drained::Recover; } - pump_finished(shared); - return PumpFlow::Exit; + return Drained::Fail(err); } // Auto-submit local writes committed since the last step. With no socket // this only queues them durably, which must not wait for a server. if let Err(err) = state.conn.flush().await { if is_disconnect(&err) { - return PumpFlow::Recover; + return Drained::Recover; } - pump_finished(shared); - return PumpFlow::Exit; + return Drained::Fail(err); } // No socket and a driver to find one, so go and find it. With no driver // there is nothing to recover to, so the pump falls through and parks, // which keeps device-private queries refreshing with no server at all. if has_driver && !state.conn.is_connected() { - return PumpFlow::Recover; + return Drained::Recover; } - PumpFlow::Proceed + Drained::Step } /// Route one pump step's result. A deliberate close tells the app why and @@ -3086,7 +3152,6 @@ where PumpFlow::Recover } else { let _ = shared.events.send(ClientEvent::Closed); - pump_finished(shared); PumpFlow::Exit } } @@ -3095,7 +3160,6 @@ where PumpFlow::Recover } else { let _ = shared.events.send(ClientEvent::Closed); - pump_finished(shared); PumpFlow::Exit } } @@ -3121,8 +3185,7 @@ where if is_disconnect(&err) { PumpFlow::Recover } else { - pump_finished(shared); - PumpFlow::Exit + PumpFlow::Fail(err) } } } diff --git a/crates/connetto-client/tests/it/main.rs b/crates/connetto-client/tests/it/main.rs index e53c11ec..b58c0cdc 100644 --- a/crates/connetto-client/tests/it/main.rs +++ b/crates/connetto-client/tests/it/main.rs @@ -68,6 +68,9 @@ mod never_synced; mod offline_start; +#[cfg(feature = "native-auth")] +mod pump_fault; + mod reconnect_live; mod residual; diff --git a/crates/connetto-client/tests/it/pump_fault.rs b/crates/connetto-client/tests/it/pump_fault.rs new file mode 100644 index 00000000..3f263fab --- /dev/null +++ b/crates/connetto-client/tests/it/pump_fault.rs @@ -0,0 +1,477 @@ +//! A pump that stops on a fault it cannot recover from says why and releases +//! everything waiting on it. + +use connetto_client::{ClientBuilder, ClientError, ClientEvent, ConnettoClient, DataDir}; +use connetto_core::messages::{ + BulkMessage, ControlMessage, FatalError, FatalErrorReason, HandshakeAck, LivePatch, +}; +use connetto_core::traits::{IncomingFrame, Transport}; +use diesel::connection::SimpleConnection; +use diesel::prelude::*; +use parking_lot::Mutex; +use std::collections::VecDeque; +use std::future::ready; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::Duration; +use tempfile::tempdir; + +const DDL: &str = "CREATE TABLE items (id INTEGER PRIMARY KEY, label TEXT)"; +const BOUND: Duration = Duration::from_secs(5); + +diesel::table! { + /// Synced test table. + items (id) { + /// Item identifier, the primary key + id -> Integer, + /// Optional item label + label -> Nullable, + } +} + +#[derive(Debug, Clone, PartialEq, Queryable)] +struct Item { + id: i32, + label: Option, +} + +/// A transport handing out queued frames, idling when none are queued, and +/// recording the subscriptions it was sent and whether it was closed. Once +/// told to drop, every send fails the way a dead socket does. +#[derive(Clone, Default)] +struct Script { + frames: Arc>>, + subscribed: Arc>>, + closed: Arc, + dropped: Arc, +} + +/// The failure a send reports once the socket is gone. +#[derive(Debug, thiserror::Error)] +#[error("the socket is gone")] +struct Gone; + +impl Script { + fn say(&self, frame: IncomingFrame) { + self.frames.lock().push_back(frame); + } + + fn last_subscribed(&self) -> String { + self.subscribed + .lock() + .last() + .cloned() + .expect("the client subscribed") + } + + /// A send fails once the socket is dropped or closed, as a real one does. + fn sent(&self) -> Result<(), Gone> { + if self.dropped.load(Ordering::Acquire) || self.closed.load(Ordering::Acquire) { + Err(Gone) + } else { + Ok(()) + } + } +} + +impl Transport for Script { + type Error = Gone; + + fn send_control( + &mut self, + message: ControlMessage, + ) -> impl Future> { + if let ControlMessage::Subscribe(subscribe) = message { + self.subscribed.lock().push(subscribe.sub_id); + } + ready(self.sent()) + } + + fn send_bulk( + &mut self, + _message: BulkMessage, + ) -> impl Future> { + ready(self.sent()) + } + + fn recv(&mut self) -> impl Future, Self::Error>> { + let frames = Arc::clone(&self.frames); + async move { + loop { + if let Some(frame) = frames.lock().pop_front() { + return Ok(Some(frame)); + } + tokio::task::yield_now().await; + } + } + } + + fn close(&mut self) -> impl Future> { + self.closed.store(true, Ordering::Release); + ready(Ok(())) + } +} + +fn ack() -> IncomingFrame { + IncomingFrame::Control(ControlMessage::HandshakeAck(HandshakeAck { + connection_id: "script".to_owned(), + session_token: "script".to_owned(), + resume_token: "script".to_owned(), + current_cursor: connetto_core::Cursor::from(Vec::new()), + schema_version: None, + initial_credits: 64, + last_applied_seq: None, + })) +} + +async fn client() -> ( + ConnettoClient