refactors

Cargo.lock

@@ -17,6 +17,26 @@ version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f26201604c87b1e01bd3d98f8d5d9a8fcbb815e8cedb41ffccbeb4bf593a35fe"
 
+[[package]]
+name = "ahash"
+version = "0.8.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2c99f64d1e06488f620f932677e24bc6e2897582980441ae90a671415bd7ec2f"
+dependencies = [
+ "cfg-if",
+ "once_cell",
+ "version_check",
+]
+
+[[package]]
+name = "aho-corasick"
+version = "0.7.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc936419f96fa211c1b9166887b38e5e40b19958e5b895be7c1f93adec7071ac"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "anyhow"
 version = "1.0.70"
@@ -134,6 +154,12 @@ version = "1.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
+[[package]]
+name = "bumpalo"
+version = "3.13.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a3e2c3daef883ecc1b5d58c15adae93470a91d425f3532ba1695849656af3fc1"
+
 [[package]]
 name = "bytemuck"
 version = "1.13.1"
@@ -179,14 +205,18 @@ dependencies = [
  "bytemuck",
  "color-eyre",
  "console-subscriber",
+ "css-minify",
  "eyre",
  "futures",
  "hyper",
+ "minify-html",
+ "minify-js 0.5.6",
  "once_cell",
  "serde",
  "serde_json",
  "time",
  "tokio",
+ "tokio-stream",
  "tracing",
  "tracing-error",
  "tracing-subscriber",
@@ -262,6 +292,12 @@ dependencies = [
  "tracing-subscriber",
 ]
 
+[[package]]
+name = "convert_case"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6245d59a3e82a7fc217c5828a6692dbc6dfb63a0c8c90495621f7b9d79704a0e"
+
 [[package]]
 name = "crc32fast"
 version = "1.3.2"
@@ -290,6 +326,30 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "css-minify"
+version = "0.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "874c6e2d19f8d4a285083b11a3241bfbe01ac3ed85f26e1e6b34888d960552bd"
+dependencies = [
+ "derive_more",
+ "indexmap",
+ "nom",
+]
+
+[[package]]
+name = "derive_more"
+version = "0.99.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4fb810d30a7c1953f91334de7244731fc3f3c10d7fe163338a35b9f640960321"
+dependencies = [
+ "convert_case",
+ "proc-macro2",
+ "quote",
+ "rustc_version",
+ "syn 1.0.109",
+]
+
 [[package]]
 name = "either"
 version = "1.8.1"
@@ -429,6 +489,16 @@ version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a9ee70c43aaf417c914396645a0fa852624801b24ebb7ae78fe8272889ac888"
 
+[[package]]
+name = "hashbrown"
+version = "0.13.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "43a3c133739dddd0d2990f9a4bdf8eb4b21ef50e4851ca85ab661199821d510e"
+dependencies = [
+ "ahash",
+ "bumpalo",
+]
+
 [[package]]
 name = "hdrhistogram"
 version = "7.5.2"
@@ -540,7 +610,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "1885e79c1fc4b10f0e172c475f458b7f7b93061064d98c3293e98c5ba0c8b399"
 dependencies = [
  "autocfg",
- "hashbrown",
+ "hashbrown 0.12.3",
 ]
 
 [[package]]
@@ -606,6 +676,40 @@ version = "0.3.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2a60c7ce501c71e03a9c9c0d35b861413ae925bd979cc7a4e30d060069aaac8d"
 
+[[package]]
+name = "minify-html"
+version = "0.11.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc4d9147754a49e80557df835eb59e743eab1bf75410a134f55dc4b9dbb692ad"
+dependencies = [
+ "aho-corasick",
+ "css-minify",
+ "lazy_static",
+ "memchr",
+ "minify-js 0.4.3",
+ "rustc-hash",
+]
+
+[[package]]
+name = "minify-js"
+version = "0.4.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c300f90ba1138b5c5daf5d9441dc9bdc67b808aac22cf638362a2647bc213be4"
+dependencies = [
+ "lazy_static",
+ "parse-js 0.10.3",
+]
+
+[[package]]
+name = "minify-js"
+version = "0.5.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "22d6c512a82abddbbc13b70609cb2beff01be2c7afff534d6e5e1c85e438fc8b"
+dependencies = [
+ "lazy_static",
+ "parse-js 0.17.0",
+]
+
 [[package]]
 name = "minimal-lexical"
 version = "0.2.1"
@@ -708,6 +812,30 @@ version = "3.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c1b04fb49957986fdce4d6ee7a65027d55d4b6d2265e5848bbb507b58ccfdb6f"
 
+[[package]]
+name = "parse-js"
+version = "0.10.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "30534759e6ad87aa144c396544747e1c25b1020bd133356fd758c8facec764e5"
+dependencies = [
+ "aho-corasick",
+ "lazy_static",
+ "memchr",
+]
+
+[[package]]
+name = "parse-js"
+version = "0.17.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9ec3b11d443640ec35165ee8f6f0559f1c6f41878d70330fe9187012b5935f02"
+dependencies = [
+ "aho-corasick",
+ "bumpalo",
+ "hashbrown 0.13.2",
+ "lazy_static",
+ "memchr",
+]
+
 [[package]]
 name = "percent-encoding"
 version = "2.2.0"
@@ -862,6 +990,21 @@ version = "0.1.21"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7ef03e0a2b150c7a90d01faf6254c9c48a41e95fb2a8c2ac1c6f0d2b9aefc342"
 
+[[package]]
+name = "rustc-hash"
+version = "1.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2"
+
+[[package]]
+name = "rustc_version"
+version = "0.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bfa0f585226d2e68097d4f95d113b15b83a82e819ab25717ec0590d9584ef366"
+dependencies = [
+ "semver",
+]
+
 [[package]]
 name = "rustversion"
 version = "1.0.12"
@@ -874,6 +1017,12 @@ version = "1.0.13"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f91339c0467de62360649f8d3e185ca8de4224ff281f66000de5eb2a77a79041"
 
+[[package]]
+name = "semver"
+version = "1.0.17"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bebd363326d05ec3e2f532ab7660680f3b02130d780c299bca73469d521bc0ed"
+
 [[package]]
 name = "serde"
 version = "1.0.157"
@@ -1048,13 +1197,14 @@ dependencies = [
 
 [[package]]
 name = "tokio-stream"
-version = "0.1.12"
+version = "0.1.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8fb52b74f05dbf495a8fba459fdc331812b96aa086d9eb78101fa0d4569c3313"
+checksum = "397c988d37662c7dda6d2208364a706264bf3d6138b11d436cbac0ad38832842"
 dependencies = [
  "futures-core",
  "pin-project-lite",
  "tokio",
+ "tokio-util",
 ]
 
 [[package]]
@@ -1236,6 +1386,12 @@ version = "0.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "830b7e5d4d90034032940e4ace0d9a9a057e7a45cd94e6c007832e39edb82f6d"
 
+[[package]]
+name = "version_check"
+version = "0.9.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "49874b5167b65d7193b8aba1567f5c7d93d001cafc34600cee003eda787e483f"
+
 [[package]]
 name = "want"
 version = "0.3.0"

Cargo.toml

@@ -6,9 +6,6 @@ edition = "2021"
 [profile.release]
 debug = true
 
-[profile.dev.package.backtrace]
-opt-level = 3
-
 
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
@@ -18,7 +15,7 @@ time = { version = "0.3.20", features = ["local-offset", "macros"] }
 bytemuck = { version = "1.13.0", features = ["derive"] }
 serde = { version = "1.0.152", features = ["derive"] }
 serde_json = "1.0.91"
-hyper = { version = "0.14.24", optional = true, features = ["server", "http1", "tcp"] }
+hyper = { version = "0.14.24", optional = true, features = ["server", "http1", "tcp", "stream"] }
 futures = { version = "0.3.27", default-features = false, features = ["std"] }
 tracing = "0.1.37"
 tracing-subscriber = { version = "0.3.16", features = ["time"] }
@@ -28,8 +25,15 @@ eyre = "0.6.8"
 color-eyre = "0.6.2"
 tracing-error = "0.2.0"
 zerocopy = "0.6.1"
+tokio-stream = { version = "0.1.14", features = ["sync"] }
+minify-html = "0.11.1"
+
+[build-dependencies]
+minify-js = { version = "0.5.6", optional = true }
+minify-html = { version = "0.11.1", optional = true }
+css-minify = { version = "0.3.1", optional = true }
 
 [features]
 default = ["debug_server"]
-debug_server = ["dep:hyper"]
+debug_server = ["dep:hyper", "minify-html", "dep:minify-js", "dep:css-minify"]
 tokio_console = ["dep:console-subscriber"]

build.rs

@@ -0,0 +1,45 @@
+fn main() {
+    #[cfg(feature = "debug_server")]
+    pack_debug_page().unwrap();
+
+    println!("cargo:rerun-if-changed=main.js");
+    println!("cargo:rerun-if-changed=index.html");
+    println!("cargo:rerun-if-changed=main.css");
+}
+
+#[cfg(feature = "debug_server")]
+fn pack_debug_page() -> Result<(), Box<dyn std::error::Error>> {
+    use std::io::Write;
+
+    use css_minify::optimizations::{Level, Minifier};
+
+    let js = std::fs::read_to_string("web/main.js").unwrap();
+    let html = std::fs::read_to_string("web/index.html").unwrap();
+    let css = std::fs::read_to_string("web/main.css").unwrap();
+
+    let mut out = Vec::new();
+    minify_js::minify(
+        &minify_js::Session::new(),
+        minify_js::TopLevelMode::Global,
+        js.as_bytes(),
+        &mut out,
+    )
+    .unwrap();
+    let js = std::str::from_utf8(&out)?;
+
+    let css = Minifier::default().minify(&css, Level::Three).unwrap();
+
+    let (head, body) = html
+        .split_once("<!--INSERT HEAD CONTENT HERE-->")
+        .expect("did not find split point in html");
+
+    let html = minify_html::minify(
+        format!("{head}<style>{css}</style><script>{js}</script>{body}").as_bytes(),
+        &minify_html::Cfg::spec_compliant(),
+    );
+
+    std::fs::File::create(std::env::var("OUT_DIR").unwrap() + "/minified.html")?
+        .write_all(&html)?;
+
+    Ok(())
+}

src/auth.rs

@@ -64,19 +64,13 @@ pub async fn dyn_ip_update(
             })?
             .into()),
         PacketKind::Error => {
-            let first_zero = packet
-                .data
-                .iter()
-                .enumerate()
-                .find_map(|(i, x)| (*x == 0).then_some(i));
-
             return Err(eyre!(
                 "{}",
-                std::str::from_utf8(first_zero.map_or(&packet.data, |i| &packet.data[..i]),)?
+                packet.as_string().unwrap_or("unknown dyn auth error")
             ));
         }
 
