Something went wrong. Try again.
very fast at protocol indexer with flexible filtering, xrpc queries, cursor-backed event stream, and more, built on fjall
rust fjall at-protocol atproto indexer
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248use std::path::Path;use std::process::{Child, Command, Output, Stdio};use std::time::{Duration, Instant};
/// a hydrant on a database in `dir` that never touches the networkfn offline_hydrant(dir: &Path) -> Command { 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()), ) .env("HYDRANT_ENABLE_FIREHOSE", "false") .env("HYDRANT_ENABLE_CRAWLER", "false") .env("HYDRANT_ENABLE_BACKFILL", "false") .env("HYDRANT_RELAY_HOSTS", "") .env("HYDRANT_SEED_HOSTS", "") .env("HYDRANT_CRAWLER_URLS", "") .stdin(Stdio::null()) .stdout(Stdio::piped()) .stderr(Stdio::piped()); cmd}
// config is env only, so anything on argv has to be refused before hydrant boots.// if it does boot, it gets a throwaway database and no network, and we kill it.fn hydrant(args: &[&str]) -> Output { let tmp = tempfile::tempdir().unwrap(); let mut child = offline_hydrant(tmp.path()).args(args).spawn().unwrap(); let deadline = Instant::now() + Duration::from_secs(30); while child.try_wait().unwrap().is_none() { if Instant::now() > deadline { child.kill().unwrap(); let out = child.wait_with_output().unwrap(); panic!( "hydrant {args:?} booted instead of exiting:\n{}", String::from_utf8_lossy(&out.stderr) ); } std::thread::sleep(Duration::from_millis(50)); } child.wait_with_output().unwrap()}
#[test]fn version_prints_and_exits() { for flag in ["--version", "-V"] { let out = hydrant(&[flag]); assert!(out.status.success(), "{flag}: {out:?}"); assert_eq!( String::from_utf8_lossy(&out.stdout), format!("hydrant {}\n", env!("CARGO_PKG_VERSION")) ); }}
#[test]fn help_prints_usage_and_exits() { for flag in ["--help", "-h"] { let out = hydrant(&[flag]); assert!(out.status.success(), "{flag}: {out:?}"); assert!(String::from_utf8_lossy(&out.stdout).starts_with("usage: hydrant")); }}
#[test]fn any_other_argument_is_an_error() { for args in [&["--port=3000"][..], &["serve"], &["--version", "extra"]] { let out = hydrant(args); assert!(!out.status.success(), "{args:?}: {out:?}"); assert!(out.stdout.is_empty(), "{args:?}: {out:?}"); assert!( String::from_utf8_lossy(&out.stderr).contains("unexpected argument"), "{args:?}: {out:?}" ); }}
/// waits for `child` to exit on its own, killing it after `timeout`fn wait_for_exit(mut child: Child, timeout: Duration) -> Option<Output> { let deadline = Instant::now() + timeout; while child.try_wait().unwrap().is_none() { if Instant::now() > deadline { child.kill().unwrap(); return None; } std::thread::sleep(Duration::from_millis(50)); } Some(child.wait_with_output().unwrap())}
fn send_signal(child: &Child, signal: &str) { let status = Command::new("kill") .args([signal, &child.id().to_string()]) .status() .unwrap(); assert!(status.success(), "kill {signal} failed");}
/// starts hydrant on `dir` and returns once its api socket is up, so it's fully runningfn start_and_wait_for_api(dir: &Path) -> Child { let socket = dir.join("api.sock"); let _ = std::fs::remove_file(&socket); let mut child = offline_hydrant(dir).spawn().unwrap(); let deadline = Instant::now() + Duration::from_secs(60); while !socket.exists() { if let Some(status) = child.try_wait().unwrap() { let out = child.wait_with_output().unwrap(); panic!( "hydrant exited with {status} before serving:\n{}", String::from_utf8_lossy(&out.stderr) ); } assert!( Instant::now() < deadline, "hydrant never bound its api socket" ); std::thread::sleep(Duration::from_millis(50)); } child}
#[test]fn sigterm_and_sigint_stop_cleanly_and_the_database_opens_again() { let tmp = tempfile::tempdir().unwrap(); // the second and third start only get past opening the database if the one before let go for signal in ["-TERM", "-INT", "-TERM"] { let child = start_and_wait_for_api(tmp.path()); send_signal(&child, signal); let out = wait_for_exit(child, Duration::from_secs(45)) .unwrap_or_else(|| panic!("hydrant didn't stop within 45s of {signal}")); let stdout = String::from_utf8_lossy(&out.stdout); let stderr = String::from_utf8_lossy(&out.stderr); assert!( out.status.success(), "{signal}: {:?}\n{stdout}{stderr}", out.status ); assert!(stdout.contains("stopped"), "{signal}: {stdout}{stderr}"); }}
#[test]fn config_warnings_are_logged() { let tmp = tempfile::tempdir().unwrap(); // the bad history ttl makes it exit right after reading the config let child = offline_hydrant(tmp.path()) .env("HYDRANT_RELAY_HOSTS", "not a url") .env("HYDRANT_HISTORY_TTL", "soon") .spawn() .unwrap(); let out = wait_for_exit(child, Duration::from_secs(30)).expect("hydrant booted"); assert!(!out.status.success(), "{out:?}"); let stdout = String::from_utf8_lossy(&out.stdout); assert!( stdout.contains("invalid relay host URL: not a url"), "{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:?}");}