Announce a pump fault and release the handles waiting on it

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

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<String, ClientE
 }
 
 /// Subscription ids and the wake signal shared with every [`LiveQuery`], so a
-/// synchronous `Drop` can queue its unsubscribe for the async pump.
+/// synchronous `Drop` can queue its unsubscribe for the async pump, and the
+/// fault that stopped the pump, if one did.
 struct Reaper {
     pending: StdMutex<Vec<String>>,
     wake: Arc<Notify>,
+    stopped: OnceLock<String>,
 }
 
 /// 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<T, F, S>(
     shared: Arc<Shared<T>>,
     alive: Weak<ClientToken>,
@@ -2784,12 +2792,16 @@ async fn pump<T, F, S>(
         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<T, F, S>(
                 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<T>(state: &mut State<T>, shared: &Shared<T>)
+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<T>(shared: &Arc<Shared<T>>, 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<T>(
     state: &mut State<T>,
     shared: &Arc<Shared<T>>,
     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)
             }
         }
     }

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;