Temporarily retain endpoints which are in active use

Cargo.toml

@@ -39,6 +39,10 @@ categories = ["game-development", "network-programming"]
 edition = "2021"
 exclude = ["docs", "build", "examples", "image"]
 
+[profile.release]
+lto = "fat"
+codegen-units = 1
+
 [[bench]]
 name = "throughput"
 harness = false
@@ -104,6 +108,10 @@ nom = "7.1.3"
 atty = "0.2.14"
 strum = "0.25.0"
 strum_macros = "0.25.2"
+tokio-uring = { version = "0.4.0", features = ["bytes"] }
+async-channel = "1.9.0"
+cfg-if = "1.0.0"
+itertools = "0.11.0"
 
 [target.'cfg(target_os = "linux")'.dependencies]
 sys-info = "0.9.1"

build/build-image/Dockerfile

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM debian:bullseye
+FROM ubuntu:lunar
 
 ARG RUST_TOOLCHAIN
 
@@ -26,8 +26,8 @@ ENV RUSTUP_HOME=/usr/local/rustup \
 RUN set -eux && \
     apt-get update && \
     apt-get install -y lsb-release jq curl wget zip build-essential software-properties-common \
-        libssl-dev pkg-config python3-pip bash-completion g++-x86-64-linux-gnu g++-mingw-w64-x86-64 && \
-    pip3 install live-server && \
+        libssl-dev pkg-config python3-pip pipx bash-completion g++-x86-64-linux-gnu g++-mingw-w64-x86-64 && \
+    pipx install live-server && \
     echo "source /etc/bash_completion" >> /root/.bashrc
 
 # install gcloud

image/Dockerfile

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM gcr.io/distroless/cc-debian12:nonroot as base
+FROM ubuntu:lunar as base
 WORKDIR /
 COPY ./license.html .
 COPY ./dependencies-src.zip .

src/cli.rs

@@ -202,58 +202,21 @@ impl Cli {
             shutdown_tx.send(()).ok();
         });
 
