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;
11#[cfg(unix)]
12use std::path::Path;
13use std::pin::Pin;
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`"
103                    .into(),
104            ));
105        }
106    }
107    TcpListener::bind(bind)
108        .await
109        .map_err(|e| Error::Internal(format!("bind {bind}: {e}")))
110}
111
112fn env_truthy(name: &str) -> bool {
113    match std::env::var(name) {
114        Ok(v) => matches!(
115            v.trim().to_ascii_lowercase().as_str(),
116            "1" | "true" | "yes" | "on"
117        ),
118        Err(_) => false,
119    }
120}
121
122#[cfg(feature = "listen-reuseport")]
123async fn bind_reuseport(bind: SocketAddr) -> Result<TcpListener> {
124    use socket2::{Domain, Protocol, Socket, Type};
125    let domain = if bind.is_ipv4() {
126        Domain::IPV4
127    } else {
128        Domain::IPV6
129    };
130    let socket = Socket::new(domain, Type::STREAM, Some(Protocol::TCP))
131        .map_err(|e| Error::Internal(format!("socket: {e}")))?;
132    socket
133        .set_reuse_address(true)
134        .map_err(|e| Error::Internal(format!("SO_REUSEADDR: {e}")))?;
135    #[cfg(all(unix, not(target_os = "solaris"), not(target_os = "illumos")))]
136    socket
137        .set_reuse_port(true)
138        .map_err(|e| Error::Internal(format!("SO_REUSEPORT: {e}")))?;
139    socket
140        .set_nonblocking(true)
141        .map_err(|e| Error::Internal(format!("nonblocking: {e}")))?;
142    socket
143        .bind(&bind.into())
144        .map_err(|e| Error::Internal(format!("bind {bind}: {e}")))?;
145    socket
146        .listen(1024)
147        .map_err(|e| Error::Internal(format!("listen: {e}")))?;
148    let std_listener: std::net::TcpListener = socket.into();
149    TcpListener::from_std(std_listener).map_err(|e| Error::Internal(format!("from_std: {e}")))
150}
151
152#[cfg(not(feature = "tls"))]
153pub async fn listen_with_listener(
154    app: App,
155    listener: std::net::TcpListener,
156    external_shutdown: Option<ExternalShutdown>,
157) -> Result<()> {
158    let ListenParts {
159        inner,
160        startups,
161        shutdowns,
162        services,
163        start_services,
164    } = app.into_listen_parts()?;
165    let listener =
166        TcpListener::from_std(listener).map_err(|e| Error::Internal(format!("from_std: {e}")))?;
167    conn::run_tcp(
168        inner,
169        startups,
170        shutdowns,
171        services,
172        start_services,
173        listener,
174        external_shutdown,
175    )
176    .await
177}
178
179#[cfg(feature = "tls")]
180pub async fn listen_with_listener(
181    app: App,
182    listener: std::net::TcpListener,
183    external_shutdown: Option<ExternalShutdown>,
184    tls: TlsOpt,
185) -> Result<()> {
186    let ListenParts {
187        inner,
188        startups,
189        shutdowns,
190        services,
191        start_services,
192    } = app.into_listen_parts()?;
193    let listener =
194        TcpListener::from_std(listener).map_err(|e| Error::Internal(format!("from_std: {e}")))?;
195    conn::run_tcp(
196        inner,
197        startups,
198        shutdowns,
199        services,
200        start_services,
201        listener,
202        external_shutdown,
203        tls,
204    )
205    .await
206}
207
208#[cfg(unix)]
209pub async fn listen_uds(
210    app: App,
211    path: &Path,
212    external_shutdown: Option<ExternalShutdown>,
213) -> Result<()> {
214    use tokio::net::UnixListener;
215
216    let ListenParts {
217        inner,
218        startups,
219        shutdowns,
220        services,
221        start_services,
222    } = app.into_listen_parts()?;
223    let _ = std::fs::remove_file(path);
224    let listener = UnixListener::bind(path)
225        .map_err(|e| Error::Internal(format!("bind uds {}: {e}", path.display())))?;
226    tracing::info!("sova listening on unix:{}", path.display());
227    conn::run_unix(
228        inner,
229        startups,
230        shutdowns,
231        services,
232        start_services,
233        listener,
234        external_shutdown,
235    )
236    .await
237}
sova_core/server/mod.rs

sova_core/server/
mod.rs
