Feat: Leave a shaped path out of the session

docs/SPEC.md

@@ -2289,6 +2289,7 @@ Hypothesis: one long-running process holding a list of paired node ids carries a
 2. **Port prediction.** When one side is symmetric and the other is not, the symmetric side's next mapped ports are predicted from the observed step, and the other side sends to a small window of them: the last mapped port plus the step times 1 to 16 (`PREDICT_WINDOW`), stopping at the ends of the range 1024 to 65535. A step of 0 predicts nothing.
 3. **Many-port attempts.** When both sides are symmetric, or prediction fails, each side opens up to 256 ports and sends to up to 256 destination ports (the birthday attempt). The limit is a library setting (`BIRTHDAY_PORTS`) and a UDP punch sends at most 500 datagrams a second in total (`SPRAY_PER_SECOND`), cycling through the socket and destination pairs. With 256 mappings on one side and 256 random destinations on the other, one hit has a probability of about 0.64.
 4. **Policing detection.** After a path is established, the library sends a bulk burst on it before declaring it usable. A path that answers 30 round trips and then delivers under a twentieth of the floor, as [Link floor](#link-floor) describes, is reported as `policed` with the measured rates. The caller then rejects it as a failure. The check is `policing::check` on the client and `policing::serve` on the peer, run on the subflow before the session handshake. It runs 30 round trips, then a bulk burst up and a bulk burst down of 1 s each, in frames of 16 KiB with at most 64 KiB unacknowledged, so a throttled path is measured within a bounded wait. Each rate is the bytes delivered divided by the time until the peer confirmed the end. A direction under a twentieth of the floor gives `Policed`. A stream reported as policed is not reused.
+   - **Shaped paths.** When the caller gives a `shaped_ratio`, the initiator also compares the usable paths of one punch with each other. A path whose measured rate is under `shaped_ratio` times the best usable path's rate in either direction is a shaped path and is dropped like a policed one, with the reason `shaped`. letify gives 0.05 (`policing.SHAPED_RATIO`). The comparison is per direction (up against the best up, down against the best down), and the measurement is the burst of item 4, which already stands for the confirmation rule of Paths and the path manager item 6 because it runs for the burst time over bounded frames. The path with the highest share of the best in its weaker direction is never dropped, so a punch never loses its only usable path. Paths that measured 0 in a direction give no best for it, and a set of one path has nothing to compare with. The reason: a path that delivers 2% of the others still carries its share of a stream until it is reinjected, so it adds head-of-line delay for small messages under bulk while adding nothing to the rate. The measurement and the evidence are in the pull request `Feat: Leave a shaped path out of the session`.
 
 A UDP punch succeeds on the first acknowledgement of its probe and then sends two more, so the peer sees one too. That shows the path works in both directions at that moment, and the session handshake over it is the proof. A punch yields at most one UDP path, however many sockets probe: `udp: N` is the number of probing sockets, which raises the chance that one pair of them gets through, and it is not a number of paths. The two sides do not choose the pair alike. The initiator alone chooses it, as its first acknowledged socket and peer address, and opens the QUIC connection from it. The responder keeps all its probing sockets listening for a QUIC handshake, as the TCP responder keeps the connection on which the hello arrives, keeps the socket and the source address of the first handshake that completes, and closes the rest. It waits no longer than the punch window plus the QUIC connect time (10 s).
 5. **No Tailcat.** The UDP address exchange moves to the Rendezvous trait and the STUN step above. letify has no Tailcat: `letify setup tailcat`, `letify client shell connect` and the Tunnel account's `tailcat` and `tailcat_port` fields do not exist, and an account that names them is refused ([Connection strategies](#connection-strategies)). A Tunnel account names an agent.
@@ -2358,7 +2359,7 @@ The sidecar is a binary named `letify-link`, built for each wheel and installed
      |---|---|---|
      | `listen` | `allow`: node ids that may start a session; `bind`: optional TCP address to listen on | `listener`: integer; `addr`: the bound address when `bind` was given |
      | `open_session` | `peer`: node id; `candidates`: see below; `config`: optional `keepalive_ms`, `resume_window_s`, `handshake_timeout_ms`, `start_timeout_ms` (how long to wait for the first handshake, default 60000) | `session`: integer; `peer` |
-     | `punch` | `peer`: node id; `role`: `initiator` or `responder`; `sockets`: TCP sockets, 0 to 8 (default 4); `udp`: probing UDP sockets, 0 to 8 (default 0), of which a punch yields at most one UDP path; `observers`: where the mappings come from, see item 3; `public_ip`: optional address reported as the mapping of every socket when `observers` is absent (default `127.0.0.1`); `police`: optional `{"floor_bps", "burst_ms"}`, the initiator's policing check; `lead_ms`, `window_ms`: optional; `answer_margin_ms`: optional, responder only, see Offers; `listener`: required for the responder | initiator: `punched`: names usable as candidates, `policed`: names of paths that were dropped, `nat`: this side's classification, `peer_nat`: the other side's, `mapped`: `{"tcp": [...], "udp": [...]}`, the addresses this side offered, `report`: the one-line account of [NAT traversal](#link-nat) item 8; responder: `sockets`: how many paths were punched, `nat`, `peer_nat`, `mapped` as for the initiator |
+     | `punch` | `peer`: node id; `role`: `initiator` or `responder`; `sockets`: TCP sockets, 0 to 8 (default 4); `udp`: probing UDP sockets, 0 to 8 (default 0), of which a punch yields at most one UDP path; `observers`: where the mappings come from, see item 3; `public_ip`: optional address reported as the mapping of every socket when `observers` is absent (default `127.0.0.1`); `police`: optional `{"floor_bps", "burst_ms", "shaped_ratio"}`, the initiator's policing check, where `shaped_ratio` is optional and turns on the shaped path rule of [NAT traversal](#link-nat) item 4 (absent: off); `lead_ms`, `window_ms`: optional; `answer_margin_ms`: optional, responder only, see Offers; `listener`: required for the responder | initiator: `punched`: names usable as candidates, `policed`: names of paths that were dropped, `nat`: this side's classification, `peer_nat`: the other side's, `mapped`: `{"tcp": [...], "udp": [...]}`, the addresses this side offered, `report`: the one-line account of [NAT traversal](#link-nat) item 8; responder: `sockets`: how many paths were punched, `nat`, `peer_nat`, `mapped` as for the initiator |
      | `rendezvous_message` | `from`: node id; `message`: hex | none besides `ok` |
      | `paths` | `session` | `paths`: `[{"id","kind","sent_bytes","received_bytes","acked_bytes","rtt_ms","srtt_ms","window","probe_rate"}]`, where `kind` is `"tcp"`, `"udp"` or `"relay"`, the transport kind of the candidate the subflow came from, and null on a session the other side started |
      | `open_stream` | `session`; `priority`: optional, `interactive` (default) or `bulk`, as Session item 7 says | `stream`: integer; `socket`: local stream socket; `secret`: present only for a `tcp:` socket |
@@ -2375,7 +2376,7 @@ The sidecar is a binary named `letify-link`, built for each wheel and installed
      | `path_lost` | `session`, `name`, `reason` |
      | `path_probe` | `session`, `name`, `throughput` (bytes per second), `rtt_ms` or null |
      | `path_preferred` | `session`, `from` or null, `to` |
-     | `path_policed` | `peer`, `name`, `proto` (`tcp` or `udp`), `up_bps`, `down_bps`: a punched path that answered pings but delivered under a twentieth of the floor, dropped before any session used it |
+     | `path_policed` | `peer`, `name`, `proto` (`tcp` or `udp`), `up_bps`, `down_bps`, `reason` (`policed`: under a twentieth of the floor, `shaped`: under `shaped_ratio` of the best path): a punched path that answered pings but was dropped before any session used it |
      | `session_incoming` | `listener`, `session`, `peer` |
      | `session_closed` | `session`, `reason` |
      | `stream_incoming` | `session`, `stream`, `socket`, `secret` (only for a `tcp:` socket) |
@@ -2401,7 +2402,7 @@ The sidecar is a binary named `letify-link`, built for each wheel and installed
      4. **Verify.** The final remote program checks the size and sha256 of `<path>.part` against the local values, then sets mode 0700 and renames it to `<path>`. A mismatch deletes the part file and fails the connect with `letify-link upload: checksum mismatch on <alias>: expected <sha256>, got <sha256>`. A failed upload never falls back to the wheel.
 
      The uploaded path is then the `link_binary` of the connect, so a missing file fails as above. The option is for development: the binary must be built for the runtime by the user, and nothing records its version.
-   - **The shim.** letify runs `letify/transport/linkshim.py`, a standard-library program, with the remote's Python over ssh. The shim starts the remote sidecar, listens on a TCP port with only the local node id in `allow`, and prints one line of JSON with the remote node id and the port. The local side dials the account's `address` at that port as a `tcp` candidate, when the account has an `address`. The shim and the local side then exchange rendezvous messages as JSON lines on the ssh command's standard input and output (`rendezvous_out` from one side is `rendezvous_message` on the other), which is what the `punch` operation uses when the account sets `link_punch = true` or has no `address`. For a punch the local side asks for 4 TCP sockets and 1 UDP socket, `observers` `"default"` (an account with an `address` keeps `public_ip` = `address`, because that address is reachable) and the policing check with a floor of 1 MiB/s and bursts of 500 ms, and the shim, as the responder, answers with the same counts.
+   - **The shim.** letify runs `letify/transport/linkshim.py`, a standard-library program, with the remote's Python over ssh. The shim starts the remote sidecar, listens on a TCP port with only the local node id in `allow`, and prints one line of JSON with the remote node id and the port. The local side dials the account's `address` at that port as a `tcp` candidate, when the account has an `address`. The shim and the local side then exchange rendezvous messages as JSON lines on the ssh command's standard input and output (`rendezvous_out` from one side is `rendezvous_message` on the other), which is what the `punch` operation uses when the account sets `link_punch = true` or has no `address`. For a punch the local side asks for 4 TCP sockets and 1 UDP socket, `observers` `"default"` (an account with an `address` keeps `public_ip` = `address`, because that address is reachable) and the policing check with a floor of 1 MiB/s, bursts of 500 ms and a `shaped_ratio` of 0.05, and the shim, as the responder, answers with the same counts.
    - **The hand-over.** When the session has started, the local side tells the shim, the shim closes its standard output and keeps running without ssh, and the ssh command ends. The shim exits when its session closes, so the remote holds nothing after the lease ends.
 
    Then no frame uses ssh. The local `LinkChannel` opens a stream with `open_stream` and speaks the framed worker protocol on it. The shim accepts each stream and starts the same worker program that ssh starts today, the bootstrap stub of [Channels](#channels), with the stream socket as its standard input and output. Each parallel transfer of [Several connections at once](#parallel-placement) is one more stream of the same session, so it needs no further punch and no further ssh connection. The session keeps the worker channel across a lost path as [Session](#link-session) item 5 says: the call sees a pause, not an error.

letify-core/letify-link/src/policing.rs

@@ -33,6 +33,34 @@ const OP_DOWN: u8 = b'D';
 const ACK: u8 = b'A';
 const END: u8 = b'E';
 
+/// The share of the best usable path's rate under which a path is shaped, in either direction.
+pub const SHAPED_RATIO: f64 = 0.05;
+
+/// The paths of one set that the shaped path rule leaves out: `true` at the index of a path
+/// whose rate in either direction is under `ratio` times the best path's rate in that
+/// direction. The path with the highest share of the best in its weaker direction is never
+/// left out, so the set keeps a path. A direction in which no path measured anything has no
+/// best, and a set of one path has nothing to compare with.
+pub fn shaped_out(rates: &[Rates], ratio: f64) -> Vec<bool> {
+    let best_up = rates.iter().map(|r| r.up_bps).fold(0.0, f64::max);
+    let best_down = rates.iter().map(|r| r.down_bps).fold(0.0, f64::max);
+    let share = |x: f64, best: f64| if best > 0.0 { x / best } else { 1.0 };
+    let weakest: Vec<f64> = rates
+        .iter()
+        .map(|r| share(r.up_bps, best_up).min(share(r.down_bps, best_down)))
+        .collect();
+    let keeper = weakest
+        .iter()
+        .enumerate()
+        .max_by(|a, b| a.1.total_cmp(b.1).then(b.0.cmp(&a.0)))
+        .map(|(i, _)| i);
+    weakest
+        .iter()
+        .enumerate()
+        .map(|(i, w)| Some(i) != keeper && *w < ratio)
+        .collect()
+}
+
 /// What the check measured.
 #[derive(Clone, Copy, Debug, PartialEq)]
 pub struct Rates {

letify-core/letify-link/src/sidecar.rs

@@ -1136,7 +1136,7 @@ async fn await_offers(
 
 /// The result of checking one punched path on the initiator.
 enum Vetted {
-    Usable(String, Punched),
+    Usable(String, Punched, Option<policing::Rates>),
     Policed(String, &'static str, policing::Rates),
     Failed(String, String),
 }
@@ -1151,14 +1151,15 @@ async fn vet<S>(
 where
     S: AsyncRead + AsyncWrite + Unpin + Send,
 {
+    let mut measured = None;
     if let Some((floor, burst)) = police {
         match policing::check(&mut stream, floor, burst).await {
-            Ok(Verdict::Usable(_)) => {}
+            Ok(Verdict::Usable(rates)) => measured = Some(rates),
             Ok(Verdict::Policed(rates)) => return Vetted::Policed(name, proto, rates),
             Err(e) => return Vetted::Failed(name, e.to_string()),
         }
     }
-    Vetted::Usable(name, wrap(stream))
+    Vetted::Usable(name, wrap(stream), measured)
 }
 
 /// Check a punched path as the responder, then hand it to the acceptor.
@@ -1207,6 +1208,14 @@ async fn op_punch(shared: &Arc<Shared>, req: &Value) -> OpResult {
             Duration::from_millis(opt_u64(p, "burst_ms", DEFAULT_BURST_MS)?),
         )),
     };
+    let shaped_ratio = match req.get("police").and_then(|p| p.get("shaped_ratio")) {
+        None | Some(Value::Null) => None,
+        Some(v) => Some(
+            v.as_f64()
+                .filter(|r| r.is_finite() && *r >= 0.0)
+                .ok_or_else(|| usage("shaped_ratio must be a number of at least 0"))?,
+        ),
+    };
     let lead = opt_u64(req, "lead_ms", DEFAULT_LEAD_MS)?;
     let window = opt_u64(req, "window_ms", DEFAULT_WINDOW_MS)?;
     let answer_margin = match req.get("answer_margin_ms") {
@@ -1479,26 +1488,64 @@ async fn op_punch(shared: &Arc<Shared>, req: &Value) -> OpResult {
                     }
                 });
             }
-            let mut names = Vec::new();
+            let mut usable: Vec<(String, Punched, Option<policing::Rates>)> = Vec::new();
             let mut policed = Vec::new();
             let mut failed = Vec::new();
             while let Some(done) = set.join_next().await {
                 match done {
-                    Ok(Vetted::Usable(name, path)) => {
-                        lock(&shared.punched).insert(name.clone(), path);
-                        names.push(name);
-                    }
+                    Ok(Vetted::Usable(name, path, rates)) => usable.push((name, path, rates)),
                     Ok(Vetted::Policed(name, proto, rates)) => {
                         shared.emit(json!({
                             "event": "path_policed", "peer": peer.to_string(), "name": name,
                             "proto": proto, "up_bps": rates.up_bps, "down_bps": rates.down_bps,
+                            "reason": "policed",
                         }));
                         policed.push(name);
                     }
                     Ok(Vetted::Failed(name, why)) => failed.push(format!("{name}: {why}")),
                     Err(e) => failed.push(e.to_string()),
                 }
             }
+            // The shaped path rule compares the paths that were measured with each other.
+            let measured: Vec<usize> = (0..usable.len())
+                .filter(|i| usable[*i].2.is_some())
+                .collect();
+            let shaped = match shaped_ratio {
+                Some(ratio) => {
+                    let rates: Vec<policing::Rates> =
+                        measured.iter().filter_map(|i| usable[*i].2).collect();
+                    policing::shaped_out(&rates, ratio)
+                }
+                None => vec![false; measured.len()],
+            };
+            let mut dropped = vec![false; usable.len()];
+            for (k, i) in measured.iter().enumerate() {
+                dropped[*i] = shaped[k];
+            }
+            let mut names = Vec::new();
+            for (i, (name, path, rates)) in usable.into_iter().enumerate() {
+                if dropped[i] {
+                    let rates = rates.unwrap_or(policing::Rates {
+                        rtt: Duration::ZERO,
+                        up_bps: 0.0,
+                        down_bps: 0.0,
+                    });
+                    let proto = if matches!(path, Punched::Udp(_)) {
+                        "udp"
+                    } else {
+                        "tcp"
+                    };
+                    shared.emit(json!({
+                        "event": "path_policed", "peer": peer.to_string(), "name": name,
+                        "proto": proto, "up_bps": rates.up_bps, "down_bps": rates.down_bps,
+                        "reason": "shaped",
+                    }));
+                    policed.push(name);
+                } else {
+                    lock(&shared.punched).insert(name.clone(), path);
+                    names.push(name);
+                }
+            }
             names.sort();
             policed.sort();
             if names.is_empty() {