-        let fut = tryhard::retry_fn({
-            let shutdown_rx = shutdown_rx.clone();
-            let mode = mode.clone();
-            move || match self.command.clone() {
-                Commands::Agent(agent) => {
-                    let config = config.clone();
-                    let shutdown_rx = shutdown_rx.clone();
-                    let mode = mode.clone();
-                    tokio::spawn(async move {
-                        agent.run(config.clone(), mode, shutdown_rx.clone()).await
-                    })
-                }
-                Commands::Proxy(runner) => {
-                    let config = config.clone();
-                    let shutdown_rx = shutdown_rx.clone();
-                    let mode = mode.clone();
-                    tokio::spawn(async move {
-                        runner
-                            .run(config.clone(), mode.clone(), shutdown_rx.clone())
-                            .await
-                    })
-                }
-                Commands::Manage(manager) => {
-                    let config = config.clone();
-                    let shutdown_rx = shutdown_rx.clone();
-                    let mode = mode.clone();
-                    tokio::spawn(async move {
-                        manager
-                            .manage(config.clone(), mode, shutdown_rx.clone())
-                            .await
-                    })
-                }
-                Commands::Relay(relay) => {
-                    let config = config.clone();
-                    let shutdown_rx = shutdown_rx.clone();
-                    let mode = mode.clone();
-                    tokio::spawn(
-                        async move { relay.relay(config, mode, shutdown_rx.clone()).await },
-                    )
-                }
-                Commands::GenerateConfigSchema(_) | Commands::Qcmp(_) => unreachable!(),
+        match self.command {
+            Commands::Agent(agent) => agent.run(config.clone(), mode, shutdown_rx.clone()).await,
+            Commands::Proxy(runner) => {
+                runner
+                    .run(config.clone(), mode.clone(), shutdown_rx.clone())
+                    .await
             }
-        })
-        .retries(3)
-        .on_retry(|_, _, error| {
-            let error = error.to_string();
-            async move {
-                tracing::warn!(%error, "error would have caused fatal crash");
+            Commands::Manage(manager) => {
+                manager
+                    .manage(config.clone(), mode, shutdown_rx.clone())
+                    .await
             }
-        });
-
-        fut.await?
+            Commands::Relay(relay) => relay.relay(config, mode, shutdown_rx.clone()).await,
+            Commands::GenerateConfigSchema(_) | Commands::Qcmp(_) => unreachable!(),
+        }
     }
 
     /// Searches for the configuration file, and panics if not found.
@@ -330,7 +293,7 @@ mod tests {
 
         let endpoints_file = tempfile::NamedTempFile::new().unwrap();
         let config = Config::default();
-        let server_port = server_socket.local_ipv4_addr().unwrap().port();
+        let server_port = server_socket.local_addr().unwrap().port();
         std::fs::write(endpoints_file.path(), {
             config.clusters.write().insert_default(
                 [Endpoint::with_metadata(
@@ -403,18 +366,18 @@ mod tests {
 
         tokio::spawn(relay.drive());
         tokio::spawn(control_plane.drive());
-        tokio::time::sleep(Duration::from_millis(500)).await;
+        tokio::time::sleep(Duration::from_millis(50)).await;
         tokio::spawn(proxy.drive());
-        tokio::time::sleep(Duration::from_millis(500)).await;
+        tokio::time::sleep(Duration::from_millis(50)).await;
         let socket = create_socket().await;
         let config = Config::default();
         let proxy_address: SocketAddr = (std::net::Ipv4Addr::LOCALHOST, 7777).into();
 
+        let server_port = server_socket.local_addr().unwrap().port();
         for _ in 0..5 {
             let token = random_three_characters();
 
             tracing::info!(?token, "writing new config");
-            let server_port = server_socket.local_ipv4_addr().unwrap().port();
             std::fs::write(endpoints_file.path(), {
                 config.clusters.write().insert_default(
                     [Endpoint::with_metadata(
@@ -436,7 +399,7 @@ mod tests {
 
             assert_eq!(
                 "hello",
-                timeout(Duration::from_millis(500), rx.recv())
+                timeout(Duration::from_millis(100), rx.recv())
                     .await
                     .expect("should have received a packet")
                     .unwrap()
@@ -449,7 +412,7 @@ mod tests {
             let msg = b"hello\xFF\xFF\xFF";
             socket.send_to(msg, &proxy_address).await.unwrap();
 
-            let result = timeout(Duration::from_millis(500), rx.recv()).await;
+            let result = timeout(Duration::from_millis(50), rx.recv()).await;
             assert!(result.is_err(), "should not have received a packet");
             tracing::info!(?token, "didn't receive bad packet");
         }

src/cli/admin.rs

@@ -87,30 +87,38 @@ impl Admin {
         &self,
         config: Arc<Config>,
         address: Option<std::net::SocketAddr>,
-    ) -> tokio::task::JoinHandle<Result<(), hyper::Error>> {
+    ) -> std::thread::JoinHandle<Result<(), hyper::Error>> {
         let address = address.unwrap_or_else(|| (std::net::Ipv6Addr::UNSPECIFIED, PORT).into());
         let health = Health::new();
         tracing::info!(address = %address, "Starting admin endpoint");
 
         let mode = self.clone();
-        let make_svc = make_service_fn(move |_conn| {
-            let config = config.clone();
-            let health = health.clone();
-            let mode = mode.clone();
-            async move {
-                let config = config.clone();
-                let health = health.clone();
-                let mode = mode.clone();
-                Ok::<_, Infallible>(service_fn(move |req| {
+        std::thread::spawn(move || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_io()
+                .build()
+                .expect("couldn't create tokio runtime in thread");
+            runtime.block_on(async move {
+                let make_svc = make_service_fn(move |_conn| {
                     let config = config.clone();
                     let health = health.clone();
                     let mode = mode.clone();
-                    async move { Ok::<_, Infallible>(mode.handle_request(req, config, health)) }
-                }))
-            }
-        });
-
-        tokio::spawn(HyperServer::bind(&address).serve(make_svc))
+                    async move {
+                        let config = config.clone();
+                        let health = health.clone();
+                        let mode = mode.clone();
+                        Ok::<_, Infallible>(service_fn(move |req| {
+                            let config = config.clone();
+                            let health = health.clone();
+                            let mode = mode.clone();
+                            async move { Ok::<_, Infallible>(mode.handle_request(req, config, health)) }
+                        }))
+                    }
+                });
+
+                HyperServer::bind(&address).serve(make_svc).await
+            })
+        })
     }
 
     fn is_ready(&self, config: &Config) -> bool {