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_sample_at: 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 {
|
||||
|
|
@ -173,6 +183,8 @@ impl Bridge {
|
|||
last_rtt_ms: 0.0,
|
||||
last_sample_at: 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!(
|
||||
"Resumed after ~{}s suspend — forcing clean reconnect", wall_gap.as_secs()
|
||||
))).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;
|
||||
// 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 {
|
||||
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| {
|
||||
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) => {
|
||||
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)
|
||||
const MIN_RTT_EXPIRY: Duration = Duration::from_secs(10);
|
||||
/// 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);
|
||||
/// Maximum RTO
|
||||
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).
|
||||
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 {
|
||||
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)
|
||||
self.cwnd = self.cwnd.saturating_add(bytes);
|
||||
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.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.
|
||||
|
|
@ -332,6 +396,53 @@ mod tests {
|
|||
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]
|
||||
fn test_isolated_slow_start_loss_does_not_exit_slow_start() {
|
||||
// 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::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::crypto::{NoiseRole, NoiseSession, SessionCipher};
|
||||
use crate::framing::{AdaptivePadder, FrameHeader, FrameKind, FramedPacket, PaddingStrategy};
|
||||
|
|
@ -675,8 +680,15 @@ impl ProtocolMachine {
|
|||
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 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 {
|
||||
// Only burn the retry counter and reset the RTO timer when the
|
||||
|
|
|
|||
|
|
@ -28,6 +28,12 @@ enum Commands {
|
|||
Init {
|
||||
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
|
||||
#[command(name = "gk", alias = "generate-key")]
|
||||
GenerateKey {
|
||||
|
|
@ -920,6 +926,38 @@ async fn run_app() -> Result<()> {
|
|||
match cmd {
|
||||
Commands::Setup { init } => { args.setup = true; args.init = init; }
|
||||
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::Links => { args.links = true; }
|
||||
Commands::Check => { args.check = true; }
|
||||
|
|
|
|||
Loading…
Reference in New Issue