sova_core/server/
mod.rs

1mod body;
2mod conn;
3mod forwarded;
4
5#[allow(unused_imports)] // used via crate::server::collect_limited (tests / callers)
6pub use body::collect_limited;
7
8use crate::app::{App, ListenParts};
9use crate::error::{Error, Result};
10use std::net::SocketAddr;
11use std::pin::Pin;
12#[cfg(unix)]
13use std::path::Path;
14use tokio::net::TcpListener;
15
16/// Peer address stored on each request for rate-limiting etc.
17#[derive(Debug, Clone, Copy)]
18pub struct ClientAddr(pub SocketAddr);
19
20/// Authenticated principal for rate-limit identity keys.
21/// Set by auth / passport when a user is hydrated; rate-limit falls back to IP if absent.
22#[derive(Debug, Clone)]
23pub struct RateLimitIdentity(pub String);
24
25pub(crate) type ExternalShutdown = Pin<Box<dyn std::future::Future<Output = ()> + Send>>;
26
27#[cfg(feature = "tls")]
28pub(crate) type TlsOpt = Option<crate::tls::TlsRuntime>;
29
30#[cfg(not(feature = "tls"))]
31pub async fn listen(
32    app: App,
33    port: Option<u16>,
34    addr: Option<SocketAddr>,
35    external_shutdown: Option<ExternalShutdown>,
36) -> Result<()> {
37    let ListenParts {
38        inner,
39        startups,
40        shutdowns,
41        services,
42        start_services,
43    } = app.into_listen_parts()?;
44    let bind = addr
45        .or_else(|| port.map(|p| SocketAddr::from(([0, 0, 0, 0], p))))
46        .ok_or_else(|| Error::Internal("listen: port or address required".into()))?;
47    let listener = bind_tcp(bind, inner.reuseport).await?;
48    conn::run_tcp(
49        inner,
50        startups,
51        shutdowns,
52        services,
53        start_services,
54        listener,
55        external_shutdown,
56    )
57    .await
58}
59
60#[cfg(feature = "tls")]
61pub async fn listen(
62    app: App,
63    port: Option<u16>,
64    addr: Option<SocketAddr>,
65    external_shutdown: Option<ExternalShutdown>,
66    tls: TlsOpt,
67) -> Result<()> {
68    let ListenParts {
69        inner,
70        startups,
71        shutdowns,
72        services,
73        start_services,
74    } = app.into_listen_parts()?;
75    let bind = addr
76        .or_else(|| port.map(|p| SocketAddr::from(([0, 0, 0, 0], p))))
77        .ok_or_else(|| Error::Internal("listen: port or address required".into()))?;
78    let listener = bind_tcp(bind, inner.reuseport).await?;
79    conn::run_tcp(
80        inner,
81        startups,
82        shutdowns,
83        services,
84        start_services,
85        listener,
86        external_shutdown,
87        tls,
88    )
89    .await
90}
91
92async fn bind_tcp(bind: SocketAddr, reuseport: bool) -> Result<TcpListener> {
93    let reuseport = reuseport || env_truthy("SOVA_REUSEPORT");
94    if reuseport {
95        #[cfg(feature = "listen-reuseport")]
96        {
97            return bind_reuseport(bind).await;
98        }
99        #[cfg(not(feature = "listen-reuseport"))]
100        {
101            return Err(Error::Internal(
102                "BoundApp::reuseport(true) / SOVA_REUSEPORT requires feature `listen-reuseport`".into(),
103            ));
104        }
105    }
106    TcpListener::bind(bind)
107        .await
108        .map_err(|e| Error::Internal(format!("bind {bind}: {e}")))
109}
110
111fn env_truthy(name: &str) -> bool {
112    match std::env::var(name) {
113        Ok(v) => matches!(
114            v.trim().to_ascii_lowercase().as_str(),
115            "1" | "true" | "yes" | "on"
116        ),
117        Err(_) => false,
118    }
119}
120
121#[cfg(feature = "listen-reuseport")]
122async fn bind_reuseport(bind: SocketAddr) -> Result<TcpListener> {
123    use socket2::{Domain, Protocol, Socket, Type};
124    let domain = if bind.is_ipv4() {
125        Domain::IPV4
126    } else {
127        Domain::IPV6
128    };
129    let socket = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))
130        .map_err(|e| Error::Internal(format!("socket: {e}")))?;
131    socket
132        .set_reuse_address(true)
133        .map_err(|e| Error::Internal(format!("SO_REUSEADDR: {e}")))?;
134    #[cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos")))]
135    socket
136        .set_reuse_port(true)
137        .map_err(|e| Error::Internal(format!("SO_REUSEPORT: {e}")))?;
138    socket
139        .set_nonblocking(true)
140        .map_err(|e| Error::Internal(format!("nonblocking: {e}")))?;
141    socket
142        .bind(&bind.into())
143        .map_err(|e| Error::Internal(format!("bind {bind}: {e}")))?;
144    socket
145        .listen(1024)
146        .map_err(|e| Error::Internal(format!("listen: {e}")))?;
147    let std_listener: std::net::TcpListener = socket.into();
148    TcpListener::from_std(std_listener).map_err(|e| Error::Internal(format!("from_std: {e}")))
149}
150
151#[cfg(not(feature = "tls"))]
152pub async fn listen_with_listener(
153    app: App,
154    listener: std::net::TcpListener,
155    external_shutdown: Option<ExternalShutdown>,
156) -> Result<()> {
157    let ListenParts {
158        inner,
159        startups,
160        shutdowns,
161        services,
162        start_services,
163    } = app.into_listen_parts()?;
164    let listener = TcpListener::from_std(listener)
165        .map_err(|e| Error::Internal(format!("from_std: {e}")))?;
166    conn::run_tcp(
167        inner,
168        startups,
169        shutdowns,
170        services,
171        start_services,
172        listener,
173        external_shutdown,
174    )
175    .await
176}
177
178#[cfg(feature = "tls")]
179pub async fn listen_with_listener(
180    app: App,
181    listener: std::net::TcpListener,
182    external_shutdown: Option<ExternalShutdown>,
183    tls: TlsOpt,
184) -> Result<()> {
185    let ListenParts {
186        inner,
187        startups,
188        shutdowns,
189        services,
190        start_services,
191    } = app.into_listen_parts()?;
192    let listener = TcpListener::from_std(listener)
193        .map_err(|e| Error::Internal(format!("from_std: {e}")))?;
194    conn::run_tcp(
195        inner,
196        startups,
197        shutdowns,
198        services,
199        start_services,
200        listener,
201        external_shutdown,
202        tls,
203    )
204    .await
205}
206
207#[cfg(unix)]
208pub async fn listen_uds(
209    app: App,
210    path: &Path,
211    external_shutdown: Option<ExternalShutdown>,
212) -> Result<()> {
213    use tokio::net::UnixListener;
214
215    let ListenParts {
216        inner,
217        startups,
218        shutdowns,
219        services,
220        start_services,
221    } = app.into_listen_parts()?;
222    let _ = std::fs::remove_file(path);
223    let listener = UnixListener::bind(path)
224        .map_err(|e| Error::Internal(format!("bind uds {}: {e}", path.display())))?;
225    tracing::info!("sova listening on unix:{}", path.display());
226    conn::run_unix(
227        inner,
228        startups,
229        shutdowns,
230        services,
231        start_services,
232        listener,
233        external_shutdown,
234    )
235    .await
236}
sova_core/server/mod.rs

sova_core/server/
mod.rs
