feat(core): A.6.1 — use OS connectivity signals to drive reconnect
Extends the A.6 supervisor with an OS-level connectivity hint so a returning network triggers a redial immediately instead of waiting out the current backoff slot (up to 60 s). The watchdog remains the authoritative loss detector — the OS signal is advisory. * `chanora_core::NetworkState` (Unknown / Online / Offline) is owned by `ChanoraSession` via a `tokio::sync::watch::Sender`. `set_network_state()` / `network_state()` are the public accessors. * The supervisor's watch-phase `select!` gains a `network_rx` branch: Offline pre-charges watchdog misses (capped at `MAX_MISSES - 1`) so the next probe failure trips immediately; Online clears stale misses. This shrinks UI-banner latency on a Wi-Fi drop from ~15 s to ~5 s. * The reconnect-loop's backoff sleep races against Online: a transition cuts the sleep short and resets the attempt counter so future losses start at the smallest backoff window again. * `chanora_bridge` adds `BridgeNetworkState` (mirror enum) and a sync `set_network_state(state)` function. On platforms with no signal wired the supervisor stays at Unknown and falls back to pure watchdog/backoff — no behavioural regression vs A.6. * Flutter adds `connectivity_plus ^6.1.0` and wires `_wireConnectivity()` in `main()`: seeds with `checkConnectivity()` then forwards every `onConnectivityChanged` to the bridge, mapping any non-`none` transport to Online. Verified on Moto G Stylus 5G (Android 14): `svc wifi disable && svc data disable` for ~40 s — reconnect banner appeared promptly because the watchdog was pre-charged. After `svc wifi enable && svc data enable` the supervisor woke from its 15 s backoff slot and reconnected within seconds; the channel tree re-rendered without user action.
This commit is contained in:
EdisonJwa
+107
-16
@@ -41,7 +41,7 @@ use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use thiserror::Error;
|
||||
use tokio::sync::{broadcast, oneshot, Mutex};
|
||||
use tokio::sync::{broadcast, oneshot, watch, Mutex};
|
||||
use tokio::task::JoinHandle;
|
||||
use tracing::{info, warn};
|
||||
|
||||
@@ -121,6 +121,20 @@ pub enum SessionEvent {
|
||||
AudioStopped,
|
||||
}
|
||||
|
||||
/// Coarse OS-reported network state. Populated by the Flutter side
|
||||
/// via `connectivity_plus`; on platforms where no signal is wired
|
||||
/// we stay at `Unknown` forever and the supervisor falls back to
|
||||
/// pure watchdog/backoff behaviour.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum NetworkState {
|
||||
/// No signal seen yet — treat as ambiguous; don't change behaviour.
|
||||
Unknown,
|
||||
/// OS reports at least one network with internet capability.
|
||||
Online,
|
||||
/// OS reports no networks available.
|
||||
Offline,
|
||||
}
|
||||
|
||||
/// Channel capacity for the broadcast events. Generous because
|
||||
/// reconnect cycles emit several events per attempt; if subscribers
|
||||
/// fall behind we'd rather skip than block the supervisor.
|
||||
@@ -161,18 +175,39 @@ struct ConnectedState {
|
||||
pub struct ChanoraSession {
|
||||
inner: Arc<Mutex<Option<ConnectedState>>>,
|
||||
events_tx: broadcast::Sender<SessionEvent>,
|
||||
/// OS-reported network state. Updated by the bridge from
|
||||
/// `connectivity_plus` callbacks. Supervisor observes via
|
||||
/// [`watch::Receiver`].
|
||||
network_tx: watch::Sender<NetworkState>,
|
||||
}
|
||||
|
||||
impl ChanoraSession {
|
||||
/// Construct an empty session. Performs no I/O.
|
||||
pub fn new() -> Self {
|
||||
let (events_tx, _) = broadcast::channel(EVENT_CHANNEL_CAPACITY);
|
||||
let (network_tx, _) = watch::channel(NetworkState::Unknown);
|
||||
Self {
|
||||
inner: Arc::new(Mutex::new(None)),
|
||||
events_tx,
|
||||
network_tx,
|
||||
}
|
||||
}
|
||||
|
||||
/// Push an OS connectivity update. Called by the bridge when
|
||||
/// `connectivity_plus` fires. Safe to call from any thread.
|
||||
pub fn set_network_state(&self, state: NetworkState) {
|
||||
// `send_if_modified` would suppress duplicate sends, but
|
||||
// `watch::Sender::send` already drops sends with no
|
||||
// subscribers gracefully. Using `send_replace` so the value
|
||||
// is updated even before any subscriber attaches.
|
||||
let _ = self.network_tx.send_replace(state);
|
||||
}
|
||||
|
||||
/// Read the latest OS connectivity state. Useful for diagnostics.
|
||||
pub fn network_state(&self) -> NetworkState {
|
||||
*self.network_tx.borrow()
|
||||
}
|
||||
|
||||
/// Subscribe to lifecycle events. The returned receiver fires
|
||||
/// on connect / lost / reconnecting / disconnected /
|
||||
/// audio-started / audio-stopped transitions. Multiple
|
||||
@@ -213,6 +248,7 @@ impl ChanoraSession {
|
||||
probe,
|
||||
cancel_rx,
|
||||
sup_inner.clone(),
|
||||
self.network_tx.subscribe(),
|
||||
));
|
||||
|
||||
let _ = self.events_tx.send(SessionEvent::Connected {
|
||||
@@ -355,6 +391,7 @@ async fn supervisor_loop(
|
||||
initial_probe: chanora_protocol::SnapshotProbe,
|
||||
mut cancel_rx: oneshot::Receiver<()>,
|
||||
sup_inner: Arc<Mutex<SupervisorInner>>,
|
||||
mut network_rx: watch::Receiver<NetworkState>,
|
||||
) {
|
||||
let mut lost_rx = initial_lost_rx;
|
||||
let mut probe = initial_probe;
|
||||
@@ -390,6 +427,39 @@ async fn supervisor_loop(
|
||||
}
|
||||
}
|
||||
}
|
||||
changed = network_rx.changed() => {
|
||||
if changed.is_err() {
|
||||
// The session was dropped; nothing more to do.
|
||||
return;
|
||||
}
|
||||
let new_state = *network_rx.borrow_and_update();
|
||||
if new_state == NetworkState::Offline {
|
||||
// OS says we're offline. Don't fabricate a
|
||||
// loss outright — captive portals and
|
||||
// transient flicker can produce false
|
||||
// negatives — but pre-charge the watchdog
|
||||
// so the next probe failure trips it
|
||||
// immediately. This shrinks UI latency from
|
||||
// ~15s to ~5s in the common case.
|
||||
misses = misses.saturating_add(1).min(WATCHDOG_MAX_MISSES - 1);
|
||||
info!(
|
||||
target: "chanora_core",
|
||||
misses,
|
||||
"connectivity: OS reports offline; pre-charging watchdog"
|
||||
);
|
||||
} else if new_state == NetworkState::Online {
|
||||
// Reset on confirmed online so we don't
|
||||
// carry stale Offline charges into the
|
||||
// healthy state.
|
||||
if misses > 0 {
|
||||
info!(
|
||||
target: "chanora_core",
|
||||
"connectivity: OS reports online; clearing watchdog misses"
|
||||
);
|
||||
}
|
||||
misses = 0;
|
||||
}
|
||||
}
|
||||
_ = watchdog.tick() => {
|
||||
match tokio::time::timeout(WATCHDOG_PROBE_TIMEOUT, probe.probe()).await {
|
||||
Ok(Ok(_)) => {
|
||||
@@ -426,14 +496,6 @@ async fn supervisor_loop(
|
||||
misses,
|
||||
"watchdog: declaring connection lost"
|
||||
);
|
||||
// Replace the dead protocol client with a
|
||||
// dropped slot so the reconnect path below
|
||||
// doesn't accidentally keep using it. We
|
||||
// also stop audio here (the reconnect path
|
||||
// does this too, but doing it now ensures
|
||||
// the mic / playback engine stops talking
|
||||
// to a stale voice_out_tx as quickly as
|
||||
// possible).
|
||||
break chanora_protocol::DisconnectReason::Error(
|
||||
"watchdog: server stopped responding".to_string(),
|
||||
);
|
||||
@@ -490,17 +552,46 @@ async fn supervisor_loop(
|
||||
"reconnect: sleeping before next attempt"
|
||||
);
|
||||
|
||||
// Sleep with cancellation support.
|
||||
let slept = tokio::select! {
|
||||
// Sleep with cancellation support. An OS
|
||||
// "online" notification short-circuits the
|
||||
// sleep and resets the attempt counter so the
|
||||
// next outage starts with the smallest backoff
|
||||
// window again.
|
||||
enum SleepOutcome { Elapsed, NetworkUp, Cancelled }
|
||||
let outcome = tokio::select! {
|
||||
biased;
|
||||
_ = &mut cancel_rx => {
|
||||
_ = &mut cancel_rx => SleepOutcome::Cancelled,
|
||||
changed = network_rx.changed() => {
|
||||
if changed.is_err() {
|
||||
SleepOutcome::Cancelled
|
||||
} else if *network_rx.borrow_and_update() == NetworkState::Online {
|
||||
SleepOutcome::NetworkUp
|
||||
} else {
|
||||
// Offline / Unknown transition — keep waiting the rest of the slot.
|
||||
tokio::select! {
|
||||
_ = &mut cancel_rx => SleepOutcome::Cancelled,
|
||||
_ = tokio::time::sleep(Duration::from_secs(delay_secs as u64)) => SleepOutcome::Elapsed,
|
||||
}
|
||||
}
|
||||
}
|
||||
_ = tokio::time::sleep(Duration::from_secs(delay_secs as u64)) => SleepOutcome::Elapsed,
|
||||
};
|
||||
match outcome {
|
||||
SleepOutcome::Cancelled => {
|
||||
info!(target: "chanora_core", "supervisor cancelled during backoff");
|
||||
return;
|
||||
}
|
||||
_ = tokio::time::sleep(Duration::from_secs(delay_secs as u64)) => true,
|
||||
};
|
||||
if !slept {
|
||||
return;
|
||||
SleepOutcome::NetworkUp => {
|
||||
info!(
|
||||
target: "chanora_core",
|
||||
attempt,
|
||||
"connectivity: OS reports online; redialling immediately"
|
||||
);
|
||||
// Reset attempt counter so future losses
|
||||
// start the backoff schedule fresh.
|
||||
attempt = 0;
|
||||
}
|
||||
SleepOutcome::Elapsed => {}
|
||||
}
|
||||
|
||||
info!(target: "chanora_core", attempt, "reconnect: dialling");
|
||||
|
||||
Reference in New Issue
Block a user