feat(rivetkit): move HTTP listener to Rust + RIVETKIT_RUNTIME_MODE env var

rivetkit-rust/packages/rivetkit-core/Cargo.toml

@@ -15,6 +15,11 @@ default = ["native-runtime"]
 native-runtime = [
 	"dep:nix",
 	"dep:reqwest",
+	"dep:axum",
+	"dep:bytes",
+	"dep:http-body-util",
+	"dep:tokio-stream",
+	"dep:tower-http",
 	"rivet-envoy-client/native-transport",
 ]
 wasm-runtime = ["rivet-envoy-client/wasm-transport"]
@@ -29,11 +34,14 @@ fs_extra = { workspace = true }
 [dependencies]
 anyhow.workspace = true
 async-trait = { workspace = true, optional = true }
+axum = { workspace = true, optional = true }
 base64.workspace = true
+bytes = { workspace = true, optional = true }
 ciborium.workspace = true
 futures.workspace = true
 http.workspace = true
 include_dir = { workspace = true }
+http-body-util = { workspace = true, optional = true }
 nix = { workspace = true, optional = true }
 parking_lot.workspace = true
 rand.workspace = true
@@ -54,7 +62,9 @@ serde_bare.workspace = true
 serde_bytes.workspace = true
 sha2.workspace = true
 subtle.workspace = true
+tokio-stream = { workspace = true, optional = true }
 tokio-util.workspace = true
+tower-http = { workspace = true, optional = true, features = ["fs"] }
 tracing.workspace = true
 url.workspace = true
 vbare.workspace = true

rivetkit-rust/packages/rivetkit-core/src/lib.rs

@@ -12,6 +12,8 @@ pub mod metrics_endpoint;
 pub mod registry;
 pub mod runtime;
 pub mod serverless;
+#[cfg(feature = "native-runtime")]
+pub mod serverless_http;
 pub(crate) mod time {
 	use std::fmt;
 	use std::future::Future;

rivetkit-rust/packages/rivetkit-core/src/registry/mod.rs

@@ -102,6 +102,9 @@ impl CoreEnvoyHandle {
 		}
 	}
 
+	/// Engine-reported drain threshold in milliseconds. `None` until the
+	/// envoy has completed its first protocol-metadata exchange with the
+	/// engine.
 	pub async fn actor_stop_threshold_ms(&self) -> Option<i64> {
 		self.handle
 			.get_protocol_metadata()

rivetkit-rust/packages/rivetkit-core/src/serverless.rs

@@ -136,10 +136,9 @@ struct NamespaceMismatch {
 	"message",
 	"incoming_too_long",
 	"Incoming message too long.",
-	"Incoming message too long. Received {size} bytes, limit is {limit} bytes."
+	"Incoming message too long. Exceeded limit of {limit} bytes."
 )]
 struct IncomingMessageTooLong {
-	size: usize,
 	limit: usize,
 }
 
@@ -252,6 +251,29 @@ impl CoreServerlessRuntime {
 		CoreEnvoyHandle::new(handle).actor_stop_threshold_ms().await
 	}
 
+	/// Listener-side body cap; reuses the `/start` payload limit.
+	pub fn max_request_body_bytes(&self) -> usize {
+		self.settings.max_start_payload_bytes
+	}
+
+	/// Canonical 413 response built through the `RivetError` system.
+	pub fn incoming_too_long_response(&self) -> ServerlessResponse {
+		let error = IncomingMessageTooLong {
+			limit: self.settings.max_start_payload_bytes,
+		}
+		.build();
+		error_response(error)
+	}
+
+	/// Canonical 400 response for malformed requests.
+	pub fn invalid_request_response(&self, reason: impl Into<String>) -> ServerlessResponse {
+		let error = InvalidRequest {
+			reason: reason.into(),
+		}
+		.build();
+		error_response(error)
+	}
+
 	pub async fn handle_request(&self, req: ServerlessRequest) -> ServerlessResponse {
 		let cors = cors_headers(&req);
 		match self.handle_request_inner(req).await {
@@ -334,7 +356,6 @@ impl CoreServerlessRuntime {
 		);
 		if req.body.len() > self.settings.max_start_payload_bytes {
 			return Err(IncomingMessageTooLong {
-				size: req.body.len(),
 				limit: self.settings.max_start_payload_bytes,
 			}
 			.build());