-        _ => return Err(eyre!("server returned unexpected packet")),
+        _ => return Err(eyre!("auth server returned unexpected packet")),
     };
 
     debug!(?result, "finished dyn ip update");

src/client.rs

@@ -15,7 +15,7 @@ use tracing::{info, instrument, trace};
 use crate::{
     auth::dyn_ip_update,
     constants::{AUTH_TIMEOUT, CALL_ACK_TIMEOUT, CALL_TIMEOUT, PING_TIMEOUT, SEND_PING_INTERVAL},
-    debug_server::peer_query,
+    http::peer_query,
     packets::{Header, Packet, PacketKind, RemConnect, REJECT_OOP, REJECT_TIMEOUT},
     ports::{PortHandler, PortStatus},
     Config, HandlerMetadata,

src/constants.rs

@@ -9,3 +9,5 @@ pub const PING_TIMEOUT: Duration = Duration::from_secs(30);
 pub const SEND_PING_INTERVAL: Duration = Duration::from_secs(20);
 
 pub const CACHE_STORE_INTERVAL: Duration = Duration::from_secs(5);
+
+pub const DEBUG_SERVER_PING_INTERVAL: Duration = Duration::from_secs(5);

src/debug_server.rs

@@ -1,119 +0,0 @@
-use futures::Future;
-use hyper::rt::Executor;
-use hyper::service::{make_service_fn, service_fn};
-use hyper::{Body, Response, Server};
-use std::convert::Infallible;
-use std::net::SocketAddr;
-use std::sync::Arc;
-use tokio::sync::Mutex;
-use tracing::error;
-use zerocopy::{AsBytes, FromBytes, LittleEndian, Unaligned};
-
-use tracing::{debug, instrument};
-
-use crate::packets::{Header, Packet};
-
-use crate::ports::PortHandler;
-use crate::spawn;
-
-#[derive(Clone)]
-struct NamedExecutor;
-impl<T: Send + 'static, Fut: Future<Output = T> + Send + 'static> Executor<Fut> for NamedExecutor {
-    fn execute(&self, fut: Fut) {
-        spawn("http worker", fut);
-    }
-}
-
-pub async fn debug_server(addr: SocketAddr, port_handler: Arc<Mutex<PortHandler>>) {
-    let server = Server::bind(&addr)
-        .executor(NamedExecutor)
-        .serve(make_service_fn(move |_conn| {
-            let port_handler = port_handler.clone();
-            async move {
-                Ok::<_, Infallible>(service_fn(move |_req| {
-                    let port_handler = port_handler.clone();
-                    async move {
-                        Ok::<_, Infallible>(Response::new(Body::from(
-                            port_handler.lock().await.status_string(),
-                        )))
-                    }
-                }))
-            }
-        }));
-
-    // Run this server for... forever!
-    if let Err(error) = server.await {
-        error!(%error, "debug server error");
-    }
-}
-
-type U16 = zerocopy::U16<LittleEndian>;
-type U32 = zerocopy::U32<LittleEndian>;
-
-#[derive(AsBytes)]
-#[repr(transparent)]
-#[allow(dead_code)]
-struct PeerQuery {
-    number: U32,
-}
-
-#[derive(FromBytes, Unaligned, Debug)]
-#[repr(packed)]
-#[allow(dead_code)]
-struct PeerReply {
-    number: U32,
-    name: [u8; 40],
-    flags: U16,
-    kind: u8,
-    hostname: [u8; 40],
-    ipaddress: [u8; 4],
-    port: U16,
-    extension: u8,
-    pin: U16,
-    timestamp: U32,
-}
-
-#[instrument]
-pub async fn peer_query(server: &SocketAddr, number: u32) -> eyre::Result<Option<String>> {
-    debug!(%number, "looking up");
-
-    let mut packet = Packet {
-        header: Header {
-            kind: 3, // Peer Query
-            length: 4,
-        },
-        data: Vec::new(),
-    };
-
-    packet.data.clear();
-    packet.data.resize(packet.header.length as usize, 0);
-
-    PeerQuery {
-        number: number.into(),
-    }
-    .write_to(packet.data.as_mut_slice())
-    .unwrap();
-
-    let mut socket = tokio::net::TcpStream::connect(server).await?;
-
-    let (mut reader, mut writer) = socket.split();
-
-    packet.send(&mut writer).await?;
-    packet.recv_into(&mut reader).await?;
-
-    Ok(if packet.kind().raw() == 5 {
-        // PeerReply
-        PeerReply::read_from(packet.data.as_slice()).and_then(|x| {
-            let i = x
-                .name
-                .iter()
-                .enumerate()
-                .find_map(|(i, c)| (*c == 0).then_some(i))
-                .unwrap_or(x.name.len());
-
-            Some(std::str::from_utf8(&x.name[..i]).ok()?.to_owned())
-        })
-    } else {
-        None
-    })
-}

src/http.rs

@@ -0,0 +1,175 @@
+use futures::Future;
+use tokio_stream::StreamExt;
+
+use hyper::rt::Executor;
+use hyper::service::{make_service_fn, service_fn};
+use hyper::{Body, Method, Response, Server, StatusCode};
+use std::convert::Infallible;
+use std::net::SocketAddr;
+use std::sync::Arc;
+use std::time::{SystemTime, UNIX_EPOCH};
+use tokio::sync::Mutex;
+use tokio_stream::wrappers::{IntervalStream, WatchStream};
+use tracing::error;
+
+use zerocopy::{AsBytes, FromBytes, LittleEndian, Unaligned};
+
+use tracing::{debug, instrument};
+
+use crate::constants::DEBUG_SERVER_PING_INTERVAL;
+use crate::packets::{Header, Packet};
+
+use crate::ports::PortHandler;
+use crate::spawn;
+
+#[derive(Clone)]
+struct NamedExecutor;
+impl<T: Send + 'static, Fut: Future<Output = T> + Send + 'static> Executor<Fut> for NamedExecutor {
+    fn execute(&self, fut: Fut) {
+        spawn("http worker", fut);
+    }
+}
+
+pub async fn debug_server(
+    addr: SocketAddr,
+    port_handler: Arc<Mutex<PortHandler>>,
+    change_receiver: tokio::sync::watch::Receiver<std::time::Instant>,
+) {
+    let server = Server::bind(&addr)
+        .executor(NamedExecutor)
+        .serve(make_service_fn(move |_conn| {
+            let port_handler = port_handler.clone();
+            let change_receiver = change_receiver.clone();
+            async move {
+                Ok::<_, Infallible>(service_fn(move |req| {
+                    let port_handler = port_handler.clone();
+                    let change_receiver = WatchStream::new(change_receiver.clone());
+                    async move {
+                        match (req.method(), req.uri().path()) {
+                            (&Method::GET, "/") => Ok(Response::new(Body::from(include_str!(
+                                concat!(env!("OUT_DIR"), "/minified.html")
+                            )))),
+
+                            (&Method::GET, "/data") => {
+                                let res = Response::builder().header("Cache-Control", "no-store");
+
+                                match serde_json::to_string(&*port_handler.lock().await) {
+                                    Ok(data) => res
+                                        .header("Content-Type", "application/json")
+                                        .body(Body::from(data)),
+                                    Err(err) => {
+                                        error!(%err, "failed to serialize data for debug server");
+                                        res.status(StatusCode::INTERNAL_SERVER_ERROR)
+                                            .body(Body::from(""))
+                                    }
+                                }
+                            }
+
+                            (&Method::GET, "/events") => Response::builder()
+                                .status(StatusCode::OK)
+                                .header("Cache-Control", "no-store")
+                                .header("Content-Type", "text/event-stream")
+                                .body(Body::wrap_stream({
+                                    change_receiver
+                                        .map(|x| ("change", x))
+                                        .merge(
+                                            IntervalStream::new(tokio::time::interval(
+                                                DEBUG_SERVER_PING_INTERVAL,
+                                            ))
+                                            .map(|x| ("ping", x.into_std())),
+                                        )
+                                        .filter_map(|(kind, time)| {
+                                            let timestamp = (SystemTime::now() + time.elapsed())
+                                                .duration_since(UNIX_EPOCH)
+                                                .ok()?
+                                                .as_secs();
+
+                                            Some(Ok::<_, Infallible>(format!(
+                                                "event:{kind}\ndata: {timestamp}\n\n"
+                                            )))
+                                        })
+                                })),
+                            _ => Response::builder()
+                                .status(StatusCode::NOT_FOUND)
+                                .body(Body::empty()),
+                        }
+                    }
+                }))
+            }
+        }));
+
+    if let Err(error) = server.await {
+        error!(%error, "debug server error");
+    }
+}
+
+type U16 = zerocopy::U16<LittleEndian>;
+type U32 = zerocopy::U32<LittleEndian>;
+
+#[derive(AsBytes)]
+#[repr(transparent)]
+#[allow(dead_code)]
+struct PeerQuery {
+    number: U32,
+}
+
+#[derive(FromBytes, Unaligned, Debug)]
+#[repr(packed)]
+#[allow(dead_code)]
+struct PeerReply {
+    number: U32,
+    name: [u8; 40],
+    flags: U16,
+    kind: u8,
+    hostname: [u8; 40],
+    ipaddress: [u8; 4],
+    port: U16,
+    extension: u8,
+    pin: U16,
+    timestamp: U32,
+}
+
+#[instrument]
+pub async fn peer_query(server: &SocketAddr, number: u32) -> eyre::Result<Option<String>> {
+    debug!(%number, "looking up");
+
+    let mut packet = Packet {
+        header: Header {
+            kind: 3, // Peer Query
+            length: 4,
+        },
+        data: Vec::new(),
+    };
+
+    packet.data.clear();
+    packet.data.resize(packet.header.length as usize, 0);
+
+    PeerQuery {
+        number: number.into(),
+    }
+    .write_to(packet.data.as_mut_slice())
+    .unwrap();
+
+    let mut socket = tokio::net::TcpStream::connect(server).await?;
+
+    let (mut reader, mut writer) = socket.split();
+
+    packet.send(&mut writer).await?;
+    packet.recv_into(&mut reader).await?;
+
+    Ok(if packet.kind().raw() == 5 {
+        // PeerReply
+        PeerReply::read_from(packet.data.as_slice()).and_then(|x| {
+            let i = x
+                .name
+                .iter()
+                .enumerate()
+                .find_map(|(i, c)| (*c == 0).then_some(i))
+                .unwrap_or(x.name.len());
+
+            Some(std::str::from_utf8(&x.name[..i]).ok()?.to_owned())
+        })
+    } else {
+        None
+    })
+}

src/main.rs

@@ -1,5 +1,4 @@
 #![warn(clippy::pedantic)]
-// #![allow(clippy::missing_errors_doc)]
 
 use std::{
     fmt::Debug,
@@ -11,8 +10,8 @@ use std::{
     time::Duration,
 };
 
-use debug_server::debug_server;
 use futures::Future;
+use http::debug_server;
 use packets::{Header, Packet};
 use serde::{Deserialize, Deserializer};
 use time::format_description::OwnedFormatItem;
@@ -20,9 +19,9 @@ use tokio::{
     io::AsyncWriteExt,
     net::{TcpListener, TcpStream},
     sync::Mutex,
-    time::{sleep, Instant},
+    time::sleep,
 };
-use tracing::{error, info, instrument, warn, Level};
+use tracing::{debug, error, info, instrument, warn, Level};
 use tracing_subscriber::fmt::time::FormatTime;
 
 use crate::packets::PacketKind;
@@ -32,7 +31,7 @@ pub mod auth;
 pub mod client;
 pub mod constants;
 #[cfg(feature = "debug_server")]
-pub mod debug_server;
+pub mod http;
 pub mod packets;
 pub mod ports;
 
@@ -185,7 +184,7 @@ fn setup_tracing(config: &Config) {
                 Level::ERROR => write!(writer, " {:>5} ", level.red())?,
             }
 
-            write!(writer, "{:23}{}", meta.target().dimmed(), ":".bold())?;
+            write!(writer, "{:18}{}", meta.target().dimmed(), ":".bold())?;
 
             /*
             if let Some(filename) = meta.file() {
@@ -248,7 +247,14 @@ async fn connection_handler(
 
     let error = match res {
         Err(_) => Some("internal server error".to_owned()),
-        Ok(Err(err)) => Some(err.to_string()),
+        Ok(Err(err)) => match err.downcast_ref::<std::io::Error>() {
+            Some(io_error) if io_error.kind() == std::io::ErrorKind::UnexpectedEof => {
+                // don't print an error on dropped connections
+                debug!(%addr, "Client dropped their connection");
+                None
+            }
+            _ => Some(err.to_string()),
+        },
         Ok(Ok(())) => None,
     };
 
@@ -265,6 +271,7 @@ async fn connection_handler(
             length: packet.data.len().try_into().unwrap(), // this will never fail, as we just truncated the vector
         };
 
+        // Attempt to notify the client of the failure
         let (_, mut writer) = stream.split();
         _ = packet.send(&mut writer).await;
     }
@@ -323,7 +330,7 @@ async fn tokio_main(config: Arc<Config>) -> eyre::Result<()> {
 
     let cache_path = PathBuf::from("cache.json");
 
-    let (change_sender, change_receiver) = tokio::sync::watch::channel(Instant::now());
+    let (change_sender, change_receiver) = tokio::sync::watch::channel(std::time::Instant::now());
 
     let mut port_handler = PortHandler::load_or_default(&cache_path, change_sender);
     port_handler.update_allowed_ports(&config.allowed_ports);
@@ -332,7 +339,7 @@ async fn tokio_main(config: Arc<Config>) -> eyre::Result<()> {
 
     spawn(
         "cache daemon",
-        cache_daemon(port_handler.clone(), cache_path, change_receiver),
+        cache_daemon(port_handler.clone(), cache_path, change_receiver.clone()),
     );
 
     #[cfg(feature = "debug_server")]
@@ -340,7 +347,7 @@ async fn tokio_main(config: Arc<Config>) -> eyre::Result<()> {
         warn!(%listen_addr, "debug server listening");
         spawn(
             "debug server",
-            debug_server(listen_addr, port_handler.clone()),
+            debug_server(listen_addr, port_handler.clone(), change_receiver),
         );
     }
 

src/packets.rs

@@ -1,7 +1,8 @@
-use std::{ffi::CStr, fmt::Debug};
+use std::fmt::Debug;
 
 use bytemuck::{Pod, Zeroable};
 use eyre::eyre;
+use serde::Serialize;
 use tokio::{
     io::{AsyncReadExt, AsyncWriteExt},
     net::tcp::{ReadHalf, WriteHalf},
@@ -68,32 +69,53 @@ impl PacketKind {
     }
 }
 
-#[derive(Default, Clone, Copy, Pod, Zeroable)]
+#[derive(Serialize, Default, Clone, Copy, Pod, Zeroable)]
 #[repr(C)]
 pub struct Header {
     pub kind: u8,
     pub length: u8,
 }
 
-#[derive(Default, Clone)]
+#[derive(Serialize, Default, Clone)]
 pub struct Packet {
     pub header: Header,
     pub data: Vec<u8>,
 }
 
+impl Packet {
+    #[must_use]
+    pub fn data(&self) -> &[u8] {
+        &self.data[..self.header.length as usize]
+    }
+
+    #[must_use]
+    pub fn as_string(&self) -> Option<&str> {
+        let data = self.data();
+        let nul = data.iter().enumerate().find(|(_i, c)| **c == 0);
+
+        let data = if let Some((i, _)) = nul {
+            &data[..i]
+        } else {
+            data
+        };
+
+        std::str::from_utf8(data).ok()
+    }
+}
+
 impl Debug for Packet {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let data = &self.data;
-
         let mut debugger = f.debug_struct("Packet");
 
         debugger.field("kind", &PacketKind::from_u8(self.header.kind));
 
-        let c_str = CStr::from_bytes_until_nul(data).ok();
+        match self.as_string() {
-        if let Some(str_data) = c_str.as_ref().and_then(|x| x.to_str().ok()) {
+            Some(string) if string.chars().all(|c| !c.is_control()) => {
-            debugger.field("data", &str_data);
+                debugger.field("data", &string);
-        } else {
+            }
-            debugger.field("data", &data);
+            _ => {
+                debugger.field("data", &self.data());
+            }
         }
 
         debugger.finish()
@@ -131,7 +153,7 @@ impl Packet {
     pub async fn recv_into_cancelation_safe(
         &mut self,
         stream: &mut ReadHalf<'_>,
-    ) -> std::io::Result<()> {
+    ) -> eyre::Result<()> {
         // Makes sure all data is available before reading
         let header_bytes = bytemuck::bytes_of_mut(&mut self.header);
         stream.peek(header_bytes).await?;
@@ -143,7 +165,7 @@ impl Packet {
     }
 
     #[allow(clippy::missing_errors_doc)]
-    pub async fn recv_into(&mut self, stream: &mut ReadHalf<'_>) -> std::io::Result<()> {
+    pub async fn recv_into(&mut self, stream: &mut ReadHalf<'_>) -> eyre::Result<()> {
         let header_bytes = bytemuck::bytes_of_mut(&mut self.header);
 
         stream.read_exact(header_bytes).await?;
@@ -152,13 +174,20 @@ impl Packet {
 
         stream.read_exact(&mut self.data).await?;
 
+        if self.header.kind == PacketKind::Error.raw() {
+            return Err(eyre!(
+                "client reported error: {:?}",
+                self.as_string().unwrap_or("unknown dyn auth error")
+            ));
+        }
+
         Ok(())
     }
 
     #[allow(clippy::missing_errors_doc)]
     pub async fn send(&self, stream: &mut WriteHalf<'_>) -> std::io::Result<()> {
         stream.write_all(bytemuck::bytes_of(&self.header)).await?;
-        stream.write_all(&self.data).await?;
+        stream.write_all(self.data()).await?;
         Ok(())
     }
 

src/ports.rs

@@ -1,5 +1,4 @@
 use std::{
-    borrow::Cow,
     collections::{BTreeSet, HashMap, HashSet},
     fmt::{Debug, Display},
     fs::File,
@@ -11,7 +10,7 @@ use std::{
 };
 
 use eyre::eyre;
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Serialize, Serializer};
 use tokio::{
     net::TcpListener,
     sync::{watch::Receiver, Mutex},
@@ -22,25 +21,27 @@ use tracing::{debug, error, info, instrument, warn};
 
 use crate::{
     constants::{CACHE_STORE_INTERVAL, PORT_OWNERSHIP_TIMEOUT, PORT_RETRY_TIME},
-    packets::Packet,
+    packets::{Packet, PacketKind},
-    spawn, Config, Number, Port, UnixTimestamp, TIME_FORMAT, TIME_ZONE_OFFSET,
+    spawn, Config, Number, Port, UnixTimestamp,
 };
 
 #[derive(Default, Serialize, Deserialize)]
 pub struct PortHandler {
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
-    pub last_update: Option<Instant>,
+    #[serde(serialize_with = "serialize_last_update")]
+    pub last_update: Option<std::time::Instant>,
 
     #[serde(skip)]
-    pub change_sender: Option<tokio::sync::watch::Sender<Instant>>,
+    pub change_sender: Option<tokio::sync::watch::Sender<std::time::Instant>>,
 
-    #[serde(skip)]
+    #[serde(skip_deserializing)]
-    port_guards: HashMap<Port, Rejector>,
+    rejectors: HashMap<Port, Rejector>,
 
     allowed_ports: AllowedList,
 
     #[serde(skip)]
     free_ports: HashSet<Port>,
+
     errored_ports: BTreeSet<(UnixTimestamp, Port)>,
     allocated_ports: HashMap<Number, Port>,
 
@@ -51,11 +52,28 @@ pub struct PortHandler {
     pub names: HashMap<Number, String>,
 }
 
+#[allow(clippy::missing_errors_doc)]
+pub fn serialize_last_update<S: Serializer>(
+    last_update: &Option<std::time::Instant>,
+    serializer: S,
+) -> Result<S::Ok, S::Error> {
+    last_update
+        .and_then(|instant| {
+            Some(
+                (SystemTime::now() + instant.elapsed())
+                    .duration_since(UNIX_EPOCH)
+                    .ok()?
+                    .as_secs(),
+            )
+        })
+        .serialize(serializer)
+}
+
 #[instrument(skip(port_handler, change_receiver))]
 pub async fn cache_daemon(
     port_handler: Arc<Mutex<PortHandler>>,
     cache_path: PathBuf,
-    mut change_receiver: Receiver<Instant>,
+    mut change_receiver: Receiver<std::time::Instant>,
 ) {
     let mut last_store = Instant::now() - 2 * CACHE_STORE_INTERVAL;
     let mut change_timeout = None;
@@ -92,129 +110,6 @@ impl<T: Display> Debug for DisplayAsDebug<T> {
     }
 }
 
-fn duration_string(duration: Duration) -> String {
-    let seconds_elapsed = duration.as_secs();
-
-    let days = seconds_elapsed / (60 * 60 * 24);
-    let hours = seconds_elapsed / (60 * 60) % 24;
-    let minutes = (seconds_elapsed / 60) % 60;
-    let seconds = seconds_elapsed % 60;
-
-    match (days > 0, hours > 0, minutes > 0) {
-        (true, _, _) => format!("{days}d {hours}h {minutes}min {seconds}s"),
-        (false, true, _) => format!("{hours}h {minutes}min {seconds}s"),
-        (false, false, true) => format!("{minutes}min {seconds}s"),
-        _ => format!("{duration:.0?}"),
-    }
-}
-
-fn format_instant(instant: Instant) -> String {
-    let when = duration_string(instant.elapsed()) + " ago";
-
-    (|| -> eyre::Result<_> {
-        let timestamp = SystemTime::now().duration_since(UNIX_EPOCH)? - instant.elapsed();
-        let date = time::OffsetDateTime::from_unix_timestamp(
-            timestamp.as_secs().try_into().expect("timestamp overflow"),
-        )?
-        .to_offset(*TIME_ZONE_OFFSET.get().unwrap())
-        .format(TIME_FORMAT.get().unwrap())?;
-
-        Ok(format!("{date} ({when})"))
-    })()
-    .unwrap_or(when)
-}
-
-fn instant_from_timestamp(timestamp: UnixTimestamp) -> Instant {
-    Instant::now() - UNIX_EPOCH.elapsed().unwrap() + Duration::from_secs(timestamp)
-}
-
-#[cfg(feature = "debug_server")]
-impl Debug for PortHandler {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        const SHOW_N_FREE_PORTS: usize = 10;
-
-        let last_update = self
-            .last_update
-            .map(|last_update| Cow::from(format_instant(last_update)))
-            .unwrap_or(Cow::from("?"));
-
-        let mut free_ports = self.free_ports.iter().copied().collect::<Vec<u16>>();
-
-        free_ports.sort_unstable();
-
-        let mut free_ports = free_ports
-            .into_iter()
-            .take(SHOW_N_FREE_PORTS)
-            .map(|x| DisplayAsDebug(x.to_string()))
-            .collect::<Vec<_>>();
-
-        if let Some(n_not_shown) = self.free_ports.len().checked_sub(SHOW_N_FREE_PORTS) {
-            if n_not_shown > 0 {
-                free_ports.push(DisplayAsDebug(format!("[{n_not_shown} more]")));
-            }
-        }
-
-        let errored_ports = self
-            .errored_ports
-            .iter()
-            .rev()
-            .map(|&(since, port)| {
-                DisplayAsDebug(format!(
-                    "{port:5}: {}",
-                    format_instant(instant_from_timestamp(since))
-                ))
-            })
-            .collect::<Vec<_>>();
-
-        let mut allocated_ports = self
-            .allocated_ports
-            .iter()
-            .map(|(&number, &port)| {
-                #[derive(Debug)]
-                #[allow(dead_code)]
-                struct State<'n> {
-                    state: PortStatus,
-                    name: &'n str,
-                    number: u32,
-                    port: u16,
-                    last_change: DisplayAsDebug<String>,
-                }
-
-                let state = &self.port_state[&port];
-
-                State {
-                    state: state.status,
-                    number,
-                    port,
-                    name: self.names.get(&number).map_or("?", |x| x.as_str()),
-                    last_change: DisplayAsDebug(format_instant(instant_from_timestamp(
-                        state.last_change,
-                    ))),
-                }
-            })
-            .collect::<Vec<_>>();
-
-        allocated_ports.sort_unstable_by(|a, b| {
-            a.state.cmp(&b.state).then(
-                self.port_state[&a.port]
-                    .last_change
-                    .cmp(&self.port_state[&b.port].last_change)
-                    .reverse(),
-            )
-        });
-
-        writeln!(f, "last update: {last_update}")?;
-        writeln!(f, "rejectors: {:#?}", self.port_guards)?;
-        writeln!(f, "allowed ports: {:?}", self.allowed_ports.0)?;
-        writeln!(f, "free ports: {free_ports:?}")?;
-
-        writeln!(f, "errored ports: {errored_ports:#?}")?;
-        writeln!(f, "allocated ports: {allocated_ports:#?}")?;
-
-        Ok(())
-    }
-}
-
 #[derive(Default, Serialize, Deserialize)]
 pub struct PortState {
     last_change: UnixTimestamp,
@@ -222,18 +117,6 @@ pub struct PortState {
     status: PortStatus,
 }
 
-impl Debug for PortState {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("PortState")
-            .field(
-                "last_change",
-                &DisplayAsDebug(format_instant(instant_from_timestamp(self.last_change))),
-            )
-            .field("status", &self.status)
-            .finish()
-    }
-}
-
 impl PortState {
     pub fn new_state(&mut self, status: PortStatus) {
         self.last_change = SystemTime::now()
@@ -273,13 +156,8 @@ impl AllowedList {
 }
 
 impl PortHandler {
-    #[must_use]
-    pub fn status_string(&self) -> String {
-        format!("{self:#?}\n")
-    }
-
     pub fn register_update(&mut self) {
-        let now = Instant::now();
+        let now = std::time::Instant::now();
         self.last_update = Some(now);
         self.change_sender
             .as_ref()
@@ -292,9 +170,16 @@ impl PortHandler {
     #[instrument(skip(self))]
     pub fn store(&self, cache: &Path) -> std::io::Result<()> {
         debug!("storing cache");
-        let temp_file = cache.with_extension(".temp");
+        let temp_file = cache.with_extension("temp");
 
-        serde_json::to_writer(BufWriter::new(File::create(&temp_file)?), self)?;
+        let mut value = serde_json::to_value(self)?;
+
+        let value_object = value.as_object_mut().unwrap();
+
+        value_object.remove("rejectors").unwrap();
+        value_object.remove("last_update").unwrap();
+
+        serde_json::to_writer(BufWriter::new(File::create(&temp_file)?), &value)?;
         std::fs::rename(temp_file, cache)?;
 
         Ok(())
@@ -310,7 +195,7 @@ impl PortHandler {
     #[instrument(skip(change_sender))]
     pub fn load_or_default(
         path: &Path,
-        change_sender: tokio::sync::watch::Sender<Instant>,
+        change_sender: tokio::sync::watch::Sender<std::time::Instant>,
     ) -> Self {
         let mut this = Self::load(path).unwrap_or_else(|error| {
             error!(?path, %error, "failed to parse cache file");
@@ -363,7 +248,7 @@ impl PortHandler {
 
         let port_guard = Rejector::start(listener, packet);
 
-        if self.port_guards.insert(port, port_guard).is_some() {
+        if self.rejectors.insert(port, port_guard).is_some() {
             unreachable!("Tried to start rejector that is already running. This should have been impossible since it requires two listeners on the same port.");
         }
     }
@@ -372,7 +257,7 @@ impl PortHandler {
     pub async fn stop_rejector(&mut self, port: Port) -> Option<(TcpListener, Packet)> {
         info!(port, "stopping rejector");
 
-        Some(self.port_guards.remove(&port)?.stop().await)
+        Some(self.rejectors.remove(&port)?.stop().await)
     }
 
     /// # Errors
@@ -400,6 +285,27 @@ struct Rejector {
     handle: JoinHandle<()>,
 }
 
+impl Serialize for Rejector {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: Serializer,
+    {
+        let packet = &self.state.1;
+        let kind = match packet.kind() {
+            PacketKind::End => "end",
+            PacketKind::Reject => "reject",
+            _ => unreachable!(),
+        };
+
+        match packet.as_string() {
+            Some(string) if string.chars().all(|c| !c.is_control()) => {
+                (kind, string).serialize(serializer)
+            }
+            _ => (kind, packet.data()).serialize(serializer),
+        }
+    }
+}
+
 impl Debug for Rejector {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("Rejector")

web/index.html

@@ -0,0 +1,18 @@
+<!DOCTYPE html>
+<html>
+
+<head>
+    <meta charset='utf-8'>
+    <meta name='viewport' content='width=device-width, initial-scale=1'>
+    <title>Centralex State</title>
+    <!--INSERT HEAD CONTENT HERE-->
+</head>
+
+<body>
+
+    <p id="last_update" />
+    <pre id="data"></pre>
+    <list id="list" />
+</body>
+
+</html>

web/main.css

@@ -0,0 +1,3 @@
+body {
+    background-color: #eee;
+}

web/main.js

@@ -0,0 +1,28 @@
+window.onload = () => {
+    const evtSource = new EventSource("/events");
+    const data = document.getElementById("data");
+    const last_update = document.getElementById("last_update");
+    evtSource.addEventListener("change", event => {
+        console.log(event);
+
+        last_update.innerText = `last update at ${new Date(+event.data * 1000)}`;
+
+        const newElement = document.createElement("li");
+        const eventList = document.getElementById("list");
+        newElement.textContent = `change at ${+event.data}`;
+        eventList.appendChild(newElement);
+
+        fetch("/data").then(res => res.json().then(res => data.innerText = JSON.stringify(res, null, 1)));
+    });
+
+    evtSource.addEventListener("ping", event => {
+        console.log(event);
+
+        last_update.innerText = `last update at ${new Date(+event.data * 1000)}`;
+
+        const newElement = document.createElement("li");
+        const eventList = document.getElementById("list");
+        newElement.textContent = `ping at ${+event.data}`;
+        eventList.appendChild(newElement);
+    });
+};