[api] serve on a listener systemd passed in for the bind address
43 unmodified lines44454647484849505152118 unmodified linesExpand all43 unmodified lines| variable | default | description || :--- | :--- | :--- || `DATABASE_PATH` | `./hydrant.db` | path to the database folder, relative to the working directory. it holds the journal, the keyspaces, the lock file and any trained `dict_*.bin`. only one process can open it at a time. a new path is the only way to get a fresh database, which is what the [fixed at creation](#fixed-when-the-database-is-created) settings need || `RUST_LOG` | `info,fjall=warn,lsm_tree=warn` | log filter directives (e.g., `debug`, `hydrant=trace`). [tracing env-filter syntax](https://docs.rs/tracing-subscriber/latest/tracing_subscriber/filter/struct.EnvFilter.html), bad directives are skipped. at `info` every firehose message already opens a log span, so `warn` is cheaper on a busy indexer. embedded: an empty string isn't treated as unset, so `""` means plain `info` with the storage engine's logs back on, and only the first `hydrant_start` in a process sets the filter, later ones keep it || `API_BIND` | `0.0.0.0:3000,[::]:3000` | comma-separated list of `<ip>:<port>` socket addresses and `unix:<path>` sockets to bind the API server to. literal IPs only (hostnames not resolved). when both an ipv4 and ipv6 entry share the same port, the v6 listener is set to v6-only to avoid bind collision, and a lone `[::]:<port>` listens dual-stack. unix sockets get `API_SOCKET_MODE`. a stale socket left by a crashed hydrant is replaced, but one that is still being listened on is not. the socket is set up in a short-lived `.hydrant-<random>/` directory next to it and renamed into place, so its directory has to be writable and its path about 20 characters shorter than the os limit (104 on macOS, 108 on linux). `none` runs without the API, and an empty value is a startup error. embedded hydrant has no default and requires it || `API_BIND` | `0.0.0.0:3000,[::]:3000` | comma-separated list of `<ip>:<port>` socket addresses and `unix:<path>` sockets to bind the API server to. literal IPs only (hostnames not resolved). when both an ipv4 and ipv6 entry share the same port, the v6 listener is set to v6-only to avoid bind collision, and a lone `[::]:<port>` listens dual-stack. unix sockets get `API_SOCKET_MODE`. a stale socket left by a crashed hydrant is replaced, but one that is still being listened on is not. the socket is set up in a short-lived `.hydrant-<random>/` directory next to it and renamed into place, so its directory has to be writable and its path about 20 characters shorter than the os limit (104 on macOS, 108 on linux). `none` runs without the API, and an empty value is a startup error. when systemd's socket activation (`LISTEN_FDS`) passes in a socket already listening on a tcp address, hydrant serves on it instead of binding, so connections made during a restart wait instead of being refused. embedded hydrant has no default and requires it || `API_SOCKET_MODE` | `0600` | octal permissions for `unix:` sockets in `API_BIND`. owner-only by default since the API has no auth, `0660` also lets the socket's group in. anything that isn't octal up to `0777` is a startup error || `API_TCP_MANAGEMENT` | `false` | also serve the [management endpoints](getting-started.md#management-endpoints-keep-private) on tcp binds. by default they're only served on `unix:` binds, since they have no auth and the socket's permissions are what guards them, so with only tcp binds there is no management API. || `GET_REPO_CONCURRENCY_LIMIT` | `2` | how many public `com.atproto.sync.getRepo` requests can build a repo at once. each one reads every record of the repo and rebuilds its tree in memory. a request past the limit gets a rate limited error straight away, it doesn't wait. the slot is released once the repo is built, before it's streamed to the client, so slow clients can keep more repos than this in memory. getRepo only works on databases that store every collection's record bodies (no `FILTER_COLLECTIONS`, `EPHEMERAL` or `ONLY_INDEX_LINKS`) || `ENABLE_DEBUG` | `false` | enable debug endpoints. binary only |118 unmodified linesExpand all2 unmodified lines345677891011121314145 unmodified linesExpand all1601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472 unmodified linesuse std::time::{Duration, Instant};
/// a hydrant on a database in `dir` that never touches the networkfn offline_hydrant(dir: &Path) -> Command { let mut cmd = Command::new(env!("CARGO_BIN_EXE_hydrant")); offline(Command::new(env!("CARGO_BIN_EXE_hydrant")), dir)}
fn offline(mut cmd: Command, dir: &Path) -> Command { cmd.env("HYDRANT_DATABASE_PATH", dir.join("db")) .env( "HYDRANT_API_BIND", format!("unix:{}", dir.join("api.sock").display()),145 unmodified linesExpand all "{stdout}{}", String::from_utf8_lossy(&out.stderr) );}
#[cfg(feature = "indexer_stream")]fn socket_activated(dir: &Path, listener: &std::net::TcpListener) -> Child { use std::os::fd::AsRawFd; use std::os::unix::process::CommandExt;
// exec keeps the shell's pid, which makes $$ hydrant's pid let mut sh = Command::new("sh"); sh.args(["-c", "LISTEN_PID=$$ exec \"$0\"", env!("CARGO_BIN_EXE_hydrant")]); let mut cmd = offline(sh, dir); cmd.env("HYDRANT_API_BIND", listener.local_addr().unwrap().to_string()) .env("LISTEN_FDS", "1"); let fd = listener.as_raw_fd(); // safety: dup2 and fcntl are async-signal-safe unsafe { cmd.pre_exec(move || { if fd == 3 { libc::fcntl(3, libc::F_SETFD, 0); } else if libc::dup2(fd, 3) < 0 { return Err(std::io::Error::last_os_error()); } Ok(()) }); } cmd.spawn().unwrap()}
/// connect succeeds off the backlog and only the read waits#[cfg(feature = "indexer_stream")]fn get_health(addr: std::net::SocketAddr) -> String { use std::io::{Read, Write}; let mut conn = std::net::TcpStream::connect(addr).expect("the listener is never closed"); conn.set_read_timeout(Some(Duration::from_secs(60))).unwrap(); conn.write_all(b"GET /health HTTP/1.1\r\nhost: hydrant\r\nconnection: close\r\n\r\n") .unwrap(); let mut response = String::new(); conn.read_to_string(&mut response).unwrap(); response}
#[cfg(feature = "indexer_stream")]#[test]fn a_socket_activated_restart_closes_streams_and_keeps_new_clients_waiting() { use futures::StreamExt; use tokio_websockets::{ClientBuilder, CloseCode};
let tmp = tempfile::tempdir().unwrap(); let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let addr = listener.local_addr().unwrap(); let rt = tokio::runtime::Builder::new_current_thread() .enable_all() .build() .unwrap();
let first = socket_activated(tmp.path(), &listener); assert!(get_health(addr).starts_with("HTTP/1.1 200")); let uri = format!("ws://{addr}/stream"); let (mut subscriber, _) = rt .block_on(ClientBuilder::new().uri(&uri).unwrap().connect()) .unwrap();
send_signal(&first, "-TERM"); let closed = rt .block_on(async { tokio::time::timeout(Duration::from_secs(10), subscriber.next()).await }) .expect("the stream must be closed when hydrant stops") .expect("a close frame, not the socket just going") .unwrap(); assert_eq!( closed.as_close().map(|(code, _)| code), Some(CloseCode::SERVICE_RESTART) ); let out = wait_for_exit(first, Duration::from_secs(45)).expect("hydrant didn't stop"); assert!(out.status.success(), "{out:?}");
let waiting = std::thread::spawn(move || get_health(addr)); std::thread::sleep(Duration::from_millis(200)); let second = socket_activated(tmp.path(), &listener); let response = waiting.join().unwrap(); assert!(response.starts_with("HTTP/1.1 200"), "{response}");
send_signal(&second, "-TERM"); let out = wait_for_exit(second, Duration::from_secs(45)).expect("hydrant didn't stop"); assert!(out.status.success(), "{out:?}");}1 unmodified line2345678910111213177 unmodified linesExpand all19119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424579 unmodified lines32532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637750 unmodified lines1 unmodified lineuse crate::state::AppState;use axum::{Router, routing::get};use futures::FutureExt;use futures::future::BoxFuture;use std::collections::HashMap;use std::os::fd::{FromRawFd, IntoRawFd, RawFd};use std::path::Path;use std::sync::{LazyLock, Mutex};use std::{net::SocketAddr, sync::Arc};use tower_http::cors::CorsLayer;use tower_http::trace::TraceLayer;
177 unmodified linesExpand all .any(|a| matches!(a, SocketAddr::V4(v4) if v4.port() == v6.port())); Some(has_v4_sibling)}
const LISTEN_FDS_START: RawFd = 3;
/// systemd holds these sockets across restarts and connects queue until hydrant is backfn inherited_listeners( lookup: impl Fn(&str) -> Option<String>, pid: u32, first: RawFd,) -> HashMap<SocketAddr, std::net::TcpListener> { let ours = lookup("LISTEN_PID").and_then(|p| p.parse::<u32>().ok()) == Some(pid); let count = lookup("LISTEN_FDS") .and_then(|n| n.parse::<RawFd>().ok()) .filter(|_| ours) .unwrap_or(0); (first..first.saturating_add(count)) .filter_map(|fd| { // safety: systemd passed this fd to this pid, and it's taken only once let socket = unsafe { socket2::Socket::from_raw_fd(fd) }; let stream = socket.r#type().ok() == Some(socket2::Type::STREAM); let addr = socket.local_addr().ok().and_then(|addr| addr.as_socket()); match addr.filter(|_| stream) { Some(addr) => Some((addr, socket.into())), None => { // dropping the socket would close an fd that isn't ours let _ = socket.into_raw_fd(); None } } }) .collect()}
static INHERITED_LISTENERS: LazyLock<Mutex<HashMap<SocketAddr, std::net::TcpListener>>> = LazyLock::new(|| { let env = |key: &str| std::env::var(key).ok(); Mutex::new(inherited_listeners(env, std::process::id(), LISTEN_FDS_START)) });
fn bind_listener( addr: SocketAddr, v6only: Option<bool>,) -> std::io::Result<tokio::net::TcpListener> { let inherited = INHERITED_LISTENERS.lock().unwrap().remove(&addr); if let Some(listener) = inherited { tracing::info!("using the listener systemd passed in for {addr}"); listener.set_nonblocking(true)?; return tokio::net::TcpListener::from_std(listener); } let domain = match addr { SocketAddr::V4(_) => socket2::Domain::IPV4, SocketAddr::V6(_) => socket2::Domain::IPV6, };79 unmodified linesmod tests { use super::*; use std::os::unix::fs::PermissionsExt;
fn listen_env(pid: u32, fds: usize) -> impl Fn(&str) -> Option<String> { move |key| match key { "LISTEN_PID" => Some(pid.to_string()), "LISTEN_FDS" => Some(fds.to_string()), _ => None, } }
#[test] fn inherited_listeners_are_found_by_address() { let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let addr = listener.local_addr().unwrap(); let fd = listener.into_raw_fd();
let mut found = inherited_listeners(listen_env(std::process::id(), 1), std::process::id(), fd); let listener = found.remove(&addr).expect("the passed listener is keyed by its address"); assert!(found.is_empty()); let client = std::net::TcpStream::connect(addr).unwrap(); let (accepted, _) = listener.accept().unwrap(); assert_eq!(accepted.peer_addr().unwrap(), client.local_addr().unwrap()); }
#[test] fn listen_fds_meant_for_another_pid_are_left_alone() { use std::os::fd::AsRawFd; let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let fd = listener.as_raw_fd(); let pid = std::process::id(); assert!(inherited_listeners(listen_env(pid + 1, 1), pid, fd).is_empty()); assert!(inherited_listeners(|_| None, pid, fd).is_empty()); listener.local_addr().unwrap(); }
#[test] fn inherited_unix_sockets_stay_open() { use std::os::fd::AsRawFd; let tmp = tempfile::tempdir().unwrap(); let path = tmp.path().join("s"); let listener = std::os::unix::net::UnixListener::bind(&path).unwrap(); let pid = std::process::id(); assert!(inherited_listeners(listen_env(pid, 1), pid, listener.as_raw_fd()).is_empty()); std::os::unix::net::UnixStream::connect(&path).unwrap(); listener.accept().unwrap(); }
#[tokio::test] async fn unix_bind_uses_the_configured_mode() { let tmp = tempfile::tempdir().unwrap(); let path = tmp.path().join("api.sock");50 unmodified lines