mirror of https://github.com/ospab/ostp.git
Compare commits
No commits in common. "7473278cc24004a6421876f9879e4126146d733b" and "e7a4f2b4a4251eac5403238e7b825f87a3b19178" have entirely different histories.
7473278cc2
...
e7a4f2b4a4
|
|
@ -16,13 +16,6 @@ use crate::app::{BridgeCommand, ConnectionStatus, UiEvent};
|
|||
use crate::config::ClientConfig;
|
||||
use crate::tunnel::{ProxyEvent, ProxyToClientMsg};
|
||||
|
||||
/// Per-address ceiling on the UoT/TCP connect attempt. Long enough that a
|
||||
/// genuinely slow mobile path still completes its handshake, short enough that
|
||||
/// a blackholed address (typically IPv6 advertised without a working route)
|
||||
/// costs seconds instead of the kernel's full SYN-retry budget before the next
|
||||
/// candidate address is tried.
|
||||
const UOT_CONNECT_TIMEOUT: Duration = Duration::from_secs(4);
|
||||
|
||||
static SOCKET_PROTECTOR: std::sync::OnceLock<Box<dyn Fn(i32) -> bool + Send + Sync>> = std::sync::OnceLock::new();
|
||||
|
||||
pub fn set_socket_protector<F>(f: F)
|
||||
|
|
@ -1091,27 +1084,7 @@ impl Bridge {
|
|||
) -> Result<crate::transport::Transport> {
|
||||
let mode = self.transport_mode.to_lowercase();
|
||||
if mode == "uot" || mode == "tcp" {
|
||||
// Bound the TCP connect. Without this it inherits the kernel's SYN
|
||||
// retry budget, which is tens of seconds (and can reach ~2 minutes).
|
||||
// That is exactly what made UoT appear to hang on mobile: callers
|
||||
// resolve every address for the server and try IPv6 first (see the
|
||||
// sort in perform_handshake_with_id), and a mobile network that
|
||||
// advertises IPv6 without a working route blackholes the SYN rather
|
||||
// than rejecting it — so the client sat through the full retry
|
||||
// budget before it ever reached the IPv4 address that would have
|
||||
// connected immediately. UDP never showed this because connect() on
|
||||
// a UDP socket only sets the default peer and returns at once.
|
||||
let stream = tokio::time::timeout(
|
||||
UOT_CONNECT_TIMEOUT,
|
||||
tokio::net::TcpStream::connect((target_ip, port)),
|
||||
)
|
||||
.await
|
||||
.map_err(|_| {
|
||||
anyhow::anyhow!(
|
||||
"TCP connect to {target_ip}:{port} timed out after {:?}",
|
||||
UOT_CONNECT_TIMEOUT
|
||||
)
|
||||
})??;
|
||||
let stream = tokio::net::TcpStream::connect((target_ip, port)).await?;
|
||||
let _ = stream.set_nodelay(true);
|
||||
let (mut read_half, mut write_half) = stream.into_split();
|
||||
|
||||
|
|
|
|||
|
|
@ -102,17 +102,6 @@ pub struct ProtocolMachine {
|
|||
_mtu: usize,
|
||||
}
|
||||
|
||||
// ── Gap recovery (see `ProtocolMachine::recover_stalled_gap`) ────────────────
|
||||
// How long the receive sequence may sit stuck behind a missing frame, with
|
||||
// later frames already buffered, before that frame is declared unrecoverable
|
||||
// and skipped. Derived from the live RTO so it scales with the path instead of
|
||||
// guessing, then clamped: the floor keeps a fast link from discarding a frame
|
||||
// that is merely late, the ceiling bounds how long a stall can be visible to
|
||||
// the user before the tunnel unblocks itself.
|
||||
const GAP_RECOVERY_RTO_MULTIPLIER: u32 = 8;
|
||||
const GAP_RECOVERY_MIN: Duration = Duration::from_secs(2);
|
||||
const GAP_RECOVERY_MAX: Duration = Duration::from_secs(10);
|
||||
|
||||
#[derive(Debug, Clone)]
|
||||
struct SentFrame {
|
||||
nonce: u64,
|
||||
|
|
@ -307,107 +296,7 @@ impl ProtocolMachine {
|
|||
Ok(ProtocolAction::HandshakePayload(Bytes::from(extracted_payload), response))
|
||||
}
|
||||
|
||||
/// Restores liveness when the receive sequence is stuck behind a frame that
|
||||
/// can never arrive.
|
||||
///
|
||||
/// Delivery is gated on `expected_recv_nonce`, so a single missing frame
|
||||
/// holds back every later frame. That is correct *while the sender can still
|
||||
/// retransmit* — but the sender drops a frame from `sent_history` once it
|
||||
/// exceeds `max_retries + 2` attempts (see the zombie eviction in
|
||||
/// `handle_tick`). After that the frame is gone for good and the two sides
|
||||
/// deadlock: the receiver buffers forever and NACKs a nonce nobody can
|
||||
/// resend.
|
||||
///
|
||||
/// That deadlock is invisible to the keepalive watchdog, which is why it
|
||||
/// presented as a hard freeze rather than a reconnect: retransmits, ACKs and
|
||||
/// NACKs keep flowing, so the client's `last_valid_recv` keeps refreshing and
|
||||
/// its stall detector never fires. The RTT readout freezes at its last value
|
||||
/// for the same reason — Pong rides in a Data frame stuck behind the gap.
|
||||
///
|
||||
/// So: once we have been stuck long enough that retransmission has provably
|
||||
/// given up, skip to the lowest buffered nonce and drain. This drops the
|
||||
/// missing frame's payload (one RelayMessage — a chunk of one stream), which
|
||||
/// is a real cost, but the alternative is a permanently dead tunnel.
|
||||
fn recover_stalled_gap(&mut self) -> Vec<ProtocolAction> {
|
||||
let mut recovered = Vec::new();
|
||||
if self.reorder_buffer.is_empty() {
|
||||
return recovered;
|
||||
}
|
||||
|
||||
// Wait out the sender's full retransmit budget before giving up, so a
|
||||
// frame that is merely late is never discarded. The sender backs off
|
||||
// exponentially, so key this off the live RTO estimate rather than a
|
||||
// flat constant, with a floor that keeps low-RTT links from skipping
|
||||
// too eagerly and a ceiling that bounds the visible freeze.
|
||||
let timeout = self
|
||||
.cc
|
||||
.rto()
|
||||
.saturating_mul(GAP_RECOVERY_RTO_MULTIPLIER)
|
||||
.clamp(GAP_RECOVERY_MIN, GAP_RECOVERY_MAX);
|
||||
if self.last_recv_advance.elapsed() < timeout {
|
||||
return recovered;
|
||||
}
|
||||
|
||||
let Some(&resume_at) = self.reorder_buffer.keys().next() else {
|
||||
return recovered;
|
||||
};
|
||||
let skipped = resume_at.saturating_sub(self.expected_recv_nonce);
|
||||
tracing::warn!(
|
||||
"Gap recovery: no progress for {:?}; skipping {} unrecoverable frame(s) \
|
||||
(nonce {} -> {}) to unblock the session",
|
||||
self.last_recv_advance.elapsed(),
|
||||
skipped,
|
||||
self.expected_recv_nonce,
|
||||
resume_at
|
||||
);
|
||||
|
||||
self.expected_recv_nonce = resume_at;
|
||||
while let Some(buffered) = self.reorder_buffer.remove(&self.expected_recv_nonce) {
|
||||
recovered.push(buffered);
|
||||
match self.expected_recv_nonce.checked_add(1) {
|
||||
Some(next) => self.expected_recv_nonce = next,
|
||||
// u64 nonce space exhausted: stop draining rather than wrap.
|
||||
// The session is finished either way; the caller's next decrypt
|
||||
// will fail and tear it down.
|
||||
None => break,
|
||||
}
|
||||
}
|
||||
self.last_recv_advance = Instant::now();
|
||||
// The peer must learn the sequence moved on, or it will keep
|
||||
// retransmitting into the void.
|
||||
self.ack_pending = true;
|
||||
|
||||
recovered
|
||||
}
|
||||
|
||||
fn handle_data_inbound(&mut self, raw_vec: &[u8]) -> Result<ProtocolAction, ProtocolError> {
|
||||
// Check for a stalled gap before classifying this frame, so the rest of
|
||||
// the function sees an already-advanced `expected_recv_nonce`. Runs here
|
||||
// rather than on Tick because both tick handlers discard DeliverApp
|
||||
// actions, and because inbound frames keep arriving throughout the stall
|
||||
// (retransmits/ACKs/NACKs/keepalives) — so this path is reliably reached.
|
||||
let recovered = self.recover_stalled_gap();
|
||||
let result = self.handle_data_inbound_frame(raw_vec)?;
|
||||
if recovered.is_empty() {
|
||||
return Ok(result);
|
||||
}
|
||||
|
||||
// Recovered payloads are older than anything this frame produces, so
|
||||
// they go first to preserve delivery order.
|
||||
let mut all = recovered;
|
||||
match result {
|
||||
ProtocolAction::Noop => {}
|
||||
ProtocolAction::Multiple(list) => all.extend(list),
|
||||
single => all.push(single),
|
||||
}
|
||||
Ok(if all.len() == 1 {
|
||||
all.pop().unwrap()
|
||||
} else {
|
||||
ProtocolAction::Multiple(all)
|
||||
})
|
||||
}
|
||||
|
||||
fn handle_data_inbound_frame(&mut self, raw_vec: &[u8]) -> Result<ProtocolAction, ProtocolError> {
|
||||
if raw_vec.len() < 12 {
|
||||
return Err(ProtocolError::Framing("data datagram too short".to_string()));
|
||||
}
|
||||
|
|
@ -1082,96 +971,4 @@ mod tests {
|
|||
let _ = client.on_event(OstpEvent::Tick).unwrap();
|
||||
let _ = server.on_event(OstpEvent::Tick).unwrap();
|
||||
}
|
||||
|
||||
/// Count how many application payloads an action tree actually delivers.
|
||||
fn delivered_payloads(action: &ProtocolAction) -> Vec<Bytes> {
|
||||
match action {
|
||||
ProtocolAction::DeliverApp(_, data) => vec![data.clone()],
|
||||
ProtocolAction::Multiple(list) => list.iter().flat_map(delivered_payloads).collect(),
|
||||
_ => Vec::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Build `count` data frames on `client`, returning them without delivering
|
||||
/// any — lets a test choose which ones to "lose" in transit.
|
||||
fn make_data_frames(client: &mut ProtocolMachine, count: u8) -> Vec<Bytes> {
|
||||
(0..count)
|
||||
.map(|i| {
|
||||
let payload = Bytes::from(vec![i; 32]);
|
||||
match client.on_event(OstpEvent::Outbound(1, payload)).unwrap() {
|
||||
ProtocolAction::SendDatagram(d) => d,
|
||||
_ => panic!("expected SendDatagram for frame {i}"),
|
||||
}
|
||||
})
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The freeze this fixes: a frame is lost, the sender eventually stops
|
||||
/// retransmitting it, and the receiver — which gates delivery on
|
||||
/// `expected_recv_nonce` — waits for it forever. Every later frame piles up
|
||||
/// undelivered while the transport itself stays healthy, so nothing upstream
|
||||
/// notices. Recovery must eventually skip the hole and release the backlog.
|
||||
#[test]
|
||||
fn test_gap_recovery_releases_permanently_stalled_frames() {
|
||||
let (mut client, mut server) = do_handshake();
|
||||
let frames = make_data_frames(&mut client, 4);
|
||||
|
||||
// Frame 0 arrives in order and is delivered straight through.
|
||||
let action = server.on_event(OstpEvent::Inbound(frames[0].clone())).unwrap();
|
||||
assert_eq!(delivered_payloads(&action).len(), 1, "in-order frame should deliver");
|
||||
|
||||
// Frame 1 is lost. 2 and 3 arrive but must be held back — delivering them
|
||||
// now would reorder the stream.
|
||||
for idx in [2usize, 3] {
|
||||
let action = server.on_event(OstpEvent::Inbound(frames[idx].clone())).unwrap();
|
||||
assert!(
|
||||
delivered_payloads(&action).is_empty(),
|
||||
"frame {idx} must stay buffered behind the missing frame"
|
||||
);
|
||||
}
|
||||
|
||||
// Stand in for "the sender exhausted its retries and dropped frame 1":
|
||||
// the sequence has not advanced for longer than the recovery timeout.
|
||||
server.last_recv_advance = Instant::now() - GAP_RECOVERY_MAX - Duration::from_secs(1);
|
||||
|
||||
// The next inbound frame (a retransmitted duplicate, which is exactly what
|
||||
// a real stalled session keeps receiving) must unblock the backlog.
|
||||
let action = server.on_event(OstpEvent::Inbound(frames[0].clone())).unwrap();
|
||||
let delivered = delivered_payloads(&action);
|
||||
assert_eq!(
|
||||
delivered.len(),
|
||||
2,
|
||||
"both buffered frames must be released once the gap is declared unrecoverable"
|
||||
);
|
||||
// ...and in order: frame 2 before frame 3.
|
||||
assert_eq!(delivered[0][0], 2);
|
||||
assert_eq!(delivered[1][0], 3);
|
||||
}
|
||||
|
||||
/// Recovery must not be trigger-happy: a frame that is merely late still has
|
||||
/// to be waited for, or we would discard data the sender is about to resend.
|
||||
#[test]
|
||||
fn test_gap_recovery_does_not_fire_before_timeout() {
|
||||
let (mut client, mut server) = do_handshake();
|
||||
let frames = make_data_frames(&mut client, 3);
|
||||
|
||||
server.on_event(OstpEvent::Inbound(frames[0].clone())).unwrap();
|
||||
let action = server.on_event(OstpEvent::Inbound(frames[2].clone())).unwrap();
|
||||
assert!(delivered_payloads(&action).is_empty());
|
||||
|
||||
// Well inside the timeout — the gap must still be respected.
|
||||
let action = server.on_event(OstpEvent::Inbound(frames[0].clone())).unwrap();
|
||||
assert!(
|
||||
delivered_payloads(&action).is_empty(),
|
||||
"must keep waiting while retransmission is still plausible"
|
||||
);
|
||||
|
||||
// And once the genuinely-late frame shows up, normal in-order delivery
|
||||
// resumes with nothing dropped.
|
||||
let action = server.on_event(OstpEvent::Inbound(frames[1].clone())).unwrap();
|
||||
let delivered = delivered_payloads(&action);
|
||||
assert_eq!(delivered.len(), 2, "late frame plus the buffered one");
|
||||
assert_eq!(delivered[0][0], 1);
|
||||
assert_eq!(delivered[1][0], 2);
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,13 +8,6 @@ import '../models/connection_state_enum.dart';
|
|||
import '../models/ostp_profile.dart';
|
||||
import 'settings_screen.dart';
|
||||
|
||||
/// Success green for the "connected" state — the button aura/border/icon and
|
||||
/// the top-bar status dot. The theme's `secondary` (#AAAAAA) reads as plain
|
||||
/// white here, which gave no visual confirmation that the tunnel actually came
|
||||
/// up. Reuses the same green already used for a healthy ping value, so
|
||||
/// "green = good" stays consistent across the UI.
|
||||
const Color kConnectedGreen = Color(0xFF22D3A5);
|
||||
|
||||
class HomeScreen extends StatefulWidget {
|
||||
final SharedPreferences prefs;
|
||||
const HomeScreen({super.key, required this.prefs});
|
||||
|
|
@ -52,7 +45,8 @@ class _HomeScreenState extends State<HomeScreen> with TickerProviderStateMixin {
|
|||
late AnimationController _pulseController;
|
||||
late AnimationController _spinController;
|
||||
|
||||
String _pingText = '-- ms';
|
||||
bool _isCheckingPing = false;
|
||||
String _pingText = 'Target Ping: -- ms';
|
||||
Color _pingColor = Colors.white54;
|
||||
|
||||
@override
|
||||
|
|
@ -426,8 +420,8 @@ class _HomeScreenState extends State<HomeScreen> with TickerProviderStateMixin {
|
|||
_prevBytesSent = bytesSent;
|
||||
_downSpeed = '${_formatBytes(dRecv)}/s';
|
||||
_upSpeed = '${_formatBytes(dSent)}/s';
|
||||
if (rttMs > 0) {
|
||||
_pingText = '$rttMs ms';
|
||||
if (rttMs > 0 && !_isCheckingPing) {
|
||||
_pingText = 'Server Ping: $rttMs ms';
|
||||
if (rttMs < 100) {
|
||||
_pingColor = const Color(0xFF22D3A5);
|
||||
} else if (rttMs < 250) {
|
||||
|
|
@ -453,6 +447,47 @@ class _HomeScreenState extends State<HomeScreen> with TickerProviderStateMixin {
|
|||
return '${(bytes / (1024 * 1024 * 1024)).toStringAsFixed(1)} GB';
|
||||
}
|
||||
|
||||
Future<void> _checkConnectionLatency() async {
|
||||
if (_state != ConnectionStateEnum.connected) return;
|
||||
|
||||
setState(() {
|
||||
_isCheckingPing = true;
|
||||
_pingText = 'Updating...';
|
||||
_pingColor = Colors.white70;
|
||||
});
|
||||
|
||||
try {
|
||||
final metricsJson = await platform.invokeMethod('getMetrics');
|
||||
if (metricsJson != null && metricsJson.isNotEmpty) {
|
||||
final Map<String, dynamic> parsed = jsonDecode(metricsJson);
|
||||
final rttMs = parsed['rtt_ms'] as int? ?? 0;
|
||||
if (mounted) {
|
||||
setState(() {
|
||||
if (rttMs > 0) {
|
||||
_pingText = 'Server Ping: $rttMs ms';
|
||||
_pingColor = rttMs < 100
|
||||
? const Color(0xFF22D3A5)
|
||||
: rttMs < 250
|
||||
? Colors.amberAccent
|
||||
: Colors.redAccent;
|
||||
} else {
|
||||
_pingText = 'Server Ping: -- ms';
|
||||
_pingColor = Colors.white54;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
} catch (e) {
|
||||
debugPrint("Failed to check latency: $e");
|
||||
}
|
||||
|
||||
if (mounted) {
|
||||
setState(() {
|
||||
_isCheckingPing = false;
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
void _setDisconnected() {
|
||||
if (!mounted) return;
|
||||
setState(() {
|
||||
|
|
@ -463,8 +498,9 @@ class _HomeScreenState extends State<HomeScreen> with TickerProviderStateMixin {
|
|||
_upSpeed = '0 B/s';
|
||||
_prevBytesRecv = 0;
|
||||
_prevBytesSent = 0;
|
||||
_pingText = '-- ms';
|
||||
_pingText = 'Target Ping: -- ms';
|
||||
_pingColor = Colors.white54;
|
||||
_isCheckingPing = false;
|
||||
});
|
||||
_pulseController.stop();
|
||||
_pulseController.value = 0.0;
|
||||
|
|
@ -542,12 +578,12 @@ class _HomeScreenState extends State<HomeScreen> with TickerProviderStateMixin {
|
|||
decoration: BoxDecoration(
|
||||
borderRadius: BorderRadius.circular(4),
|
||||
color: _state == ConnectionStateEnum.connected
|
||||
? kConnectedGreen
|
||||
? theme.colorScheme.secondary
|
||||
: theme.colorScheme.primary,
|
||||
boxShadow: [
|
||||
BoxShadow(
|
||||
color: _state == ConnectionStateEnum.connected
|
||||
? kConnectedGreen.withOpacity(0.5)
|
||||
? theme.colorScheme.secondary.withOpacity(0.5)
|
||||
: theme.colorScheme.primary.withOpacity(0.5),
|
||||
blurRadius: 10,
|
||||
)
|
||||
|
|
@ -601,7 +637,7 @@ class _HomeScreenState extends State<HomeScreen> with TickerProviderStateMixin {
|
|||
|
||||
Widget _buildStage(ThemeData theme) {
|
||||
Color getAccentColor() {
|
||||
if (_state == ConnectionStateEnum.connected) return kConnectedGreen;
|
||||
if (_state == ConnectionStateEnum.connected) return theme.colorScheme.secondary;
|
||||
return theme.colorScheme.primary;
|
||||
}
|
||||
|
||||
|
|
@ -739,23 +775,37 @@ class _HomeScreenState extends State<HomeScreen> with TickerProviderStateMixin {
|
|||
opacity: _state == ConnectionStateEnum.connected ? 1.0 : 0.0,
|
||||
duration: const Duration(milliseconds: 300),
|
||||
child: Padding(
|
||||
padding: const EdgeInsets.only(top: 10),
|
||||
padding: const EdgeInsets.only(top: 16),
|
||||
child: Container(
|
||||
padding: const EdgeInsets.symmetric(horizontal: 12, vertical: 6),
|
||||
margin: const EdgeInsets.symmetric(horizontal: 16),
|
||||
padding: const EdgeInsets.symmetric(horizontal: 16, vertical: 12),
|
||||
decoration: BoxDecoration(
|
||||
color: Colors.white.withOpacity(0.03),
|
||||
borderRadius: BorderRadius.circular(12),
|
||||
borderRadius: BorderRadius.circular(20),
|
||||
border: Border.all(color: Colors.white.withOpacity(0.06)),
|
||||
),
|
||||
child: Row(
|
||||
mainAxisSize: MainAxisSize.min,
|
||||
mainAxisAlignment: MainAxisAlignment.spaceBetween,
|
||||
children: [
|
||||
Icon(Icons.speed_rounded, size: 13, color: _pingColor),
|
||||
const SizedBox(width: 6),
|
||||
Expanded(
|
||||
child: Column(
|
||||
crossAxisAlignment: CrossAxisAlignment.start,
|
||||
children: [
|
||||
const Text(
|
||||
'CONNECTION TEST',
|
||||
style: TextStyle(
|
||||
fontSize: 10,
|
||||
fontWeight: FontWeight.bold,
|
||||
color: Colors.white38,
|
||||
letterSpacing: 0.8,
|
||||
),
|
||||
),
|
||||
const SizedBox(height: 4),
|
||||
Text(
|
||||
_pingText,
|
||||
overflow: TextOverflow.ellipsis,
|
||||
style: TextStyle(
|
||||
fontSize: 13,
|
||||
fontSize: 15,
|
||||
fontWeight: FontWeight.bold,
|
||||
color: _pingColor,
|
||||
),
|
||||
|
|
@ -763,6 +813,32 @@ class _HomeScreenState extends State<HomeScreen> with TickerProviderStateMixin {
|
|||
],
|
||||
),
|
||||
),
|
||||
const SizedBox(width: 8),
|
||||
_isCheckingPing
|
||||
? const SizedBox(
|
||||
width: 20, height: 20,
|
||||
child: CircularProgressIndicator(strokeWidth: 2, color: Colors.white70),
|
||||
)
|
||||
: TextButton.icon(
|
||||
onPressed: _checkConnectionLatency,
|
||||
icon: Icon(Icons.speed_rounded, size: 16, color: theme.colorScheme.primary),
|
||||
label: Text(
|
||||
'Test Ping',
|
||||
style: TextStyle(
|
||||
fontWeight: FontWeight.bold,
|
||||
fontSize: 13,
|
||||
color: theme.colorScheme.primary,
|
||||
),
|
||||
),
|
||||
style: TextButton.styleFrom(
|
||||
padding: const EdgeInsets.symmetric(horizontal: 12, vertical: 8),
|
||||
backgroundColor: theme.colorScheme.primary.withOpacity(0.1),
|
||||
shape: RoundedRectangleBorder(borderRadius: BorderRadius.circular(12)),
|
||||
),
|
||||
),
|
||||
],
|
||||
),
|
||||
),
|
||||
),
|
||||
)
|
||||
],
|
||||
|
|
|
|||
Loading…
Reference in New Issue