mirror of https://github.com/ospab/ostp.git
Compare commits
2 Commits
ebfc751471
...
108bab6a90
| Author | SHA1 | Date |
|---|---|---|
|
|
108bab6a90 | |
|
|
f7e9215331 |
|
|
@ -137,6 +137,16 @@ pub struct Bridge {
|
||||||
last_rtt_ms: f64,
|
last_rtt_ms: f64,
|
||||||
last_sample_at: Instant,
|
last_sample_at: Instant,
|
||||||
last_valid_recv: Instant,
|
last_valid_recv: Instant,
|
||||||
|
/// Set when a suspend/resume is detected, cleared once a reconnect actually
|
||||||
|
/// succeeds. Waking is precisely when the network is least likely to be
|
||||||
|
/// ready — Wi-Fi has not reassociated yet — so a single attempt fired
|
||||||
|
/// milliseconds after resume usually fails, and a one-shot forced reconnect
|
||||||
|
/// then fell back to the ordinary 25s stall heuristic. That heuristic keys
|
||||||
|
/// off a monotonic clock which does not advance while the machine is
|
||||||
|
/// asleep, so it could take a further 25s of real uptime to fire, or not
|
||||||
|
/// fire at all. Retrying until success removes the dependency on either.
|
||||||
|
forced_reconnect_pending: bool,
|
||||||
|
last_forced_reconnect_try: Instant,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl Bridge {
|
impl Bridge {
|
||||||
|
|
@ -173,6 +183,8 @@ impl Bridge {
|
||||||
last_rtt_ms: 0.0,
|
last_rtt_ms: 0.0,
|
||||||
last_sample_at: Instant::now(),
|
last_sample_at: Instant::now(),
|
||||||
last_valid_recv: Instant::now(),
|
last_valid_recv: Instant::now(),
|
||||||
|
forced_reconnect_pending: false,
|
||||||
|
last_forced_reconnect_try: Instant::now(),
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -255,7 +267,27 @@ impl Bridge {
|
||||||
let _ = tx.send(UiEvent::Log(format!(
|
let _ = tx.send(UiEvent::Log(format!(
|
||||||
"Resumed after ~{}s suspend — forcing clean reconnect", wall_gap.as_secs()
|
"Resumed after ~{}s suspend — forcing clean reconnect", wall_gap.as_secs()
|
||||||
))).await;
|
))).await;
|
||||||
|
self.forced_reconnect_pending = true;
|
||||||
|
self.last_forced_reconnect_try = Instant::now() - Duration::from_secs(60);
|
||||||
|
}
|
||||||
|
|
||||||
|
// Keep retrying a resume-triggered reconnect until one lands.
|
||||||
|
// The first attempt fires within half a second of waking, when
|
||||||
|
// the NIC is typically still reassociating, so treating it as
|
||||||
|
// one-shot left the tunnel dead until some other timer noticed.
|
||||||
|
if self.running
|
||||||
|
&& self.forced_reconnect_pending
|
||||||
|
&& self.last_forced_reconnect_try.elapsed() >= Duration::from_secs(3)
|
||||||
|
{
|
||||||
|
self.last_forced_reconnect_try = Instant::now();
|
||||||
self.handle_keepalive(true, &mut sessions_opt, &mut udp_rx_opt, &mut proxy_guard, &mut stream_map, &tx, &proxy_tx, &mut proxy_rx).await;
|
self.handle_keepalive(true, &mut sessions_opt, &mut udp_rx_opt, &mut proxy_guard, &mut stream_map, &tx, &proxy_tx, &mut proxy_rx).await;
|
||||||
|
// handle_keepalive refreshes last_valid_recv only when a
|
||||||
|
// session was actually established, so this is a real
|
||||||
|
// success check rather than "we tried".
|
||||||
|
if self.last_valid_recv.elapsed() < Duration::from_secs(3) {
|
||||||
|
self.forced_reconnect_pending = false;
|
||||||
|
let _ = tx.send(UiEvent::Log("Reconnected after suspend".into())).await;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
if self.running {
|
if self.running {
|
||||||
self.emit_metrics(&tx).await;
|
self.emit_metrics(&tx).await;
|
||||||
|
|
@ -272,7 +304,12 @@ impl Bridge {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
proxy_ev = proxy_rx.recv(), if self.running && sessions_opt.as_ref().map(|s| {
|
proxy_ev = proxy_rx.recv(), if self.running && sessions_opt.as_ref().map(|s| {
|
||||||
s.iter().any(|ses| ses.machine.in_flight_count() < ses.machine.cwnd_packets().clamp(16, 16384))
|
// Upper bound matches MAX_CWND_PACKETS in ostp-core's congestion
|
||||||
|
// controller. The old 16384 ceiling let ~20 MB sit in flight,
|
||||||
|
// which on a mobile uplink is minutes of buffered queue rather
|
||||||
|
// than throughput — the app kept handing over data long after
|
||||||
|
// the path had stopped draining it.
|
||||||
|
s.iter().any(|ses| ses.machine.in_flight_count() < ses.machine.cwnd_packets().clamp(16, 1024))
|
||||||
}).unwrap_or(true) => {
|
}).unwrap_or(true) => {
|
||||||
self.handle_proxy_event(proxy_ev, &mut sessions_opt, &mut stream_map, &tx, &proxy_tx).await;
|
self.handle_proxy_event(proxy_ev, &mut sessions_opt, &mut stream_map, &tx, &proxy_tx).await;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -65,6 +65,18 @@ const MIN_CWND_PACKETS: u64 = 2;
|
||||||
/// Min RTT expiry window (after which we re-probe)
|
/// Min RTT expiry window (after which we re-probe)
|
||||||
const MIN_RTT_EXPIRY: Duration = Duration::from_secs(10);
|
const MIN_RTT_EXPIRY: Duration = Duration::from_secs(10);
|
||||||
/// Minimum RTO (RFC 6298: 1s in TCP; we use 50ms since we own the protocol)
|
/// Minimum RTO (RFC 6298: 1s in TCP; we use 50ms since we own the protocol)
|
||||||
|
/// Absolute ceiling on the congestion window, in packets. At a ~1200-byte MTU
|
||||||
|
/// this is roughly 1.2 MB in flight — already far above the bandwidth-delay
|
||||||
|
/// product of any link this protocol realistically runs over, so anything
|
||||||
|
/// beyond it is standing queue, not throughput. The client previously allowed
|
||||||
|
/// up to 16384 packets (~20 MB), which on a mobile uplink is minutes of buffer.
|
||||||
|
const MAX_CWND_PACKETS: u64 = 1024;
|
||||||
|
/// SRTT/min_rtt ratio at which slow start stops. Doubling is what fills a deep
|
||||||
|
/// buffer fastest, so growth must end when the queue starts building rather
|
||||||
|
/// than waiting for a loss that a deep buffer may never produce.
|
||||||
|
const RTT_INFLATION_EXIT_SLOW_START: f64 = 2.0;
|
||||||
|
/// SRTT/min_rtt ratio treated as a standing queue that must be actively drained.
|
||||||
|
const RTT_INFLATION_BACKOFF: f64 = 4.0;
|
||||||
const RTO_MIN: Duration = Duration::from_millis(50);
|
const RTO_MIN: Duration = Duration::from_millis(50);
|
||||||
/// Maximum RTO
|
/// Maximum RTO
|
||||||
const RTO_MAX: Duration = Duration::from_secs(16);
|
const RTO_MAX: Duration = Duration::from_secs(16);
|
||||||
|
|
@ -198,9 +210,46 @@ impl CongestionController {
|
||||||
|
|
||||||
/// Congestion-window growth shared by both ACK paths (slow start / probe).
|
/// Congestion-window growth shared by both ACK paths (slow start / probe).
|
||||||
fn grow_window(&mut self, bytes: u64) {
|
fn grow_window(&mut self, bytes: u64) {
|
||||||
// State machine
|
// ── Delay-based congestion signal ────────────────────────────────────
|
||||||
|
// A loss-only controller is blind on a deeply-buffered path, and mobile
|
||||||
|
// carrier buffers are very deep: they absorb a burst instead of dropping
|
||||||
|
// it, so no loss is ever signalled and cwnd keeps growing. The queue —
|
||||||
|
// not the link — is what grows, and the standing delay it adds shows up
|
||||||
|
// as RTT inflating far above the path's floor. Left unchecked this is a
|
||||||
|
// positive feedback loop: bigger queue -> larger RTT samples -> larger
|
||||||
|
// SRTT -> larger RTO -> retransmits pile on -> bigger queue, which is
|
||||||
|
// how a session ends up reporting multi-second (even multi-minute) RTT
|
||||||
|
// and stalls video until the buffer finally drains or the user
|
||||||
|
// reconnects. Treat sustained RTT inflation as congestion in its own
|
||||||
|
// right, exactly as it is.
|
||||||
|
let inflation = if self.rtt_initialized && !self.min_rtt.is_zero() {
|
||||||
|
self.srtt.as_secs_f64() / self.min_rtt.as_secs_f64()
|
||||||
|
} else {
|
||||||
|
1.0
|
||||||
|
};
|
||||||
|
|
||||||
|
if inflation >= RTT_INFLATION_BACKOFF {
|
||||||
|
// Standing queue is severe — actively drain it.
|
||||||
|
self.cwnd = (self.cwnd / 2).max(MIN_CWND_PACKETS * self.mtu);
|
||||||
|
self.ssthresh = self.cwnd;
|
||||||
|
self.phase = Phase::ProbeBandwidth;
|
||||||
|
tracing::debug!(cwnd = self.cwnd, inflation, "congestion: draining standing queue");
|
||||||
|
self.clamp_cwnd();
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
match self.phase {
|
match self.phase {
|
||||||
Phase::SlowStart => {
|
Phase::SlowStart => {
|
||||||
|
// Exponential doubling is what fills a deep buffer fastest, so
|
||||||
|
// leave slow start as soon as the queue starts to build rather
|
||||||
|
// than waiting for the loss that may never come.
|
||||||
|
if inflation >= RTT_INFLATION_EXIT_SLOW_START {
|
||||||
|
self.ssthresh = self.cwnd;
|
||||||
|
self.phase = Phase::ProbeBandwidth;
|
||||||
|
tracing::debug!(cwnd = self.cwnd, inflation, "congestion: RTT inflation ended slow start");
|
||||||
|
self.clamp_cwnd();
|
||||||
|
return;
|
||||||
|
}
|
||||||
// Exponential growth: increase cwnd by acked bytes (doubles per RTT)
|
// Exponential growth: increase cwnd by acked bytes (doubles per RTT)
|
||||||
self.cwnd = self.cwnd.saturating_add(bytes);
|
self.cwnd = self.cwnd.saturating_add(bytes);
|
||||||
if self.cwnd >= self.ssthresh {
|
if self.cwnd >= self.ssthresh {
|
||||||
|
|
@ -213,6 +262,21 @@ impl CongestionController {
|
||||||
self.cwnd = self.cwnd.saturating_add(bytes * self.mtu / self.cwnd.max(1));
|
self.cwnd = self.cwnd.saturating_add(bytes * self.mtu / self.cwnd.max(1));
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
self.clamp_cwnd();
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Hard ceiling on the congestion window.
|
||||||
|
///
|
||||||
|
/// Independent of any estimate: no real path this protocol runs over has a
|
||||||
|
/// bandwidth-delay product anywhere near this, so a window above it is
|
||||||
|
/// buffered queue rather than data in transit. Without it, slow start on a
|
||||||
|
/// buffer that never drops could grow the window into the tens of megabytes.
|
||||||
|
fn clamp_cwnd(&mut self) {
|
||||||
|
let ceiling = MAX_CWND_PACKETS.saturating_mul(self.mtu);
|
||||||
|
if self.cwnd > ceiling {
|
||||||
|
self.cwnd = ceiling;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Record a loss event.
|
/// Record a loss event.
|
||||||
|
|
@ -332,6 +396,53 @@ mod tests {
|
||||||
assert!(cc.cwnd() < initial);
|
assert!(cc.cwnd() < initial);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// The bufferbloat case: a deep buffer absorbs everything, so NOTHING is
|
||||||
|
/// ever lost, but the standing queue inflates RTT. A loss-only controller
|
||||||
|
/// grows cwnd forever here — which is how a session ends up reporting
|
||||||
|
/// multi-second RTT and stalling video.
|
||||||
|
#[test]
|
||||||
|
fn test_rtt_inflation_halts_growth_without_any_loss() {
|
||||||
|
let mut cc = CongestionController::new(1200);
|
||||||
|
|
||||||
|
// Establish a low path floor; this becomes min_rtt.
|
||||||
|
for _ in 0..4 {
|
||||||
|
cc.on_send(1200);
|
||||||
|
cc.on_ack(1200, Duration::from_millis(20));
|
||||||
|
}
|
||||||
|
let cwnd_before = cc.cwnd();
|
||||||
|
|
||||||
|
// Queue builds: RTT climbs far above the floor, still zero loss.
|
||||||
|
for _ in 0..20 {
|
||||||
|
cc.on_send(1200);
|
||||||
|
cc.on_ack(1200, Duration::from_millis(400));
|
||||||
|
}
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
cc.cwnd() <= cwnd_before,
|
||||||
|
"cwnd kept growing while the queue was inflating RTT ({} -> {})",
|
||||||
|
cwnd_before,
|
||||||
|
cc.cwnd()
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// cwnd must never exceed the absolute ceiling, however long slow start
|
||||||
|
/// runs unopposed — above it the window is buffered queue, not throughput.
|
||||||
|
#[test]
|
||||||
|
fn test_cwnd_never_exceeds_absolute_ceiling() {
|
||||||
|
let mut cc = CongestionController::new(1200);
|
||||||
|
// Constant RTT: no inflation signal, so only the hard cap can stop this.
|
||||||
|
for _ in 0..5000 {
|
||||||
|
cc.on_send(1200);
|
||||||
|
cc.on_ack(1200, Duration::from_millis(30));
|
||||||
|
}
|
||||||
|
assert!(
|
||||||
|
cc.cwnd() <= MAX_CWND_PACKETS * 1200,
|
||||||
|
"cwnd {} exceeded the {}-packet ceiling",
|
||||||
|
cc.cwnd(),
|
||||||
|
MAX_CWND_PACKETS
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_isolated_slow_start_loss_does_not_exit_slow_start() {
|
fn test_isolated_slow_start_loss_does_not_exit_slow_start() {
|
||||||
// A single dropped packet (wireless noise, a brief handover blip) is
|
// A single dropped packet (wireless noise, a brief handover blip) is
|
||||||
|
|
|
||||||
|
|
@ -4,6 +4,11 @@ use thiserror::Error;
|
||||||
use std::collections::{BTreeMap, VecDeque};
|
use std::collections::{BTreeMap, VecDeque};
|
||||||
use std::time::{Duration, Instant};
|
use std::time::{Duration, Instant};
|
||||||
|
|
||||||
|
/// Upper bound on a single frame's retransmit timer, after exponential backoff
|
||||||
|
/// is applied to the adaptive RTO. Past this the session is dead from the
|
||||||
|
/// user's point of view, and waiting longer only delays recovery.
|
||||||
|
const MAX_EFFECTIVE_RTO: Duration = Duration::from_secs(8);
|
||||||
|
|
||||||
use crate::congestion::CongestionController;
|
use crate::congestion::CongestionController;
|
||||||
use crate::crypto::{NoiseRole, NoiseSession, SessionCipher};
|
use crate::crypto::{NoiseRole, NoiseSession, SessionCipher};
|
||||||
use crate::framing::{AdaptivePadder, FrameHeader, FrameKind, FramedPacket, PaddingStrategy};
|
use crate::framing::{AdaptivePadder, FrameHeader, FrameKind, FramedPacket, PaddingStrategy};
|
||||||
|
|
@ -675,8 +680,15 @@ impl ProtocolMachine {
|
||||||
break;
|
break;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Exponential backoff, but bounded in absolute terms. base_rto is
|
||||||
|
// itself adaptive and can reach RTO_MAX (16s) on a congested path;
|
||||||
|
// multiplying that by the 64x backoff cap yields a frame that sits
|
||||||
|
// unretransmitted for ~17 MINUTES, long past the point where the
|
||||||
|
// session is simply dead to the user. Cap the product so backoff
|
||||||
|
// stays a backoff rather than an outage.
|
||||||
let backoff_factor = 1u64 << (frame.retries as u64).min(6);
|
let backoff_factor = 1u64 << (frame.retries as u64).min(6);
|
||||||
let effective_rto = Duration::from_millis(base_rto_ms.saturating_mul(backoff_factor));
|
let effective_rto = Duration::from_millis(base_rto_ms.saturating_mul(backoff_factor))
|
||||||
|
.min(MAX_EFFECTIVE_RTO);
|
||||||
|
|
||||||
if now.duration_since(frame.last_sent) >= effective_rto {
|
if now.duration_since(frame.last_sent) >= effective_rto {
|
||||||
// Only burn the retry counter and reset the RTO timer when the
|
// Only burn the retry counter and reset the RTO timer when the
|
||||||
|
|
|
||||||
|
|
@ -28,6 +28,12 @@ enum Commands {
|
||||||
Init {
|
Init {
|
||||||
mode: String,
|
mode: String,
|
||||||
},
|
},
|
||||||
|
/// Hash a password for the web panel's `api.password_hash` config field
|
||||||
|
#[command(name = "hash-password", alias = "hp")]
|
||||||
|
HashPassword {
|
||||||
|
/// The password to hash. Omit to be prompted (keeps it out of shell history).
|
||||||
|
password: Option<String>,
|
||||||
|
},
|
||||||
/// Generate a new secure access key
|
/// Generate a new secure access key
|
||||||
#[command(name = "gk", alias = "generate-key")]
|
#[command(name = "gk", alias = "generate-key")]
|
||||||
GenerateKey {
|
GenerateKey {
|
||||||
|
|
@ -920,6 +926,38 @@ async fn run_app() -> Result<()> {
|
||||||
match cmd {
|
match cmd {
|
||||||
Commands::Setup { init } => { args.setup = true; args.init = init; }
|
Commands::Setup { init } => { args.setup = true; args.init = init; }
|
||||||
Commands::Init { mode } => { args.init = Some(mode); }
|
Commands::Init { mode } => { args.init = Some(mode); }
|
||||||
|
Commands::HashPassword { password } => {
|
||||||
|
// The panel stores only a hash, and until now nothing in the CLI
|
||||||
|
// could produce one: `ostp init server` writes password_hash: ""
|
||||||
|
// and the only generator lived inside the Unix-only Server+Panel
|
||||||
|
// wizard branch, leaving no supported way to set up API auth on a
|
||||||
|
// plain server.
|
||||||
|
let password = match password {
|
||||||
|
Some(p) => p,
|
||||||
|
None => {
|
||||||
|
print!("Password: ");
|
||||||
|
use std::io::Write as _;
|
||||||
|
std::io::stdout().flush().ok();
|
||||||
|
let mut buf = String::new();
|
||||||
|
std::io::stdin().read_line(&mut buf)?;
|
||||||
|
buf.trim_end_matches(['\r', '\n']).to_string()
|
||||||
|
}
|
||||||
|
};
|
||||||
|
if password.is_empty() {
|
||||||
|
anyhow::bail!("password must not be empty");
|
||||||
|
}
|
||||||
|
// Must match api.rs's handle_login byte for byte.
|
||||||
|
let hash = format!(
|
||||||
|
"{:x}",
|
||||||
|
<sha2::Sha256 as sha2::Digest>::digest(password.as_bytes())
|
||||||
|
);
|
||||||
|
println!();
|
||||||
|
println!("Add this to the \"api\" section of your config:");
|
||||||
|
println!();
|
||||||
|
println!(" \"password_hash\": \"{hash}\"");
|
||||||
|
println!();
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
Commands::GenerateKey { format, count } => { args.generate_key = true; args.format = format; args.count = count; }
|
Commands::GenerateKey { format, count } => { args.generate_key = true; args.format = format; args.count = count; }
|
||||||
Commands::Links => { args.links = true; }
|
Commands::Links => { args.links = true; }
|
||||||
Commands::Check => { args.check = true; }
|
Commands::Check => { args.check = true; }
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue