feat(qconnect): recover from terminal states via idle-retry + startup backoff

Add `reconnect_idle_retry_ms` (default 0 = legacy terminate). When > 0, the
transport loop no longer breaks permanently on Exhausted: the new
`idle_retry_after_exhausted` helper idles (shutdown-cancellable), resets the
attempt counter / backoff to base, and retries (gap #7). The real config
sets 60s. The reset happens ONLY after Exhausted plus a deliberate idle
wait, never on connect, so the issue #358 latch is untouched.

Critical cross-file coupling: `service.rs`'s `MaxReconnectAttemptsExceeded`
handler previously dropped the runtime and broke the event loop. With
idle-retry active the transport stays alive and re-emits events, so the
handler now surfaces lifecycle Exhausted to the UI but keeps the runtime and
continues the loop (only the legacy path still drops + breaks). When the
re-armed transport later reaches SessionEstablished, the existing handler
moves the lifecycle back to Connected.

Startup auto-connect now retries on a bounded, monotonic schedule
(`startup_retry_schedule` = 2s/5s/15s/30s) instead of giving up after one
failed attempt (gap #8), re-resolving the transport config each iteration.

cc
This commit is contained in:
vicrodh committed 2026-06-01 03:50:18 -06:00
1 parent 7628c284a6
commit 85a811c834
6 files changed
+263 -44

No files matched your search

@@ -18,6 +18,12 @@ pub struct WsTransportConfig {
/// ///
/// `None` means unlimited (legacy behavior, retained for tests). /// `None` means unlimited (legacy behavior, retained for tests).
pub reconnect_max_attempts: Option<u32>, pub reconnect_max_attempts: Option<u32>,
/// When `> 0`, reaching `Exhausted` no longer terminates the transport
/// loop: it idles this long (shutdown-cancellable), resets the attempt
/// counter / backoff to base, and retries instead of giving up (gap #7).
/// Default `0` preserves the legacy terminate-on-exhausted behavior used
/// by tests; the real config sets 60s.
pub reconnect_idle_retry_ms: u64,
pub connect_timeout_ms: u64, pub connect_timeout_ms: u64,
pub keepalive_interval_ms: u64, pub keepalive_interval_ms: u64,
pub auto_subscribe: bool, pub auto_subscribe: bool,
@@ -34,6 +40,7 @@ impl Default for WsTransportConfig {
reconnect_backoff_ms: 2_000, reconnect_backoff_ms: 2_000,
reconnect_backoff_max_ms: 30_000, reconnect_backoff_max_ms: 30_000,
reconnect_max_attempts: Some(10), reconnect_max_attempts: Some(10),
reconnect_idle_retry_ms: 0,
connect_timeout_ms: 10_000, connect_timeout_ms: 10_000,
keepalive_interval_ms: 30_000, keepalive_interval_ms: 30_000,
auto_subscribe: true, auto_subscribe: true,
+145 -6
View File
@@ -356,7 +356,21 @@ async fn run_native_transport_loop(
.await .await
{ {
ReconnectOutcome::Continue => continue, ReconnectOutcome::Continue => continue,
ReconnectOutcome::Shutdown | ReconnectOutcome::Exhausted => break, ReconnectOutcome::Shutdown => break,
ReconnectOutcome::Exhausted => {
if idle_retry_after_exhausted(
&mut shutdown_rx,
&mut reconnect_attempt,
&mut backoff,
base_backoff,
config.reconnect_idle_retry_ms,
)
.await
{
continue;
}
break;
}
} }
} }
Err(_) => { Err(_) => {
@@ -372,7 +386,21 @@ async fn run_native_transport_loop(
.await .await
{ {
ReconnectOutcome::Continue => continue, ReconnectOutcome::Continue => continue,
ReconnectOutcome::Shutdown | ReconnectOutcome::Exhausted => break, ReconnectOutcome::Shutdown => break,
ReconnectOutcome::Exhausted => {
if idle_retry_after_exhausted(
&mut shutdown_rx,
&mut reconnect_attempt,
&mut backoff,
base_backoff,
config.reconnect_idle_retry_ms,
)
.await
{
continue;
}
break;
}
} }
} }
}; };
@@ -417,7 +445,21 @@ async fn run_native_transport_loop(
.await .await
{ {
ReconnectOutcome::Continue => continue, ReconnectOutcome::Continue => continue,
ReconnectOutcome::Shutdown | ReconnectOutcome::Exhausted => break, ReconnectOutcome::Shutdown => break,
ReconnectOutcome::Exhausted => {
if idle_retry_after_exhausted(
&mut shutdown_rx,
&mut reconnect_attempt,
&mut backoff,
base_backoff,
config.reconnect_idle_retry_ms,
)
.await
{
continue;
}
break;
}
} }
} }
emit(&events_tx, TransportEvent::Authenticated); emit(&events_tx, TransportEvent::Authenticated);
@@ -451,7 +493,21 @@ async fn run_native_transport_loop(
.await .await
{ {
ReconnectOutcome::Continue => continue, ReconnectOutcome::Continue => continue,
ReconnectOutcome::Shutdown | ReconnectOutcome::Exhausted => break, ReconnectOutcome::Shutdown => break,
ReconnectOutcome::Exhausted => {
if idle_retry_after_exhausted(
&mut shutdown_rx,
&mut reconnect_attempt,
&mut backoff,
base_backoff,
config.reconnect_idle_retry_ms,
)
.await
{
continue;
}
break;
}
} }
} }
@@ -489,7 +545,21 @@ async fn run_native_transport_loop(
.await .await
{ {
ReconnectOutcome::Continue => continue, ReconnectOutcome::Continue => continue,
ReconnectOutcome::Shutdown | ReconnectOutcome::Exhausted => break, ReconnectOutcome::Shutdown => break,
ReconnectOutcome::Exhausted => {
if idle_retry_after_exhausted(
&mut shutdown_rx,
&mut reconnect_attempt,
&mut backoff,
base_backoff,
config.reconnect_idle_retry_ms,
)
.await
{
continue;
}
break;
}
} }
} }
emit(&events_tx, TransportEvent::Subscribed); emit(&events_tx, TransportEvent::Subscribed);
@@ -635,7 +705,21 @@ async fn run_native_transport_loop(
.await .await
{ {
ReconnectOutcome::Continue => continue, ReconnectOutcome::Continue => continue,
ReconnectOutcome::Shutdown | ReconnectOutcome::Exhausted => break, ReconnectOutcome::Shutdown => break,
ReconnectOutcome::Exhausted => {
if idle_retry_after_exhausted(
&mut shutdown_rx,
&mut reconnect_attempt,
&mut backoff,
base_backoff,
config.reconnect_idle_retry_ms,
)
.await
{
continue;
}
break;
}
} }
} }
@@ -701,6 +785,39 @@ async fn handle_reconnect_delay(
ReconnectOutcome::Continue ReconnectOutcome::Continue
} }
/// Long-backoff idle retry after the reconnect loop has Exhausted its bounded
/// attempts (gap #7). When `idle_retry_ms == 0` this is a no-op returning
/// `false` (legacy terminate behavior). Otherwise it sleeps `idle_retry_ms`
/// (cancellable by shutdown); on a real shutdown it returns `false` without
/// re-arming. On a clean wake it resets the attempt counter / backoff to base
/// and returns `true` so the caller can `continue` the loop.
///
/// NOTE: this resets the counters ONLY after Exhausted + a deliberate idle
/// wait — never on connect. The issue #358 latch (reset only on the first
/// SESSION_STATE) is therefore untouched.
async fn idle_retry_after_exhausted(
shutdown_rx: &mut watch::Receiver<bool>,
reconnect_attempt: &mut u32,
backoff: &mut u64,
base_backoff: u64,
idle_retry_ms: u64,
) -> bool {
if idle_retry_ms == 0 {
return false;
}
tokio::select! {
_ = tokio::time::sleep(Duration::from_millis(idle_retry_ms)) => {}
_ = shutdown_rx.changed() => {
if *shutdown_rx.borrow() {
return false;
}
}
}
*reconnect_attempt = 0;
*backoff = base_backoff;
true
}
/// Side-channel result from `handle_incoming_binary`: we need to know whether /// Side-channel result from `handle_incoming_binary`: we need to know whether
/// any decoded frame contained a `MESSAGE_TYPE_SRVR_CTRL_SESSION_STATE` event, /// any decoded frame contained a `MESSAGE_TYPE_SRVR_CTRL_SESSION_STATE` event,
/// because that is the signal the reconnect loop uses to reset its backoff /// because that is the signal the reconnect loop uses to reset its backoff
@@ -1061,6 +1178,28 @@ mod tests {
assert!(keepalive_is_stale(now, now - deadline - 1, 2, deadline)); assert!(keepalive_is_stale(now, now - deadline - 1, 2, deadline));
} }
/// gap #7: with `idle_retry_ms == 0` the helper is a no-op — it returns
/// `false` (terminate) and leaves the counters untouched.
#[tokio::test]
async fn idle_retry_disabled_when_interval_zero() {
let (_tx, mut rx) = watch::channel(false);
let mut attempt = 5u32;
let mut backoff = 8u64;
assert!(!idle_retry_after_exhausted(&mut rx, &mut attempt, &mut backoff, 2, 0).await);
assert_eq!(attempt, 5);
}
/// gap #7: a shutdown signalled during the idle wait must cancel the retry
/// and NOT re-arm (returns `false`, counters untouched).
#[tokio::test]
async fn idle_retry_cancelled_by_shutdown_does_not_rearm() {
let (tx, mut rx) = watch::channel(false);
let _ = tx.send(true);
let mut attempt = 5u32;
let mut backoff = 8u64;
assert!(!idle_retry_after_exhausted(&mut rx, &mut attempt, &mut backoff, 2, 60_000).await);
}
#[test] #[test]
fn qcloud_frame_roundtrip() { fn qcloud_frame_roundtrip() {
let payload = vec![0x01, 0x02, 0x03, 0x04]; let payload = vec![0x01, 0x02, 0x03, 0x04];
+37 -17
View File
@@ -183,6 +183,12 @@ impl QconnectServiceState {
let app_for_errors = app_handle.clone(); let app_for_errors = app_handle.clone();
let inner_for_loop = Arc::clone(&self.inner); let inner_for_loop = Arc::clone(&self.inner);
let app_handle_for_status = app_handle.clone(); let app_handle_for_status = app_handle.clone();
// gap #7: when the transport is configured to idle-retry after
// Exhausted, the transport loop does NOT terminate — it idles then
// re-arms and emits fresh ReconnectScheduled / SessionEstablished
// events. The event loop must therefore stay alive and keep the runtime,
// otherwise the re-armed transport has no listener.
let idle_retry_active = config.reconnect_idle_retry_ms > 0;
let event_loop = tauri::async_runtime::spawn(async move { let event_loop = tauri::async_runtime::spawn(async move {
log::info!("[QConnect/EventLoop] Started listening for transport events"); log::info!("[QConnect/EventLoop] Started listening for transport events");
@@ -377,23 +383,30 @@ impl QconnectServiceState {
"last_reason": last_reason, "last_reason": last_reason,
}), }),
); );
// Auto-stop the runtime: the transport loop // Surface the Exhausted lifecycle to the UI in
// already broke (Exhausted), so no more events // both modes. The difference is what we do with
// will fire. Take the runtime out so a fresh // the runtime + event loop afterwards:
// user-initiated `connect()` succeeds, mark the {
// lifecycle as Exhausted with the cloud-side let mut guard = inner_for_loop.lock().await;
// reason, and exit the event loop. The runtime guard.lifecycle_state = QconnectLifecycleState::Exhausted;
// we just dropped owns this very task's guard.last_error = Some(format!(
// JoinHandle — that's fine, dropping a handle "Reconnect attempts exhausted ({attempts}): {last_reason}"
// detaches; we then `break` so the task ends ));
// naturally (issue #358). if !idle_retry_active {
let mut guard = inner_for_loop.lock().await; // Legacy terminate path: the transport
guard.lifecycle_state = QconnectLifecycleState::Exhausted; // loop already broke, so no more events
guard.last_error = Some(format!( // will fire. Take the runtime out so a
"Reconnect attempts exhausted ({attempts}): {last_reason}" // fresh user-initiated `connect()`
)); // succeeds. Dropping the runtime detaches
guard.runtime = None; // this task's own JoinHandle (fine — we
drop(guard); // break right after) (issue #358).
guard.runtime = None;
}
// gap #7 (idle_retry_active): leave the
// runtime in place. The transport is idling
// and will re-arm; this loop must keep
// consuming its subsequent events.
}
let _ = app_handle_for_status.emit( let _ = app_handle_for_status.emit(
"qconnect:status_changed", "qconnect:status_changed",
json!({ json!({
@@ -403,6 +416,13 @@ impl QconnectServiceState {
"last_reason": last_reason, "last_reason": last_reason,
}), }),
); );
if idle_retry_active {
// Keep the event loop alive for the re-armed
// transport. Skip handle_transport_event for
// this terminal-diagnostic event and wait for
// the next one.
continue;
}
break; break;
} }
} }
+63 -21
View File
@@ -98,6 +98,14 @@ pub fn save_last_known_state(state: bool) {
/// activation are complete. /// activation are complete.
pub struct QconnectCliOverride(pub Option<bool>); pub struct QconnectCliOverride(pub Option<bool>);
/// Bounded, monotonically-increasing backoff schedule for the startup
/// auto-connect retry loop (gap #8). A failed initial connect previously left
/// the lifecycle Off forever; now we retry on this fixed schedule and give up
/// for the session after the last step. Pure so it can be unit-tested headless.
pub fn startup_retry_schedule() -> [u64; 4] {
[2_000, 5_000, 15_000, 30_000]
}
/// Trigger QConnect auto-connect AFTER the runtime is fully bootstrapped /// Trigger QConnect auto-connect AFTER the runtime is fully bootstrapped
/// (client init + OAuth restore + CoreBridge auth + session activation). /// (client init + OAuth restore + CoreBridge auth + session activation).
/// ///
@@ -137,28 +145,62 @@ pub async fn maybe_auto_connect_after_bootstrap(
let app_state = app_handle.state::<crate::AppState>(); let app_state = app_handle.state::<crate::AppState>();
let core_bridge = app_handle.state::<crate::core_bridge::CoreBridgeState>(); let core_bridge = app_handle.state::<crate::core_bridge::CoreBridgeState>();
// Use default options; auto-discovery via qws/createToken resolves // gap #8: retry the initial connect on a bounded backoff schedule
// endpoint+JWT (see transport.rs::resolve_transport_config). // instead of giving up after a single failed attempt. Each iteration
let config = match crate::qconnect::transport::resolve_transport_config( // re-resolves the transport config (endpoint+JWT via qws/createToken)
Default::default(), // because the failure may be a transient credential/network issue that
&app_state, // clears on a later attempt.
) let schedule = startup_retry_schedule();
.await // One immediate attempt, then one per scheduled delay.
{ for attempt in 0..=schedule.len() {
Ok(c) => c, // Use default options; auto-discovery via qws/createToken resolves
Err(e) => { // endpoint+JWT (see transport.rs::resolve_transport_config).
log::warn!("[QConnect] startup auto-connect transport resolve failed: {e}"); let resolved = crate::qconnect::transport::resolve_transport_config(
return; Default::default(),
} &app_state,
}; )
.await;
if let Err(e) = service match resolved {
.connect(app_handle.clone(), core_bridge.0.clone(), config) Ok(config) => {
.await match service
{ .connect(app_handle.clone(), core_bridge.0.clone(), config)
log::warn!("[QConnect] startup auto-connect failed: {e}"); .await
} else { {
log::info!("[QConnect] startup auto-connect succeeded"); Ok(_) => {
log::info!("[QConnect] startup auto-connect succeeded");
return;
}
Err(e) => {
log::warn!(
"[QConnect] startup auto-connect attempt {} failed: {e}",
attempt + 1
);
}
}
}
Err(e) => {
log::warn!(
"[QConnect] startup auto-connect attempt {} transport resolve failed: {e}",
attempt + 1
);
}
}
// If there is a remaining scheduled delay, wait it out before the
// next attempt; otherwise we have exhausted the schedule.
match schedule.get(attempt) {
Some(delay_ms) => {
tokio::time::sleep(std::time::Duration::from_millis(*delay_ms)).await;
}
None => {
log::warn!(
"[QConnect] startup auto-connect gave up for this session after {} attempts",
attempt + 1
);
return;
}
}
} }
}); });
} }
+8
View File
@@ -1143,3 +1143,11 @@ fn build_set_position_request_omits_queue_item_when_unknown() {
assert!(req.current_queue_item.is_none()); assert!(req.current_queue_item.is_none());
assert_eq!(req.current_position, Some(1_000)); assert_eq!(req.current_position, Some(1_000));
} }
#[test]
fn startup_retry_schedule_is_bounded_and_monotonic() {
let s = super::startup::startup_retry_schedule();
assert_eq!(s.len(), 4);
assert!(s.windows(2).all(|w| w[0] < w[1]));
assert_eq!(*s.last().unwrap(), 30_000);
}
+3
View File
@@ -284,6 +284,9 @@ pub(crate) async fn resolve_transport_config(
// gap #12: the real Qobuz endpoint must AUTHENTICATE; a missing JWT here is // gap #12: the real Qobuz endpoint must AUTHENTICATE; a missing JWT here is
// a hard credential error, not a silent skip. // a hard credential error, not a silent skip.
config.require_jwt = true; config.require_jwt = true;
// gap #7: instead of terminating after the bounded reconnect attempts are
// exhausted, idle 60s then rearm — the user no longer has to re-toggle.
config.reconnect_idle_retry_ms = 60_000;
config.reconnect_backoff_ms = options config.reconnect_backoff_ms = options
.reconnect_backoff_ms .reconnect_backoff_ms
.unwrap_or(config.reconnect_backoff_ms); .unwrap_or(config.reconnect_backoff_ms);