Measure a link at rest: 132 bytes against a QUIC connection's 38,106 (094)

M1 of design 011. A Pure node allows 1,000 to 1,000,000 senders; each
sends one order and, once told it is stored, rests. At rest a link costs
the node 132 bytes (the sender allowed, 73; its number heard, about 60)
and no transport state; with a million the node runs in 139 MB. A quinn
0.11 service holding the same exchange's clients connected at rest costs
38,106 bytes a connection: 289 times as much, past the design's 1/100.
A link at rest takes its next record in 0.56 ms to stored, one X25519 on
the node included; a new QUIC connection takes 1.06 ms and 2.8 times the
receiver's CPU, without storing.

research/experiments/019-idle-links: links.c, many senders written from
serve/link.h's packet format alone (a second implementation), and the
quinn service and clients (quic/, built offline); run.py makes both
measurements.

1 files changed, 329 insertions, 0 deletions

diff --git a/research/experiments/019-idle-links/quic/src/main.rs b/research/experiments/019-idle-links/quic/src/main.rs
new file mode 100644
index 0000000..cdb0376
--- /dev/null
+++ b/research/experiments/019-idle-links/quic/src/main.rs
@@ -0,0 +1,329 @@
+//! A QUIC service holding many clients at rest, for M1 of
+//! docs/design/011-pure-transport.md: the same exchange as a Pure link's
+//! (a client sends one order, the service stores it, says so, and the two
+//! stay connected, quiet), over quinn 0.11 with rustls (AES-128-GCM).
+//!
+//!   idle-quic serve CERT.der KEY.der [PORT]
+//!       a service on 127.0.0.1 (the port on stdout as JSON), one thread;
+//!       a line of JSON on stdout at every 1,000th order stored. Its
+//!       stateless reset key comes from KEY.der, so one started again on
+//!       the same port resets the connections of the one before at once
+//!   idle-quic connect PORT FIRST COUNT WINDOW
+//!       clients FIRST.. each connect, send order 1 and wait until it is
+//!       stored, WINDOW at a time over one endpoint; a line of JSON; the
+//!       connections stay open until stdin closes; then COUNT orders more
+//!       (number 2) over them, and another line
+//!   idle-quic steady PORT FIRST COUNT PERIOD_MS SECONDS WINDOW
+//!       clients FIRST.. connect (WINDOW at a time), then each makes an
+//!       order every PERIOD_MS for SECONDS and sends it over its connection;
+//!       one that fails (no answer in a second, or the connection lost)
+//!       connects again and sends it again. Then a line of JSON: orders made
+//!       and stored, connections made again, the orders stored in each
+//!       100 ms, the time from an order made to it stored
+use std::collections::HashMap;
+use std::net::SocketAddr;
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant};
+
+use quinn::crypto::rustls::QuicClientConfig;
+use quinn::{ClientConfig, Endpoint, IdleTimeout, ServerConfig, TransportConfig, VarInt};
+use rustls::client::danger::{HandshakeSignatureValid, ServerCertVerified, ServerCertVerifier};
+use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, ServerName, UnixTime};
+use rustls::{DigitallySignedStruct, SignatureScheme};
+use tokio::io::AsyncReadExt;
+
+fn transport() -> Arc<TransportConfig> {
+    let mut t = TransportConfig::default();
+    t.max_idle_timeout(Some(IdleTimeout::from(VarInt::from_u32(3_600_000)))); // an hour: connections at rest stay
+    t.keep_alive_interval(None);
+    Arc::new(t)
+}
+
+/// The order: the client's number and the order's.
+fn order(client: u64, seq: u64) -> [u8; 64] {
+    let mut bytes = [0u8; 64];
+    bytes[..8].copy_from_slice(&client.to_le_bytes());
+    bytes[8..16].copy_from_slice(&seq.to_le_bytes());
+    bytes[16..19].copy_from_slice(b"pen");
+    bytes
+}
+
+fn serve(cert: Vec<u8>, key: Vec<u8>, port: u16) {
+    let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
+    runtime.block_on(async move {
+        let reset = ring::hmac::Key::new(ring::hmac::HMAC_SHA256, ring::digest::digest(&ring::digest::SHA256, &key).as_ref());
+        let mut config = ServerConfig::with_single_cert(
+            vec![CertificateDer::from(cert)],
+            PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key)),
+        )
+        .unwrap();
+        config.transport_config(transport());
+        let socket = std::net::UdpSocket::bind(SocketAddr::from(([127, 0, 0, 1], port))).unwrap();
+        let endpoint = Endpoint::new(quinn::EndpointConfig::new(Arc::new(reset)), Some(config), socket,
+                                     quinn::default_runtime().unwrap()).unwrap();
+        println!("{{\"port\":{}}}", endpoint.local_addr().unwrap().port());
+        // What the service keeps: each client's last order heard, and its connection.
+        let heard: Arc<Mutex<HashMap<u64, u64>>> = Arc::default();
+        let kept: Arc<Mutex<Vec<quinn::Connection>>> = Arc::default();
+        let stored = Arc::new(std::sync::atomic::AtomicU64::new(0));
+        while let Some(incoming) = endpoint.accept().await {
+            let (heard, kept, stored) = (heard.clone(), kept.clone(), stored.clone());
+            tokio::spawn(async move {
+                let Ok(connection) = incoming.await else { return };
+                kept.lock().unwrap().push(connection.clone());
+                while let Ok((mut send, mut recv)) = connection.accept_bi().await {
+                    let Ok(bytes) = recv.read_to_end(1024).await else { return };
+                    if bytes.len() < 16 {
+                        return;
+                    }
+                    let client = u64::from_le_bytes(bytes[..8].try_into().unwrap());
+                    let seq = u64::from_le_bytes(bytes[8..16].try_into().unwrap());
+                    let fresh = {
+                        let mut heard = heard.lock().unwrap();
+                        let last = heard.entry(client).or_insert(0);
+                        let fresh = seq > *last;
+                        *last = (*last).max(seq);
+                        fresh
+                    };
+                    let _ = send.write_all(&[1]).await;
+                    let _ = send.finish();
+                    if fresh {
+                        let n = stored.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
+                        if n % 1000 == 0 {
+                            println!("{{\"stored\":{n}}}");
+                        }
+                    }
+                }
+            });
+        }
+    });
+}
+
+#[derive(Debug)]
+struct Anyone; // the test's own service: its certificate is not checked
+
+impl ServerCertVerifier for Anyone {
+    fn verify_server_cert(&self, _: &CertificateDer<'_>, _: &[CertificateDer<'_>], _: &ServerName<'_>, _: &[u8],
+                          _: UnixTime) -> Result<ServerCertVerified, rustls::Error> {
+        Ok(ServerCertVerified::assertion())
+    }
+    fn verify_tls12_signature(&self, message: &[u8], cert: &CertificateDer<'_>, dss: &DigitallySignedStruct)
+                              -> Result<HandshakeSignatureValid, rustls::Error> {
+        rustls::crypto::verify_tls12_signature(message, cert, dss,
+            &rustls::crypto::ring::default_provider().signature_verification_algorithms)
+    }
+    fn verify_tls13_signature(&self, message: &[u8], cert: &CertificateDer<'_>, dss: &DigitallySignedStruct)
+                              -> Result<HandshakeSignatureValid, rustls::Error> {
+        rustls::crypto::verify_tls13_signature(message, cert, dss,
+            &rustls::crypto::ring::default_provider().signature_verification_algorithms)
+    }
+    fn supported_verify_schemes(&self) -> Vec<SignatureScheme> {
+        rustls::crypto::ring::default_provider().signature_verification_algorithms.supported_schemes()
+    }
+}
+
+fn percentile(values: &mut [f64], p: f64) -> f64 {
+    if values.is_empty() {
+        return 0.0;
+    }
+    values.sort_by(|a, b| a.partial_cmp(b).unwrap());
+    values[((values.len() as f64 * p) as usize).min(values.len() - 1)]
+}
+
+fn client_config() -> ClientConfig {
+    let crypto = rustls::ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
+        .with_protocol_versions(&[&rustls::version::TLS13])
+        .unwrap()
+        .dangerous()
+        .with_custom_certificate_verifier(Arc::new(Anyone))
+        .with_no_client_auth();
+    let mut config = ClientConfig::new(Arc::new(QuicClientConfig::try_from(crypto).unwrap()));
+    config.transport_config(transport());
+    config
+}
+
+async fn exchange(connection: &quinn::Connection, client: u64, seq: u64) -> bool {
+    let Ok((mut send, mut recv)) = connection.open_bi().await else { return false };
+    if send.write_all(&order(client, seq)).await.is_err() || send.finish().is_err() {
+        return false;
+    }
+    matches!(recv.read_to_end(16).await, Ok(answer) if answer == [1])
+}
+
+fn connect(port: u16, first: u64, count: u64, window: usize) {
+    let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
+    runtime.block_on(async move {
+        let mut endpoint = Endpoint::client(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();
+        endpoint.set_default_client_config(client_config());
+        let server = SocketAddr::from(([127, 0, 0, 1], port));
+        let window = Arc::new(tokio::sync::Semaphore::new(window));
+        let kept: Arc<Mutex<Vec<quinn::Connection>>> = Arc::default();
+        let waited: Arc<Mutex<Vec<f64>>> = Arc::default();
+        let began = Instant::now();
+        let mut tasks = Vec::new();
+        for client in first..first + count {
+            let permit = window.clone().acquire_owned().await.unwrap();
+            let (endpoint, kept, waited) = (endpoint.clone(), kept.clone(), waited.clone());
+            tasks.push(tokio::spawn(async move {
+                let started = Instant::now();
+                let Ok(connecting) = endpoint.connect(server, "localhost") else { return false };
+                let Ok(connection) = connecting.await else { return false };
+                let ok = exchange(&connection, client, 1).await;
+                waited.lock().unwrap().push(started.elapsed().as_secs_f64() * 1000.0);
+                kept.lock().unwrap().push(connection);
+                drop(permit);
+                ok
+            }));
+        }
+        let mut failed = 0;
+        for task in tasks {
+            if !task.await.unwrap_or(false) {
+                failed += 1;
+            }
+        }
+        let seconds = began.elapsed().as_secs_f64();
+        let mut w = waited.lock().unwrap().clone();
+        println!("{{\"clients\":{count},\"failed\":{failed},\"seconds\":{seconds:.3},\"per_second\":{:.0},\"stored_ms_p50\":{:.3},\"stored_ms_p99\":{:.3}}}",
+                 count as f64 / seconds, percentile(&mut w, 0.5), percentile(&mut w, 0.99));
+        // At rest until stdin closes; then one more order each, over the same connections.
+        let mut rest = Vec::new();
+        let _ = tokio::io::stdin().read_to_end(&mut rest).await;
+        let connections = kept.lock().unwrap().clone();
+        let began = Instant::now();
+        let mut again = 0;
+        for (k, connection) in connections.iter().enumerate() {
+            if exchange(connection, first + k as u64, 2).await {
+                again += 1;
+            }
+        }
+        println!("{{\"again\":{again},\"seconds\":{:.3}}}", began.elapsed().as_secs_f64());
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    });
+}
+
+/// Clients that keep sending, connecting again when their connection fails.
+fn steady(port: u16, first: u64, count: u64, period: Duration, seconds: u64, window: usize) {
+    let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
+    runtime.block_on(async move {
+        let mut endpoint = Endpoint::client(SocketAddr::from(([127, 0, 0, 1], 0))).unwrap();
+        endpoint.set_default_client_config(client_config());
+        let server = SocketAddr::from(([127, 0, 0, 1], port));
+        let connecting = Arc::new(tokio::sync::Semaphore::new(window));
+        let began = Instant::now();
+        let end = began + Duration::from_secs(seconds) + Duration::from_secs(2);
+        let buckets = ((seconds + 70) * 10) as usize;
+        let timeline: Arc<Vec<std::sync::atomic::AtomicU32>> =
+            Arc::new((0..buckets).map(|_| std::sync::atomic::AtomicU32::new(0)).collect());
+        let waited: Arc<Mutex<Vec<f64>>> = Arc::default();
+        let (made, again) = (Arc::new(std::sync::atomic::AtomicU64::new(0)), Arc::new(std::sync::atomic::AtomicU64::new(0)));
+        let mut tasks = Vec::new();
+        for (k, client) in (first..first + count).enumerate() {
+            let (endpoint, connecting, timeline, waited, made, again) =
+                (endpoint.clone(), connecting.clone(), timeline.clone(), waited.clone(), made.clone(), again.clone());
+            let start = began + Duration::from_secs(2) + period.mul_f64(k as f64 / count as f64);
+            tasks.push(tokio::spawn(async move {
+                let mut connection: Option<quinn::Connection> = None;
+                let mut seq = 0u64;
+                let mut next = start;
+                loop {
+                    if connection.is_none() {
+                        let permit = connecting.acquire().await.unwrap();
+                        if let Ok(c) = endpoint.connect(server, "localhost") {
+                            if let Ok(Ok(c)) = tokio::time::timeout(Duration::from_secs(2), c).await {
+                                connection = Some(c);
+                            }
+                        }
+                        drop(permit);
+                        if connection.is_none() {
+                            tokio::time::sleep(Duration::from_millis(200)).await;
+                            continue;
+                        }
+                    }
+                    if Instant::now() >= end {
+                        return;
+                    }
+                    tokio::time::sleep_until(next.into()).await;
+                    seq += 1;
+                    made.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                    let order_made = Instant::now();
+                    next += period;
+                    loop {
+                        let c = connection.clone().unwrap();
+                        let done = tokio::time::timeout(Duration::from_secs(1), exchange(&c, client, seq)).await;
+                        if matches!(done, Ok(true)) {
+                            let now = Instant::now();
+                            let bucket = ((now - began).as_millis() / 100) as usize;
+                            if bucket < timeline.len() {
+                                timeline[bucket].fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                            }
+                            waited.lock().unwrap().push((now - order_made).as_secs_f64() * 1000.0);
+                            break;
+                        }
+                        // Lost: a new connection (a handshake), and the order again.
+                        again.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+                        connection = None;
+                        while connection.is_none() {
+                            let permit = connecting.acquire().await.unwrap();
+                            if let Ok(c) = endpoint.connect(server, "localhost") {
+                                if let Ok(Ok(c)) = tokio::time::timeout(Duration::from_secs(2), c).await {
+                                    connection = Some(c);
+                                }
+                            }
+                            drop(permit);
+                            if connection.is_none() {
+                                tokio::time::sleep(Duration::from_millis(200)).await;
+                            }
+                        }
+                    }
+                    if Instant::now() >= end {
+                        return;
+                    }
+                }
+            }));
+        }
+        for task in tasks {
+            let _ = task.await;
+        }
+        let mut w = waited.lock().unwrap().clone();
+        let stored = w.len();
+        let timeline: Vec<u32> = timeline.iter().map(|b| b.load(std::sync::atomic::Ordering::Relaxed)).collect();
+        let last = timeline.iter().rposition(|&b| b > 0).unwrap_or(0);
+        let (mut longest, mut gap, mut since) = (0usize, 0usize, None::<usize>);
+        for (b, &n) in timeline.iter().enumerate().take(last + 1) {
+            if n > 0 {
+                if let Some(s) = since {
+                    if b - s - 1 > longest {
+                        longest = b - s - 1;
+                        gap = s + 1;
+                    }
+                }
+                since = Some(b);
+            }
+        }
+        let series: Vec<String> = timeline[..=last].iter().map(|n| n.to_string()).collect();
+        println!("{{\"clients\":{count},\"made\":{},\"stored\":{stored},\"again\":{},\"stored_ms_p50\":{:.3},\"stored_ms_p99\":{:.3},\"stored_ms_max\":{:.3},\"longest_gap_ms\":{},\"gap_from_ms\":{},\"timeline_100ms\":[{}]}}",
+                 made.load(std::sync::atomic::Ordering::Relaxed), again.load(std::sync::atomic::Ordering::Relaxed),
+                 percentile(&mut w, 0.5), percentile(&mut w, 0.99), percentile(&mut w, 1.0), longest * 100, gap * 100,
+                 series.join(","));
+    });
+}
+
+fn main() {
+    let args: Vec<String> = std::env::args().collect();
+    match args.get(1).map(String::as_str) {
+        Some("serve") if args.len() == 4 || args.len() == 5 => serve(std::fs::read(&args[2]).unwrap(),
+                                                                    std::fs::read(&args[3]).unwrap(),
+                                                                    args.get(4).map_or(0, |p| p.parse().unwrap())),
+        Some("steady") if args.len() == 8 => steady(args[2].parse().unwrap(), args[3].parse().unwrap(),
+                                                    args[4].parse().unwrap(),
+                                                    Duration::from_millis(args[5].parse().unwrap()),
+                                                    args[6].parse().unwrap(), args[7].parse().unwrap()),
+        Some("connect") if args.len() == 6 => connect(args[2].parse().unwrap(), args[3].parse().unwrap(),
+                                                      args[4].parse().unwrap(), args[5].parse().unwrap()),
+        _ => {
+            eprintln!("idle-quic serve CERT.der KEY.der [PORT] | idle-quic connect PORT FIRST COUNT WINDOW | \
+                       idle-quic steady PORT FIRST COUNT PERIOD_MS SECONDS WINDOW");
+            std::process::exit(2);
+        }
+    }
+}