fix: fix ds echo build (#1032)

<!-- Please make sure there is an issue that this PR is correlated to. -->

## Changes

<!-- If there are frontend changes, please include screenshots. -->

infra/default-builds/dockerfiles/test-ds-echo/Cargo.toml

@@ -4,10 +4,12 @@ version = "0.1.0"
 edition = "2021"
 
 [dependencies]
-tokio = { version = "1.29", features = ["full"] }
-reqwest = "0.11"
 anyhow = "1.0"
+futures-util = { version = "0.3", features = ["sink"] }
 hyper = { version = "0.14", features = ["server"] }
+reqwest = "0.11"
+tokio = { version = "1.29", features = ["full"] }
+tokio-util = "0.7.11"
 
 [profile.release]
 opt-level = 'z'

infra/default-builds/dockerfiles/test-ds-echo/src/main.rs

@@ -1,10 +1,9 @@
 use std::{convert::Infallible, env, net::SocketAddr, process::Command};
 
 use anyhow::{Context, Result};
-use tokio::{
-	io::{AsyncBufReadExt, AsyncWriteExt},
-	net::{TcpListener, UdpSocket},
-};
+use futures_util::{sink::SinkExt, stream::StreamExt};
+use tokio::net::{TcpListener, UdpSocket};
+use tokio_util::codec::{BytesCodec, Framed};
 
 #[tokio::main]
 async fn main() -> Result<()> {
@@ -22,44 +21,21 @@ async fn main() -> Result<()> {
 		String::from_utf8_lossy(&output.stdout)
 	);
 
-	// TEMP: Expose dev port
-	tokio::spawn(with_select_term(echo_http_server(28234)));
-
-	// TODO: Add back
-	// Echo servers (bridge networking)
-	// if let Ok(http_port) = env::var("PORT_test_http") {
-	// 	let http_port: u16 = http_port.parse()?;
-	// 	tokio::spawn(with_select_term(echo_http_server(http_port)));
-	// }
-	//
-	// if let Ok(tcp_port) = env::var("PORT_test_tcp") {
-	// 	let tcp_port: u16 = tcp_port.parse()?;
-	// 	tokio::spawn(with_select_term(echo_tcp_server(tcp_port)));
-	// }
-	//
-	// if let Ok(udp_port) = env::var("PORT_test_udp") {
-	// 	let udp_port: u16 = udp_port.parse()?;
-	// 	tokio::spawn(with_select_term(echo_udp_server(udp_port)));
-	// }
-	//
-	// // Echo servers (host networking)
-	// if let Ok(http_port) = env::var("HOST_PORT_HTTP") {
-	// 	let http_port: u16 = http_port.parse()?;
-	// 	tokio::spawn(with_select_term(echo_http_server(http_port)));
-	// }
-	//
-	// if let Ok(tcp_port) = env::var("HOST_PORT_TCP") {
-	// 	let tcp_port: u16 = tcp_port.parse()?;
-	// 	tokio::spawn(with_select_term(echo_tcp_server(tcp_port)));
-	// }
-	//
-	// if let Ok(udp_port) = env::var("HOST_PORT_UDP") {
-	// 	let udp_port: u16 = udp_port.parse()?;
-	// 	tokio::spawn(with_select_term(echo_udp_server(udp_port)));
-	// }
-
-	// Lobby ready
-	// lobby_ready().await?;
+	// Echo servers
+	if let Ok(http_port) = env::var("HTTP_PORT") {
+		let http_port: u16 = http_port.parse()?;
+		tokio::spawn(with_select_term(echo_http_server(http_port)));
+	}
+
+	if let Ok(tcp_port) = env::var("TCP_PORT") {
+		let tcp_port: u16 = tcp_port.parse()?;
+		tokio::spawn(with_select_term(echo_tcp_server(tcp_port)));
+	}
+
+	if let Ok(udp_port) = env::var("UDP_PORT") {
+		let udp_port: u16 = udp_port.parse()?;
+		tokio::spawn(with_select_term(echo_udp_server(udp_port)));
+	}
 
 	// Wait indefinitely
 	println!("Waiting indefinitely...");
@@ -69,25 +45,6 @@ async fn main() -> Result<()> {
 	Ok(())
 }
 
-// async fn lobby_ready() -> Result<()> {
-// 	let url = format!(
-// 		"{}/matchmaker/lobbies/ready",
-// 		env::var("RIVET_API_ENDPOINT").context("RIVET_API_ENDPOINT")?
-// 	);
-// 	let token = env::var("RIVET_TOKEN").context("RIVET_TOKEN")?;
-//
-// 	let client = reqwest::Client::new();
-// 	client
-// 		.post(&url)
-// 		.header("Content-Type", "application/json")
-// 		.header("Authorization", format!("Bearer {}", token))
-// 		.send()
-// 		.await?;
-//
-// 	println!("Success, waiting indefinitely");
-// 	Ok(())
-// }
-
 /// Waits for the SIGTERM signal.
 async fn wait_term() -> Result<()> {
 	use tokio::signal::unix::{signal, SignalKind};
@@ -141,22 +98,13 @@ async fn echo_tcp_server(port: u16) -> Result<()> {
 	loop {
 		let (socket, _) = listener.accept().await.context("accept failed")?;
 		tokio::spawn(async move {
-			let mut reader = tokio::io::BufReader::new(socket);
-			let mut line = String::new();
-			loop {
-				let bytes_read = reader.read_line(&mut line).await.expect("read line failed");
-				if bytes_read == 0 {
-					break;
-				}
-
-				// Echo the line
-				reader
-					.get_mut()
-					.write_all(format!("{line}\n").as_bytes())
+			let mut framed = Framed::new(socket, BytesCodec::new());
+
+			while let Some(res) = framed.next().await {
+				framed
+					.send(res.expect("read failed"))
 					.await
 					.expect("write failed");
-				reader.get_mut().flush().await.expect("flush failed");
-				line.clear();
 			}
 		});
 	}

infra/default-builds/outputs/game-multiplayer-tag.txt

@@ -1 +1 @@
-game-multiplayer:1717019324
+game-multiplayer:1723062223

infra/default-builds/outputs/test-ds-echo-tag.txt

@@ -1 +1 @@
-test-ds-echo:1719995033
+test-ds-echo:1723071093

infra/default-builds/outputs/test-fail-immediately-tag.txt

@@ -1 +1 @@
-test-fail-immediately:1717019324
+test-fail-immediately:1723062223

infra/default-builds/outputs/test-hang-indefinitely-tag.txt

@@ -1 +1 @@
-test-hang-indefinitely:1717019324
+test-hang-indefinitely:1723062223

infra/default-builds/outputs/test-mm-lobby-echo-tag.txt

@@ -1 +1 @@
-test-mm-lobby-echo:1717019324
+test-mm-lobby-echo:1723062223

infra/default-builds/outputs/test-mm-lobby-ready-tag.txt

@@ -1 +1 @@
-test-mm-lobby-ready:1717019324
+test-mm-lobby-ready:1723062223

infra/default-builds/outputs/test-mm-player-connect-tag.txt

@@ -1 +1 @@
-test-mm-player-connect:1717019324
+test-mm-player-connect:1723062223

svc/pkg/ds/ops/server-create/tests/print_test_data.rs

@@ -94,7 +94,7 @@ async fn print_test_data(ctx: TestCtx) {
 				)),
 			}]},
 		)),
-		label: Some("game_service".to_owned()),
+		label: Some("cloud".to_owned()),
 		..Default::default()
 	})
 	.await
@@ -128,7 +128,6 @@ async fn print_test_data(ctx: TestCtx) {
 		cloud_token = ?cloud_token_res.token.clone().unwrap().token,
 		invalid_token = ?invalid_token.token.clone().unwrap().token,
 		build_id = ?build_res.build_id.unwrap(),
-		game_id = ?game_id,
 		"test data");
 	//
 	// let runtime = Some(