crates/estate-front/src/forward.rs
//! **A plain TCP port on the front, forwarded to the PRIMARY** — gunnar's
//! forge door (tcp/2222 → the primary's git listener) and holger's via-front road; the ONE
//! copy (RA5-34: gunnar-front kept a second one, and neither capped its connections or let
//! an idle one go). [`ForwardOptions`]: an address it refuses (a front's utility address), how
//! long a primary that does not answer yet is dialled again, how many connections it carries
//! at once (past it a new one is closed at once, said), and how long a silent one is kept.
//!
//! The target is the verdict, not a setting: the primary's utility address out
//! of what the installer fed, looked at again by the product's loop, and a
//! change RE-TARGETS LIVE ([`Forward::retarget`]): the next connection goes to
//! the new primary, no restart, and connections already open finish where they
//! began. Until a primary is known a connection is closed at once — refused,
//! never parked. Byte for byte (`tokio::io::copy_bidirectional`): a protocol
//! that is end to end by itself (SSH) stays end to end. Pure Rust, no
//! nftables, no iptables.
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, RwLock};
use std::time::Duration;
/// How long ONE dial to the primary may take.
const DIAL: Duration = Duration::from_secs(5);
/// The pause between two dials inside [`ForwardOptions::dial_window`].
const REDIAL: Duration = Duration::from_secs(1);
/// **How a forward carries** (RA5-34).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ForwardOptions {
/// A connection that arrives at this local address is closed at once (a front's utility
/// address: boxes speak to the front over the tunnel).
pub refused: Option<Ipv4Addr>,
/// How long a client is held while the primary cannot be dialled yet, dialled again every
/// second (a front that just booted reaches its primary only once its tunnel is up).
pub dial_window: Duration,
/// The most connections carried at once; past it a new one is closed at once.
pub max_connections: usize,
/// A connection with no byte either way for this long is closed.
pub idle: Duration,
}
/// The connections a forward carries at once, by default.
pub const MAX_CONNECTIONS: usize = 512;
/// How long a silent connection is kept, by default.
pub const IDLE: Duration = Duration::from_secs(15 * 60);
impl Default for ForwardOptions {
fn default() -> Self {
ForwardOptions {
refused: None,
dial_window: Duration::from_secs(10),
max_connections: MAX_CONNECTIONS,
idle: IDLE,
}
}
}
/// Pure: **where the port goes** — the primary's utility address, `port`.
/// `None` until the election has spoken.
pub fn target_of(primary_utility: Option<Ipv4Addr>, port: u16) -> Option<SocketAddrV4> {
primary_utility.map(|ip| SocketAddrV4::new(ip, port))
}
/// The forward: a listener, and the one target it hands connections to.
#[derive(Clone)]
pub struct Forward {
target: Arc<RwLock<Option<SocketAddrV4>>>,
/// Where it listens (the bound address — a test binds port 0).
pub local: SocketAddr,
what: &'static str,
live: Arc<AtomicUsize>,
}
/// **Dial the primary, again and again, for up to `window`**: each try reads the target anew
/// (a retarget meanwhile is followed), waits at most [`DIAL`], and the next comes after
/// [`REDIAL`]. Answers the address last tried and the last outcome.
async fn dial_within(
target: &RwLock<Option<SocketAddrV4>>,
first: SocketAddrV4,
window: Duration,
) -> (
SocketAddrV4,
Result<std::io::Result<tokio::net::TcpStream>, tokio::time::error::Elapsed>,
) {
let until = tokio::time::Instant::now() + window;
let mut to = first;
loop {
let r = tokio::time::timeout(
DIAL.min(window.max(Duration::from_millis(1))),
tokio::net::TcpStream::connect(to),
)
.await;
if matches!(r, Ok(Ok(_))) || tokio::time::Instant::now() + REDIAL >= until {
return (to, r);
}
tokio::time::sleep(REDIAL).await;
to = target.read().ok().and_then(|t| *t).unwrap_or(to);
}
}
/// **Carry bytes both ways until both sides are done, or `idle` passes with none** (a half
/// close is passed on: one side's end shuts the other's write, and the other way still flows).
async fn carry(
a: tokio::net::TcpStream,
b: tokio::net::TcpStream,
idle: Duration,
) -> std::io::Result<()> {
use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};
let (mut ar, mut aw) = a.into_split();
let (mut br, mut bw) = b.into_split();
let (mut abuf, mut bbuf) = (vec![0u8; 16 * 1024], vec![0u8; 16 * 1024]);
let (mut a_done, mut b_done) = (false, false);
while !(a_done && b_done) {
tokio::select! {
r = ar.read(&mut abuf), if !a_done => match r? {
0 => { a_done = true; let _ = bw.shutdown().await; }
n => bw.write_all(&abuf[..n]).await?,
},
r = br.read(&mut bbuf), if !b_done => match r? {
0 => { b_done = true; let _ = aw.shutdown().await; }
n => aw.write_all(&bbuf[..n]).await?,
},
_ = tokio::time::sleep(idle) => return Err(std::io::Error::new(std::io::ErrorKind::TimedOut, "idle")),
}
}
Ok(())
}
/// One live connection, counted while it lives.
struct Held(Arc<AtomicUsize>);
impl Drop for Held {
fn drop(&mut self) {
self.0.fetch_sub(1, Ordering::SeqCst);
}
}
impl Forward {
/// **Listen on `bind`**, and forward every connection to the current
/// target, with the default [`ForwardOptions`]. `what` names it in the log (`"forge"`).
pub async fn bind(bind: SocketAddr, what: &'static str) -> std::io::Result<Forward> {
Forward::bind_with(bind, what, ForwardOptions::default()).await
}
/// [`Forward::bind`] with `opts`.
pub async fn bind_with(
bind: SocketAddr,
what: &'static str,
opts: ForwardOptions,
) -> std::io::Result<Forward> {
let listener = tokio::net::TcpListener::bind(bind).await?;
let f = Forward {
target: Arc::new(RwLock::new(None)),
local: listener.local_addr()?,
what,
live: Arc::new(AtomicUsize::new(0)),
};
let (target, live) = (Arc::clone(&f.target), Arc::clone(&f.live));
// A32-14, stays a task: the forward's accept loop, for the life of its listener.
tokio::spawn(async move {
let mut full_said = std::time::Instant::now() - Duration::from_secs(60);
loop {
let (client, from) = match listener.accept().await {
Ok(c) => c,
Err(e) => {
tracing::warn!(error = %e, "front: {what} forward: accept failed");
tokio::time::sleep(Duration::from_millis(200)).await;
continue;
}
};
if let (Some(r), Ok(SocketAddr::V4(local))) = (opts.refused, client.local_addr()) {
if *local.ip() == r {
tracing::debug!(%from, %local, "front: {what} forward: a connection at the refused address is closed");
continue;
}
}
if live.fetch_add(1, Ordering::SeqCst) >= opts.max_connections {
live.fetch_sub(1, Ordering::SeqCst);
if full_said.elapsed() >= Duration::from_secs(60) {
tracing::warn!(%from, max = opts.max_connections, "front: {what} forward: it carries as many connections as it may; a new one is closed (said once a minute)");
full_said = std::time::Instant::now();
}
continue;
}
let held = Held(Arc::clone(&live));
let target = Arc::clone(&target);
// A32-14, stays a task: one forwarded connection, ended by either side closing or its idle time.
tokio::spawn(async move {
let _held = held;
let Some(first) = target.read().ok().and_then(|t| *t) else {
tracing::info!(%from, "front: {what} forward: no primary elected yet — the connection is closed");
return;
};
let (to, dialled) = dial_within(&target, first, opts.dial_window).await;
match dialled {
Ok(Ok(up)) => {
let _ = client.set_nodelay(true);
let _ = up.set_nodelay(true);
if let Err(e) = carry(client, up, opts.idle).await {
tracing::debug!(%from, %to, error = %e, "front: {what} forward: the session ended");
}
}
Ok(Err(e)) => {
tracing::warn!(%from, %to, error = %e, "front: {what} forward: the primary refused")
}
Err(_) => {
tracing::warn!(%from, %to, "front: {what} forward: the primary did not answer in time")
}
}
});
}
});
Ok(f)
}
/// **Point the forward at `to`** — live, for the next connection. Answers
/// whether it changed.
pub fn retarget(&self, to: Option<SocketAddrV4>) -> bool {
let Ok(mut t) = self.target.write() else {
return false;
};
if *t == to {
return false;
}
tracing::info!(from = ?*t, to = ?to, what = self.what, "front: forward re-targeted — the next connection goes to the primary");
*t = to;
true
}
pub fn target(&self) -> Option<SocketAddrV4> {
self.target.read().ok().and_then(|t| *t)
}
/// The connections it carries now.
pub fn carried(&self) -> usize {
self.live.load(Ordering::SeqCst)
}
}
#[cfg(test)]
mod tests {
use super::*;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
#[test]
fn the_forward_goes_to_the_primary_the_installer_names() {
assert_eq!(target_of(None, 2222), None, "no verdict, no target");
assert_eq!(
target_of(Some(Ipv4Addr::new(10, 13, 8, 2)), 2222),
Some(SocketAddrV4::new(Ipv4Addr::new(10, 13, 8, 2), 2222))
);
}
/// A stand-in listener that says its name first, as sshd does.
async fn banner(name: &'static str) -> SocketAddrV4 {
let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
let SocketAddr::V4(at) = l.local_addr().unwrap() else {
unreachable!()
};
tokio::spawn(async move {
loop {
let (mut s, _) = l.accept().await.unwrap();
tokio::spawn(async move {
s.write_all(format!("SSH-2.0-{name}\r\n").as_bytes())
.await
.unwrap();
let mut b = [0u8; 16];
if let Ok(n) = s.read(&mut b).await {
let _ = s.write_all(&b[..n]).await;
}
});
}
});
at
}
async fn hello(f: &Forward) -> Option<String> {
let mut s = tokio::net::TcpStream::connect(f.local).await.unwrap();
let mut b = vec![0u8; 64];
let n = tokio::time::timeout(Duration::from_secs(5), s.read(&mut b))
.await
.ok()?
.ok()?;
if n == 0 {
return None;
}
s.write_all(b"ping").await.unwrap();
let mut echo = [0u8; 4];
s.read_exact(&mut echo).await.unwrap();
assert_eq!(&echo, b"ping", "bytes flow both ways");
Some(String::from_utf8_lossy(&b[..n]).trim().to_string())
}
/// Byte for byte to the primary; closed while there is none; and a new
/// primary is taken LIVE — the next connection goes there, no restart.
#[tokio::test]
async fn the_forward_carries_the_session_and_retargets_live() {
let (a, b) = (banner("main").await, banner("new-main").await);
let f = Forward::bind("127.0.0.1:0".parse().unwrap(), "test")
.await
.unwrap();
assert_eq!(
hello(&f).await,
None,
"no primary yet: the connection is closed"
);
assert!(f.retarget(Some(a)));
assert_eq!(hello(&f).await.as_deref(), Some("SSH-2.0-main"));
assert!(!f.retarget(Some(a)), "the same primary is no change");
assert!(f.retarget(Some(b)));
assert_eq!(hello(&f).await.as_deref(), Some("SSH-2.0-new-main"));
assert_eq!(f.target(), Some(b));
}
/// ★ **RA5-34: the one forward caps its connections and lets a silent one go.** Neither
/// copy (this one and gunnar-front's tcp/2222) did: a caller could hold any number of
/// sockets on the front for ever. Past the cap a new connection is closed at once; one with
/// no byte either way for `idle` is closed; one the caller ended frees its place.
#[tokio::test]
async fn the_forward_caps_its_connections_and_lets_an_idle_one_go() {
let a = banner("main").await;
let opts = ForwardOptions {
max_connections: 2,
idle: Duration::from_millis(400),
..ForwardOptions::default()
};
let f = Forward::bind_with("127.0.0.1:0".parse().unwrap(), "test", opts)
.await
.unwrap();
assert!(f.retarget(Some(a)));
let mut held = Vec::new();
for _ in 0..2 {
let mut s = tokio::net::TcpStream::connect(f.local).await.unwrap();
let mut b = [0u8; 64];
assert!(s.read(&mut b).await.unwrap() > 0, "carried");
held.push(s);
}
let mut third = tokio::net::TcpStream::connect(f.local).await.unwrap();
let mut b = [0u8; 64];
let n = tokio::time::timeout(Duration::from_secs(2), third.read(&mut b))
.await
.expect("closed at once")
.unwrap_or(0);
assert_eq!(n, 0, "past the cap: closed, nothing carried");
// Silent past `idle`: both held ones are let go, and their places are free again.
tokio::time::sleep(Duration::from_millis(900)).await;
for s in &mut held {
let n = tokio::time::timeout(Duration::from_secs(2), s.read(&mut b))
.await
.expect("closed")
.unwrap_or(0);
assert_eq!(n, 0, "an idle connection is closed");
}
assert_eq!(f.carried(), 0);
assert_eq!(
hello(&f).await.as_deref(),
Some("SSH-2.0-main"),
"a place is free again"
);
}
/// The refused address (a front's utility address) is closed; any other is carried.
#[tokio::test]
async fn a_connection_at_the_refused_address_is_closed() {
let a = banner("main").await;
let opts = ForwardOptions {
refused: Some(Ipv4Addr::LOCALHOST),
..ForwardOptions::default()
};
let f = Forward::bind_with("0.0.0.0:0".parse().unwrap(), "test", opts)
.await
.unwrap();
assert!(f.retarget(Some(a)));
let at = |ip: Ipv4Addr| Forward {
local: SocketAddr::new(ip.into(), f.local.port()),
..f.clone()
};
assert_eq!(
hello(&at(Ipv4Addr::LOCALHOST)).await,
None,
"the refused address: closed"
);
assert_eq!(
hello(&at(Ipv4Addr::new(127, 0, 0, 2))).await.as_deref(),
Some("SSH-2.0-main"),
"any other address: carried"
);
}
}