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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314package hydrant
import ( "bufio" "context" "encoding/json" "errors" "fmt" "io" "net" "net/http" "os" "path/filepath" "strings" "testing" "time")
// settings for a hydrant that never touches the network, so the test runs offlinefunc offlineSettings(dir string) map[string]string { return map[string]string{ "HYDRANT_DATABASE_PATH": filepath.Join(dir, "db"), "HYDRANT_API_BIND": "unix:" + filepath.Join(dir, "api.sock"), "HYDRANT_RELAY_HOSTS": "", "HYDRANT_SEED_HOSTS": "", "HYDRANT_CRAWLER_URLS": "", "HYDRANT_ENABLE_CRAWLER": "false", "RUST_LOG": "warn", }}
// a short temp dir, because unix socket paths are capped at ~104 bytes on darwinfunc shortTempDir(t *testing.T) string { t.Helper() dir, err := os.MkdirTemp("/tmp", "hydrant-ffi-") if err != nil { t.Fatal(err) } t.Cleanup(func() { os.RemoveAll(dir) }) return dir}
// starts hydrant and shuts it down when the test ends, before its dir is removedfunc startHydrant(t *testing.T, settings map[string]string) *Hydrant { t.Helper() h, err := Start(settings) if err != nil { t.Fatalf("start: %v", err) } t.Cleanup(func() { if err := h.Shutdown(10 * time.Second); err != nil { t.Errorf("shutdown: %v", err) } }) return h}
func unixClient(socket string) *http.Client { return &http.Client{ Timeout: 5 * time.Second, Transport: &http.Transport{ DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { var d net.Dialer return d.DialContext(ctx, "unix", socket) }, }, }}
func waitReady(t *testing.T, h *Hydrant, client *http.Client) { t.Helper() deadline := time.Now().Add(30 * time.Second) for time.Now().Before(deadline) { select { case <-h.Done(): t.Fatalf("hydrant stopped before its api came up: %v", h.Err()) default: } if resp, err := client.Get("http://hydrant/stats"); err == nil { resp.Body.Close() if resp.StatusCode == http.StatusOK { return } } time.Sleep(100 * time.Millisecond) } t.Fatal("hydrant api never answered on the socket")}
func request(t *testing.T, client *http.Client, method, path, body string) string { t.Helper() req, err := http.NewRequest(method, "http://hydrant"+path, strings.NewReader(body)) if err != nil { t.Fatal(err) } if body != "" { req.Header.Set("Content-Type", "application/json") } resp, err := client.Do(req) if err != nil { t.Fatalf("%s %s: %v", method, path, err) } defer resp.Body.Close() out, _ := io.ReadAll(resp.Body) if resp.StatusCode != http.StatusOK { t.Fatalf("%s %s: status %d: %s", method, path, resp.StatusCode, out) } return string(out)}
// opens a /stream websocket and leaves it open until the test endsfunc subscribe(t *testing.T, socket string) net.Conn { t.Helper() conn, err := net.Dial("unix", socket) if err != nil { t.Fatal(err) } t.Cleanup(func() { conn.Close() }) fmt.Fprint(conn, "GET /stream HTTP/1.1\r\nHost: hydrant\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n"+ "Sec-WebSocket-Version: 13\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\r\n") conn.SetReadDeadline(time.Now().Add(5 * time.Second)) status, err := bufio.NewReader(conn).ReadString('\n') if err != nil { t.Fatalf("read upgrade response: %v", err) } if !strings.Contains(status, "101") { t.Errorf("/stream didn't upgrade over the socket: %q", status) } return conn}
func TestEmbeddedHydrantServesOverUnixSocket(t *testing.T) { dir := shortTempDir(t) settings := offlineSettings(dir) socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:")
// the embedded config must come only from the settings map envDB := filepath.Join(dir, "from-env") t.Setenv("HYDRANT_DATABASE_PATH", envDB)
h := startHydrant(t, settings) client := unixClient(socket) waitReady(t, h, client)
if _, err := os.Stat(settings["HYDRANT_DATABASE_PATH"]); err != nil { t.Errorf("database not created at the configured path: %v", err) } if _, err := os.Stat(envDB); err == nil { t.Errorf("hydrant read HYDRANT_DATABASE_PATH from the process environment") }
info, err := os.Stat(socket) if err != nil { t.Fatal(err) } if mode := info.Mode().Perm(); mode != 0o600 { t.Errorf("socket mode is %o, want 600", mode) }
request(t, client, http.MethodPatch, "/filter", `{"mode":"filter","signals":["sh.tangled.repo"]}`) var filter struct { Mode string `json:"mode"` Signals []string `json:"signals"` } if err := json.Unmarshal([]byte(request(t, client, http.MethodGet, "/filter", "")), &filter); err != nil { t.Fatal(err) } if filter.Mode != "filter" || len(filter.Signals) != 1 || filter.Signals[0] != "sh.tangled.repo" { t.Errorf("filter didn't stick: %+v", filter) }
request(t, client, http.MethodGet, "/stream/head", "")
subscribe(t, socket)
t.Run("a second hydrant on the same database is refused", func(t *testing.T) { again := offlineSettings(dir) again["HYDRANT_API_BIND"] = "unix:" + filepath.Join(dir, "other.sock") _, err := Start(again) if err == nil || !strings.Contains(err.Error(), "Locked") { t.Fatalf("got %v, want the database lock to refuse it", err) } })
t.Run("a socket that is still served is not taken over", func(t *testing.T) { other := offlineSettings(shortTempDir(t)) other["HYDRANT_API_BIND"] = settings["HYDRANT_API_BIND"] h2 := startHydrant(t, other) select { case <-h2.Done(): if !strings.Contains(h2.Err().Error(), "listening on this socket") { t.Errorf("unexpected error: %v", h2.Err()) } case <-time.After(30 * time.Second): t.Fatal("second hydrant kept running on a socket that was in use") } // the first one must still be serving request(t, client, http.MethodGet, "/stats", "") })}
func TestShutdownLetsTheSameDatabaseStartAgain(t *testing.T) { dir := shortTempDir(t) settings := offlineSettings(dir) socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:") client := unixClient(socket)
h, err := Start(settings) if err != nil { t.Fatalf("start: %v", err) } waitReady(t, h, client) request(t, client, http.MethodPatch, "/filter", `{"mode":"filter","signals":["sh.tangled.repo"]}`) // a subscriber holds a stream thread, which has to end too subscribe(t, socket)
if err := h.Shutdown(10 * time.Second); err != nil { t.Fatalf("shutdown: %v", err) } if !errors.Is(h.Err(), ErrShutdown) { t.Errorf("Err after Shutdown is %v, want ErrShutdown", h.Err()) } if err := h.Shutdown(time.Second); err == nil { t.Error("a second Shutdown should be refused") }
// the database lock refuses this unless the first hydrant really let go again := startHydrant(t, settings) waitReady(t, again, client) if filter := request(t, client, http.MethodGet, "/filter", ""); !strings.Contains(filter, "sh.tangled.repo") { t.Errorf("the filter from before the shutdown is gone: %s", filter) }}
func TestShutdownAfterTheSubscriberHungUp(t *testing.T) { dir := shortTempDir(t) settings := offlineSettings(dir) socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:") h, err := Start(settings) if err != nil { t.Fatalf("start: %v", err) } waitReady(t, h, unixClient(socket))
// hydrant's goodbye to this socket is a write into a closed pipe, and on // linux that raises SIGPIPE on a rust thread, which go doesn't handle. the // sleep lets it get there before the shutdown cuts that task off subscribe(t, socket).Close() time.Sleep(200 * time.Millisecond) if err := h.Shutdown(10 * time.Second); err != nil { t.Fatalf("shutdown: %v", err) }}
func TestSocketModeIsConfigurable(t *testing.T) { dir := shortTempDir(t) settings := offlineSettings(dir) settings["HYDRANT_API_SOCKET_MODE"] = "0660" socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:") h := startHydrant(t, settings) waitReady(t, h, unixClient(socket))
info, err := os.Stat(socket) if err != nil { t.Fatal(err) } if mode := info.Mode().Perm(); mode != 0o660 { t.Fatalf("socket mode is %o, want 660", mode) }}
func TestNoApiKeepsRunning(t *testing.T) { dir := shortTempDir(t) settings := offlineSettings(dir) settings["HYDRANT_API_BIND"] = "none" h := startHydrant(t, settings) // with nothing to serve, a serve future that returned right away would end hydrant here select { case <-h.Done(): t.Fatalf("hydrant stopped without an api: %v", h.Err()) case <-time.After(2 * time.Second): } entries, err := os.ReadDir(dir) if err != nil { t.Fatal(err) } for _, e := range entries { if e.Type()&os.ModeSocket != 0 { t.Fatalf("no api should bind nothing, found socket %s", e.Name()) } }}
func TestStartRejectsBadSettings(t *testing.T) { for name, tc := range map[string]struct { mutate func(map[string]string) want string }{ "missing bind": {func(s map[string]string) { delete(s, "HYDRANT_API_BIND") }, "HYDRANT_API_BIND is required"}, "bad bind": {func(s map[string]string) { s["HYDRANT_API_BIND"] = "nope" }, "invalid HYDRANT_API_BIND"}, "bad setting": {func(s map[string]string) { s["HYDRANT_HISTORY_TTL"] = "soon" }, "HYDRANT_HISTORY_TTL"}, "bad mode": {func(s map[string]string) { s["HYDRANT_API_SOCKET_MODE"] = "0800" }, "HYDRANT_API_SOCKET_MODE"}, } { t.Run(name, func(t *testing.T) { settings := offlineSettings(shortTempDir(t)) tc.mutate(settings) _, err := Start(settings) if err == nil || !strings.Contains(err.Error(), tc.want) { t.Fatalf("got %v, want an error mentioning %q", err, tc.want) } }) }}