fix(offline): hysteresis + cheap probe + streaming liveness override (#467)
Long sessions died when a single transient blip flipped the app to offline and the queue then skipped every cached-but-not-downloaded track. Harden the detector: - Consecutive-failure hysteresis: declare offline only after 2 failed polls, not a single blip. - Replace the HEAD https://1.1.1.1 probe with a concurrent race of cheap generate_204 endpoints, so a saturated link (long Hi-Res tracks) no longer times the probe out. - Positive liveness override: if a segment downloaded within the last 45s we are online by definition, regardless of the probe. cc
This commit is contained in:
3 files changed
+127
-64
No files matched your search
@@ -73,6 +73,10 @@ struct ThrottleInner {
|
||||
/// Timestamp of the most recent audio underrun. `None` means we have
|
||||
/// not observed one this session.
|
||||
last_underrun: Option<Instant>,
|
||||
/// Timestamp of the most recent successful segment download. `None`
|
||||
/// until the first segment lands. Used as a positive liveness signal
|
||||
/// for the offline detector (issue #467).
|
||||
last_successful_download: Option<Instant>,
|
||||
}
|
||||
|
||||
pub struct ThrottleState {
|
||||
@@ -92,16 +96,20 @@ impl ThrottleState {
|
||||
/// Feed a fresh per-segment bandwidth measurement (MB/s). Called from
|
||||
/// the CMAF streaming loop every few segments.
|
||||
pub fn record_segment_bandwidth(&self, mbps: f64) {
|
||||
if !mbps.is_finite() || mbps <= 0.0 {
|
||||
return;
|
||||
}
|
||||
if let Ok(mut inner) = self.inner.write() {
|
||||
// Liveness signal (issue #467): reaching this path means a segment
|
||||
// batch just downloaded — bytes are flowing, so we are online,
|
||||
// independent of whether the throughput sample is usable for the
|
||||
// EMA. Recorded before the validity guard on purpose.
|
||||
inner.last_successful_download = Some(Instant::now());
|
||||
if mbps.is_finite() && mbps > 0.0 {
|
||||
inner.bandwidth_ema_mbps = Some(match inner.bandwidth_ema_mbps {
|
||||
Some(prev) => prev * (1.0 - BANDWIDTH_EMA_ALPHA) + mbps * BANDWIDTH_EMA_ALPHA,
|
||||
None => mbps,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Signal that an audio buffer underrun just happened. Forces the
|
||||
/// throttle into panic mode for `PANIC_WINDOW_SECS`.
|
||||
@@ -116,6 +124,18 @@ impl ThrottleState {
|
||||
self.inner.read().ok().and_then(|i| i.bandwidth_ema_mbps)
|
||||
}
|
||||
|
||||
/// Seconds since the last successful segment download, or `None` if no
|
||||
/// bytes have been pulled this session. Positive liveness signal for the
|
||||
/// offline detector (issue #467): if segments are flowing we are online
|
||||
/// by definition, regardless of what a connectivity probe reports.
|
||||
pub fn seconds_since_download(&self) -> Option<u64> {
|
||||
self.inner
|
||||
.read()
|
||||
.ok()
|
||||
.and_then(|i| i.last_successful_download)
|
||||
.map(|t| t.elapsed().as_secs())
|
||||
}
|
||||
|
||||
/// True when an underrun was recorded within `PANIC_WINDOW_SECS`.
|
||||
pub fn in_panic_mode(&self) -> bool {
|
||||
let inner = match self.inner.read() {
|
||||
|
||||
@@ -1361,10 +1361,35 @@ pub async fn v2_get_offline_status(
|
||||
});
|
||||
}
|
||||
|
||||
// Check network connectivity
|
||||
let has_network = crate::offline::check_network_connectivity().await;
|
||||
// Fix E (issue #467): positive liveness override. If audio segments were
|
||||
// flowing within the liveness window we are, by definition, online —
|
||||
// regardless of what the connectivity probe says. Heavy Hi-Res tracks
|
||||
// (long Dream Theater / Yes) saturate the link and can time the probe
|
||||
// out; active streaming is the strongest possible "online" signal.
|
||||
const LIVENESS_WINDOW_SECS: u64 = 45;
|
||||
if let Some(secs) = qbz_audio::network_throttle::state().seconds_since_download() {
|
||||
if secs <= LIVENESS_WINDOW_SECS {
|
||||
crate::offline::reset_connectivity_failures();
|
||||
return Ok(crate::offline::OfflineStatus {
|
||||
is_offline: false,
|
||||
reason: None,
|
||||
manual_mode_enabled: false,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
if !has_network {
|
||||
// Check network connectivity (cheap concurrent probe) and apply
|
||||
// consecutive-failure hysteresis (Fix A, issue #467): a single failed poll
|
||||
// must not flip offline — only OFFLINE_FAILURE_THRESHOLD consecutive
|
||||
// failures do.
|
||||
let has_network = crate::offline::check_network_connectivity().await;
|
||||
let failures = crate::offline::record_connectivity_result(has_network);
|
||||
|
||||
if !has_network && failures >= crate::offline::OFFLINE_FAILURE_THRESHOLD {
|
||||
log::info!(
|
||||
"[offline] Declaring offline after {} consecutive connectivity failures",
|
||||
failures
|
||||
);
|
||||
return Ok(crate::offline::OfflineStatus {
|
||||
is_offline: true,
|
||||
reason: Some(crate::offline::OfflineReason::NoNetwork),
|
||||
@@ -1372,6 +1397,15 @@ pub async fn v2_get_offline_status(
|
||||
});
|
||||
}
|
||||
|
||||
if !has_network {
|
||||
log::info!(
|
||||
"[offline] Connectivity probe failed ({}/{}); staying online (hysteresis)",
|
||||
failures,
|
||||
crate::offline::OFFLINE_FAILURE_THRESHOLD
|
||||
);
|
||||
}
|
||||
|
||||
// Online, or a transient failure still under the threshold: stay online.
|
||||
Ok(crate::offline::OfflineStatus {
|
||||
is_offline: false,
|
||||
reason: None,
|
||||
|
||||
@@ -633,75 +633,84 @@ impl OfflineState {
|
||||
}
|
||||
}
|
||||
|
||||
/// Counter for alternating between neutral endpoint and Qobuz checks.
|
||||
/// Using atomic for thread safety across async calls.
|
||||
static CHECK_COUNTER: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
|
||||
/// Consecutive failed connectivity polls. The offline flag is only raised
|
||||
/// once this reaches `OFFLINE_FAILURE_THRESHOLD`, so a single transient blip
|
||||
/// (Wi-Fi micro-drop, DNS hiccup, or a saturated link losing the probe race
|
||||
/// during a big Hi-Res download) can no longer flip the app offline. See
|
||||
/// issue #467.
|
||||
static CONSECUTIVE_FAILURES: std::sync::atomic::AtomicU32 = std::sync::atomic::AtomicU32::new(0);
|
||||
|
||||
/// Check network connectivity using a hybrid strategy:
|
||||
/// - 9 out of 10 checks go to a neutral endpoint (1.1.1.1) to verify basic internet
|
||||
/// - 1 out of 10 checks go to Qobuz to verify service availability
|
||||
/// Number of consecutive failed polls (~30 s apart) required before the app
|
||||
/// is declared offline. 2 ≈ 60 s of confirmed loss — lax enough to ride out
|
||||
/// blips, still reacts to a real outage within about a minute.
|
||||
pub const OFFLINE_FAILURE_THRESHOLD: u32 = 2;
|
||||
|
||||
/// Record a connectivity-poll outcome and return the running consecutive
|
||||
/// failure count. Any success resets the counter to 0.
|
||||
pub fn record_connectivity_result(success: bool) -> u32 {
|
||||
use std::sync::atomic::Ordering;
|
||||
if success {
|
||||
CONSECUTIVE_FAILURES.store(0, Ordering::Relaxed);
|
||||
0
|
||||
} else {
|
||||
CONSECUTIVE_FAILURES.fetch_add(1, Ordering::Relaxed) + 1
|
||||
}
|
||||
}
|
||||
|
||||
/// Reset the consecutive-failure counter. Called on a strong positive signal
|
||||
/// (active streaming liveness, manual-mode change) so the next real loss
|
||||
/// starts counting from zero.
|
||||
pub fn reset_connectivity_failures() {
|
||||
CONSECUTIVE_FAILURES.store(0, std::sync::atomic::Ordering::Relaxed);
|
||||
}
|
||||
|
||||
/// Lightweight "generate_204"-style connectivity probes. A success is a
|
||||
/// near-empty 204/200 response, so the probe barely competes for bandwidth
|
||||
/// even while a large Hi-Res download is saturating the link — the exact
|
||||
/// condition (long Dream Theater / Yes tracks) that timed the old
|
||||
/// `HEAD https://1.1.1.1` probe out and produced a false offline (issue #467).
|
||||
const PROBE_ENDPOINTS: &[&str] = &[
|
||||
"https://connectivitycheck.gstatic.com/generate_204",
|
||||
"https://www.gstatic.com/generate_204",
|
||||
"https://cloudflare.com/cdn-cgi/trace",
|
||||
];
|
||||
|
||||
/// Check basic internet connectivity with a cheap, concurrent probe.
|
||||
///
|
||||
/// This reduces load on Qobuz API and avoids false positives from rate limiting
|
||||
/// when the app is making many concurrent API calls.
|
||||
/// Races the lightweight endpoints above and returns `true` as soon as any
|
||||
/// one succeeds. This is intentionally *not* a Qobuz-service check: a slow or
|
||||
/// rate-limited Qobuz must never read as "offline" (issue #467). Actual Qobuz
|
||||
/// API failures are handled with retries/backoff in the fetch path instead.
|
||||
pub async fn check_network_connectivity() -> bool {
|
||||
let counter = CHECK_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
|
||||
let check_qobuz = counter % 10 == 0;
|
||||
|
||||
let client = reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(10))
|
||||
.build();
|
||||
|
||||
let client = match client {
|
||||
let client = match reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(8))
|
||||
.build()
|
||||
{
|
||||
Ok(c) => c,
|
||||
Err(_) => return false,
|
||||
};
|
||||
|
||||
let endpoint = if check_qobuz {
|
||||
"https://www.qobuz.com"
|
||||
} else {
|
||||
// Cloudflare DNS - highly reliable, low latency
|
||||
"https://1.1.1.1"
|
||||
};
|
||||
|
||||
// Try up to 2 times before declaring offline
|
||||
for attempt in 1..=2 {
|
||||
match client.head(endpoint).send().await {
|
||||
let mut set = tokio::task::JoinSet::new();
|
||||
for endpoint in PROBE_ENDPOINTS {
|
||||
let client = client.clone();
|
||||
let endpoint = (*endpoint).to_string();
|
||||
set.spawn(async move {
|
||||
match client.get(&endpoint).send().await {
|
||||
Ok(response) => {
|
||||
if response.status().is_success() || response.status().is_redirection() {
|
||||
response.status().is_success() || response.status().is_redirection()
|
||||
}
|
||||
Err(_) => false,
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
while let Some(joined) = set.join_next().await {
|
||||
if matches!(joined, Ok(true)) {
|
||||
set.abort_all();
|
||||
return true;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
log::warn!(
|
||||
"Network check attempt {} to {} failed: {}",
|
||||
attempt,
|
||||
endpoint,
|
||||
e
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Wait 1 second before retry
|
||||
if attempt < 2 {
|
||||
tokio::time::sleep(Duration::from_secs(1)).await;
|
||||
}
|
||||
}
|
||||
|
||||
// If neutral endpoint failed, try Qobuz as fallback (maybe it's a DNS issue)
|
||||
if !check_qobuz {
|
||||
log::info!("Neutral endpoint failed, trying Qobuz as fallback...");
|
||||
match client.head("https://www.qobuz.com").send().await {
|
||||
Ok(response) => {
|
||||
if response.status().is_success() || response.status().is_redirection() {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
Err(e) => {
|
||||
log::warn!("Qobuz fallback check also failed: {}", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
log::info!("Network connectivity check failed after all attempts");
|
||||
log::info!("[offline] Connectivity probe failed on all endpoints");
|
||||
false
|
||||
}
|
||||
Reference in new issue
Block a user