diff --git a/AGENTS.md b/AGENTS.md index d6fdfb6..cab2f16 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -70,6 +70,8 @@ See `examples/statusphere.rs` for a usage example. Non-rust programs embed hydrant through the C api in `ffi/` (`hydrant-ffi`, a workspace member built as a staticlib), which starts hydrant from the binary's `HYDRANT_*` settings via `Config::from_lookup` and leaves all data access to the HTTP API, ideally on a `unix:` bind. `ffi/go` wraps it for go. See `docs/embedding.md`. +`Hydrant::shutdown` waits for the database to close, so every std thread hydrant starts has to end once `AppState::stop` is set: timer loops sleep on `state.stop.wait(..)` instead of `thread::sleep`, and channel workers check `state.stop.is_set()` per message. A thread that doesn't keeps the database open and makes shutdown time out. + ## General conventions ### Correctness over convenience diff --git a/docs/embedding.md b/docs/embedding.md index 6645450..f8bcbde 100644 --- a/docs/embedding.md +++ b/docs/embedding.md @@ -33,9 +33,10 @@ the socket is created owner-only (`0600`), so only the embedding program's user ## lifecycle - `hydrant_start` returns once the database is open and hydrant is starting. the api binds in the background, so poll it (eg. `GET /stats`) until it answers. -- `hydrant_wait` blocks until hydrant stops, which only happens when it fails (for example when its socket is already in use), and gives the reason. -- there is no stop: hydrant's background workers have no shutdown path, so it runs until the process exits. when `hydrant_wait` returns, treat it as fatal and exit, since some of that background work outlives the failure. -- only one hydrant can use a database at a time, a second `hydrant_start` on it fails. +- `hydrant_wait` blocks until hydrant stops: when it fails (for example when its socket is already in use) it gives the reason, and after `hydrant_shutdown` it returns 0. +- `hydrant_shutdown` stops hydrant and waits for its database to close, which gives its memory back. anything it was in the middle of is cut off like a crash, and the next start picks up from its saved cursors. a hydrant that failed still holds its database until it's shut down. +- `hydrant_free` frees the handle once nothing is blocked in `hydrant_wait` on it. +- only one hydrant can use a database at a time, a second `hydrant_start` on it fails until the first one is shut down. ## go @@ -55,8 +56,15 @@ if err != nil { } go func() { <-h.Done() - log.Fatalf("hydrant stopped: %v", h.Err()) + if !errors.Is(h.Err(), hydrant.ErrShutdown) { + log.Printf("hydrant failed: %v", h.Err()) + } }() + +// later, eg. once it's done backfilling +if err := h.Shutdown(30 * time.Second); err != nil { + return err +} ``` then reach the api with an `http.Client` whose transport dials the socket. `nu tests/ffi_embed.nu` builds the library and runs the go package's tests, which double as an example. diff --git a/ffi/go/hydrant.go b/ffi/go/hydrant.go index 362e921..ade4eee 100644 --- a/ffi/go/hydrant.go +++ b/ffi/go/hydrant.go @@ -19,14 +19,18 @@ import ( "encoding/json" "errors" "fmt" + "time" "unsafe" ) -// Hydrant is a hydrant running in this process. it can't be stopped, because hydrant's -// background threads have no shutdown path, so it runs until the process exits. +// ErrShutdown is what Err reports after Shutdown stopped hydrant. +var ErrShutdown = errors.New("hydrant was shut down") + +// Hydrant is a hydrant running in this process, until Shutdown. type Hydrant struct { - done chan struct{} - err error + handle *C.hydrant_t + done chan struct{} + err error } // Start runs hydrant with the settings its binary reads from the environment, like @@ -47,25 +51,46 @@ func Start(settings map[string]string) (*Hydrant, error) { return nil, takeErr(cErr) } - h := &Hydrant{done: make(chan struct{})} + h := &Hydrant{handle: handle, done: make(chan struct{})} go func() { var cErr *C.char if C.hydrant_wait(handle, &cErr) != 0 { h.err = takeErr(cErr) } else { - h.err = errors.New("hydrant exited") + h.err = ErrShutdown } close(h.done) }() return h, nil } -// Done is closed once hydrant stops, which only happens when it fails. treat it as fatal and -// exit, since some of hydrant's background work outlives the failure and can't be stopped. +// Done is closed once hydrant stops, because it failed or Shutdown stopped it. a failed +// hydrant still holds its database until Shutdown. func (h *Hydrant) Done() <-chan struct{} { return h.done } +// Shutdown stops hydrant and waits up to timeout for its database to close, so its memory +// is given back and Start can open the same database again. whatever hydrant was in the +// middle of is cut off like a crash, which it recovers from on the next start. call it +// once, from one goroutine, also after a failure. it errors if the database is still open +// after timeout, and the Hydrant can't be used again either way. +func (h *Hydrant) Shutdown(timeout time.Duration) error { + if h.handle == nil { + return errors.New("hydrant was already shut down") + } + var cErr *C.char + var err error + if C.hydrant_shutdown(h.handle, C.uint64_t(timeout.Milliseconds()), &cErr) != 0 { + err = takeErr(cErr) + } + // the wait goroutine still uses the handle until it sees the stop + <-h.done + C.hydrant_free(h.handle) + h.handle = nil + return err +} + // Err says why hydrant stopped. it's nil while hydrant is running. func (h *Hydrant) Err() error { select { diff --git a/ffi/go/hydrant_test.go b/ffi/go/hydrant_test.go index af9a182..4f6a79b 100644 --- a/ffi/go/hydrant_test.go +++ b/ffi/go/hydrant_test.go @@ -4,6 +4,7 @@ import ( "bufio" "context" "encoding/json" + "errors" "fmt" "io" "net" @@ -39,6 +40,21 @@ func shortTempDir(t *testing.T) string { return dir } +// starts hydrant and shuts it down when the test ends, before its dir is removed +func 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, @@ -92,6 +108,26 @@ func request(t *testing.T, client *http.Client, method, path, body string) strin return string(out) } +// opens a /stream websocket and leaves it open until the test ends +func subscribe(t *testing.T, socket string) { + 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) + } +} + func TestEmbeddedHydrantServesOverUnixSocket(t *testing.T) { dir := shortTempDir(t) settings := offlineSettings(dir) @@ -101,10 +137,7 @@ func TestEmbeddedHydrantServesOverUnixSocket(t *testing.T) { envDB := filepath.Join(dir, "from-env") t.Setenv("HYDRANT_DATABASE_PATH", envDB) - h, err := Start(settings) - if err != nil { - t.Fatalf("start: %v", err) - } + h := startHydrant(t, settings) client := unixClient(socket) waitReady(t, h, client) @@ -137,21 +170,7 @@ func TestEmbeddedHydrantServesOverUnixSocket(t *testing.T) { request(t, client, http.MethodGet, "/stream/head", "") - conn, err := net.Dial("unix", socket) - if err != nil { - t.Fatal(err) - } - defer 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) - } + subscribe(t, socket) t.Run("a second hydrant on the same database is refused", func(t *testing.T) { again := offlineSettings(dir) @@ -163,18 +182,9 @@ func TestEmbeddedHydrantServesOverUnixSocket(t *testing.T) { }) t.Run("a socket that is still served is not taken over", func(t *testing.T) { - // not shortTempDir: the refused hydrant's workers keep using its database after it - // fails, and deleting it underneath them only makes noise - otherDir, err := os.MkdirTemp("/tmp", "hydrant-ffi-") - if err != nil { - t.Fatal(err) - } - other := offlineSettings(otherDir) + other := offlineSettings(shortTempDir(t)) other["HYDRANT_API_BIND"] = settings["HYDRANT_API_BIND"] - h2, err := Start(other) - if err != nil { - t.Fatalf("start: %v", err) - } + h2 := startHydrant(t, other) select { case <-h2.Done(): if !strings.Contains(h2.Err().Error(), "listening on this socket") { @@ -188,15 +198,45 @@ func TestEmbeddedHydrantServesOverUnixSocket(t *testing.T) { }) } -func TestSocketModeIsConfigurable(t *testing.T) { +func TestShutdownLetsTheSameDatabaseStartAgain(t *testing.T) { dir := shortTempDir(t) settings := offlineSettings(dir) - settings["HYDRANT_API_SOCKET_MODE"] = "0660" socket := strings.TrimPrefix(settings["HYDRANT_API_BIND"], "unix:") + client := unixClient(socket) + h, err := Start(settings) if err != nil { - t.Fatal(err) + 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 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) @@ -212,10 +252,7 @@ func TestNoApiKeepsRunning(t *testing.T) { dir := shortTempDir(t) settings := offlineSettings(dir) settings["HYDRANT_API_BIND"] = "none" - h, err := Start(settings) - if err != nil { - t.Fatal(err) - } + h := startHydrant(t, settings) // with nothing to serve, a serve future that returned right away would end hydrant here select { case <-h.Done(): diff --git a/ffi/include/hydrant.h b/ffi/include/hydrant.h index 9665c98..3106618 100644 --- a/ffi/include/hydrant.h +++ b/ffi/include/hydrant.h @@ -4,6 +4,8 @@ /* run hydrant inside another program. hydrant serves its usual http api, ideally on a * `unix:` bind, and the embedder talks to it over that. see ffi/src/lib.rs for details. */ +#include + #ifdef __cplusplus extern "C" { #endif @@ -12,15 +14,22 @@ typedef struct HydrantHandle hydrant_t; /* starts hydrant from a json object of the HYDRANT_* settings the binary reads from its * environment, plus RUST_LOG. HYDRANT_API_BIND is required, and `none` runs without an api. - * returns NULL and sets *err on failure. the handle lives until the process exits, there is - * no stop. */ + * returns NULL and sets *err on failure. */ hydrant_t *hydrant_start(const char *settings_json, char **err); -/* blocks until hydrant stops, which only happens when it fails: returns -1 with *err set. - * returns 0 if it exited cleanly. treat either as fatal and exit, since some of hydrant's - * background work outlives the failure and can't be stopped. */ +/* blocks until hydrant stops. returns -1 with *err set when it failed, or 0 once hydrant_shutdown + * stopped it. a failed hydrant still holds its database until hydrant_shutdown. */ int hydrant_wait(const hydrant_t *handle, char **err); +/* stops hydrant and waits up to timeout_ms for its database to close, so its memory is + * given back and hydrant_start can open the same database again. returns -1 with *err set + * if it's still open after that, or if it was already shut down. */ +int hydrant_shutdown(const hydrant_t *handle, uint64_t timeout_ms, char **err); + +/* frees a handle, telling it to stop first if hydrant_shutdown wasn't called. nothing may + * still be blocked in hydrant_wait on it. */ +void hydrant_free(hydrant_t *handle); + /* frees an error string from this library. */ void hydrant_free_string(char *s); diff --git a/ffi/src/lib.rs b/ffi/src/lib.rs index d7b3054..766ef94 100644 --- a/ffi/src/lib.rs +++ b/ffi/src/lib.rs @@ -1,8 +1,8 @@ //! a C api for running hydrant inside another program. //! //! hydrant keeps serving its usual http api, and the embedder talks to it over that, ideally -//! on a `unix:` bind. the C side only starts it and learns when it stops, so no events or -//! callbacks ever cross the boundary. +//! on a `unix:` bind. the C side only starts and stops it and learns when it stopped, so no +//! events or callbacks ever cross the boundary. use std::any::Any; use std::collections::HashMap; @@ -10,17 +10,31 @@ use std::ffi::{CStr, CString, c_char, c_int}; use std::io::IsTerminal; use std::panic::{AssertUnwindSafe, catch_unwind}; use std::sync::{Arc, Condvar, Mutex, PoisonError}; +use std::time::Duration; use hydrant::config::Config; use hydrant::control::{ApiBinds, Hydrant}; use hydrant::deps::futures::FutureExt; -/// a running hydrant. it has no way to stop, since hydrant's background threads have no -/// shutdown path, so it lives until the process exits. +/// a hydrant started by `hydrant_start`, until `hydrant_free`. pub struct HydrantHandle { exit: Arc, - // never read, but dropping it would take hydrant's tasks down with it - _runtime: tokio::runtime::Runtime, + // taken by the first `hydrant_shutdown` + running: Mutex>, +} + +struct Running { + runtime: tokio::runtime::Runtime, + hydrant: Hydrant, +} + +impl Running { + fn shutdown(self, timeout: Duration) -> Result<(), String> { + // not a plain drop, because that waits on blocking tasks, and some of those wait on + // threads only `Hydrant::shutdown` stops + self.runtime.shutdown_background(); + self.hydrant.shutdown(timeout).map_err(report) + } } #[derive(Default)] @@ -29,6 +43,18 @@ struct Exit { changed: Condvar, } +impl Exit { + /// the first result is the one `hydrant_wait` reports, so a failure isn't + /// overwritten by the shutdown that follows it + fn finish(&self, result: Result<(), String>) { + let mut current = self.result.lock().unwrap_or_else(PoisonError::into_inner); + if current.is_none() { + *current = Some(result); + self.changed.notify_all(); + } + } +} + /// starts hydrant from a json object of the `HYDRANT_*` settings the binary reads from its /// environment, plus `RUST_LOG`. the process environment and `.env` are ignored. /// `HYDRANT_API_BIND` is required, since the binary's default would expose the api on every @@ -65,22 +91,20 @@ pub unsafe extern "C" fn hydrant_start( } } -/// blocks until hydrant stops, which only happens when it fails, and returns -1 with `*err` -/// set to why. returns 0 if hydrant somehow exited cleanly. safe to call from several threads, -/// and again after it returned. -/// -/// treat a return as fatal and exit: some of hydrant's background work outlives the failure -/// and can't be stopped, so the process can't just start another one on the same database. +/// blocks until hydrant stops. returns -1 with `*err` set to why when it failed, or 0 once +/// `hydrant_shutdown` stopped it. safe to call from several threads, and again after it +/// returned. a failed hydrant still holds its database until `hydrant_shutdown`. /// /// # Safety -/// `handle` must come from `hydrant_start`, and `err` must be null or writable. +/// `handle` must come from `hydrant_start` and not be freed yet, and `err` must be null or +/// writable. #[unsafe(no_mangle)] pub unsafe extern "C" fn hydrant_wait( handle: *const HydrantHandle, err: *mut *mut c_char, ) -> c_int { let result = guard(|| { - // SAFETY: the caller promises a handle from `hydrant_start`, and handles are never freed + // SAFETY: the caller promises a live handle from `hydrant_start` let handle = unsafe { handle.as_ref() }.ok_or("handle is null")?; let exit = &handle.exit; let result = exit.result.lock().unwrap_or_else(PoisonError::into_inner); @@ -102,6 +126,68 @@ pub unsafe extern "C" fn hydrant_wait( } } +/// stops hydrant and waits up to `timeout_ms` for its database to close, which gives its +/// memory back and lets `hydrant_start` open the same database again. `hydrant_wait` returns +/// once this does. whatever hydrant was in the middle of is cut off like a crash, which it +/// recovers from on the next start. +/// +/// returns -1 with `*err` set if the database is still open after the timeout, or if it was +/// already shut down. the handle stays valid until `hydrant_free` either way. +/// +/// # Safety +/// `handle` must come from `hydrant_start` and not be freed yet, and `err` must be null or +/// writable. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn hydrant_shutdown( + handle: *const HydrantHandle, + timeout_ms: u64, + err: *mut *mut c_char, +) -> c_int { + let result = guard(|| { + // SAFETY: the caller promises a live handle from `hydrant_start` + let handle = unsafe { handle.as_ref() }.ok_or("handle is null")?; + let running = handle + .running + .lock() + .unwrap_or_else(PoisonError::into_inner) + .take() + .ok_or("hydrant was already shut down")?; + let result = running.shutdown(Duration::from_millis(timeout_ms)); + handle.exit.finish(Ok(())); + result + }); + match result { + Ok(()) => 0, + Err(e) => { + // SAFETY: the caller promises `err` is null or writable + unsafe { set_err(err, e) }; + -1 + } + } +} + +/// frees a handle. one that wasn't shut down is told to stop without waiting for it, so its +/// database closes a moment later. +/// +/// # Safety +/// `handle` must be null or come from `hydrant_start` and not be freed yet, and nothing may +/// still be blocked in `hydrant_wait` on it. +#[unsafe(no_mangle)] +pub unsafe extern "C" fn hydrant_free(handle: *mut HydrantHandle) { + if handle.is_null() { + return; + } + // SAFETY: the caller promises it came from `Box::into_raw` in `hydrant_start` + let handle = unsafe { Box::from_raw(handle) }; + let running = handle + .running + .into_inner() + .unwrap_or_else(PoisonError::into_inner); + if let Some(running) = running { + let _ = guard(|| running.shutdown(Duration::ZERO)); + } +} + /// frees an error string returned by this library. /// /// # Safety @@ -138,6 +224,7 @@ fn start(settings_json: &str) -> Result { let exit = Arc::new(Exit::default()); runtime.spawn({ let exit = exit.clone(); + let hydrant = hydrant.clone(); async move { let serve = binds .map(|binds| hydrant.serve(binds).boxed()) @@ -157,15 +244,13 @@ fn start(settings_json: &str) -> Result { panic_message(&*panic) )) }); - *exit.result.lock().unwrap_or_else(PoisonError::into_inner) = - Some(result.map_err(report)); - exit.changed.notify_all(); + exit.finish(result.map_err(report)); } }); Ok(HydrantHandle { exit, - _runtime: runtime, + running: Mutex::new(Some(Running { runtime, hydrant })), }) } diff --git a/src/backfill/manager.rs b/src/backfill/manager.rs index 03514aa..5c715da 100644 --- a/src/backfill/manager.rs +++ b/src/backfill/manager.rs @@ -100,9 +100,7 @@ pub fn queue_gone_backfills(state: &Arc) -> Result<()> { pub fn retry_worker(state: Arc) { let db = &state.db; info!("retry worker started"); - loop { - std::thread::sleep(Duration::from_secs(60)); - + while !state.stop.wait(Duration::from_secs(60)) { let now = chrono::Utc::now().timestamp(); let mut candidates = Vec::new(); for guard in db.indexer.resync.iter() { diff --git a/src/control/hydrant.rs b/src/control/hydrant.rs index f053206..219555c 100644 --- a/src/control/hydrant.rs +++ b/src/control/hydrant.rs @@ -3,6 +3,7 @@ use miette::{IntoDiagnostic, Result}; use std::future::Future; use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; use tokio::sync::watch; use tracing::{debug, error, info, warn}; use url::Url; @@ -209,6 +210,35 @@ impl Hydrant { async move { crate::api::serve_debug(state, port).await } } + /// stops hydrant's own threads and waits until its database is closed, which + /// gives its memory back and lets the same path be opened again. + /// + /// the tasks [`run`](Self::run) and [`serve`](Self::serve) spawned live on your + /// tokio runtime and can't be stopped from here, so shut that runtime down first + /// (`shutdown_background`, since dropping it would wait on threads this stops), + /// and drop every other clone of this handle and any open stream. whatever they + /// were doing is cut off like a crash, which hydrant recovers from on the next + /// start. + /// + /// fails if the database is still open after `timeout`, which means something + /// still holds on to it. + pub fn shutdown(self, timeout: Duration) -> Result<()> { + self.state.stop.trigger(); + let db = Arc::downgrade(&self.state.db.inner); + drop(self); + + let deadline = Instant::now() + timeout; + while db.strong_count() > 0 { + if Instant::now() >= deadline { + miette::bail!( + "the database is still open after {timeout:?}, something still holds a hydrant handle" + ); + } + std::thread::sleep(Duration::from_millis(10)); + } + Ok(()) + } + /// iterate sequentially over the current head body of every record in /// `collection`, ordered by (repo, rkey). /// @@ -412,4 +442,51 @@ mod tests { Ok(()) } + + /// starts a hydrant that never touches the network on its own runtime. + fn start_offline(path: &std::path::Path) -> (tokio::runtime::Runtime, Hydrant) { + let runtime = tokio::runtime::Runtime::new().unwrap(); + let config = Config { + relays: Vec::new(), + seed_hosts: Vec::new(), + crawler_sources: Vec::new(), + enable_crawler: Some(false), + ..test_config(path) + }; + let hydrant = runtime.block_on(Hydrant::new(config)).unwrap(); + runtime.spawn({ + let hydrant = hydrant.clone(); + async move { hydrant.run()?.await } + }); + // long enough for run to have spawned its threads + std::thread::sleep(Duration::from_millis(300)); + (runtime, hydrant) + } + + #[test] + fn shutdown_closes_the_database_for_the_next_start() { + let tmp = tempdir().unwrap(); + let (runtime, hydrant) = start_offline(tmp.path()); + runtime.shutdown_background(); + hydrant.shutdown(Duration::from_secs(10)).unwrap(); + + // fjall locks the database, so this only opens if the first one let go + let (runtime, hydrant) = start_offline(tmp.path()); + runtime.shutdown_background(); + hydrant.shutdown(Duration::from_secs(10)).unwrap(); + } + + #[test] + fn shutdown_fails_while_a_clone_holds_the_database() { + let tmp = tempdir().unwrap(); + let (runtime, hydrant) = start_offline(tmp.path()); + runtime.shutdown_background(); + let forgotten = hydrant.clone(); + + let err = hydrant.shutdown(Duration::from_millis(200)).unwrap_err(); + assert!(err.to_string().contains("still open"), "{err}"); + + // the clone still works and is the last one, so it can finish the job + forgotten.shutdown(Duration::from_secs(10)).unwrap(); + } } diff --git a/src/control/hydrant/run.rs b/src/control/hydrant/run.rs index 6dfd0c4..d76b20a 100644 --- a/src/control/hydrant/run.rs +++ b/src/control/hydrant/run.rs @@ -123,48 +123,16 @@ impl Hydrant { .into_diagnostic()?; } - // 8. cursor / counts persist thread + // 8. cursor / counts persist thread. it persists once more on the way + // out, so a shutdown keeps everything up to that point std::thread::spawn({ let state = state.clone(); let persist_interval = config.cursor_save_interval; - move || loop { - std::thread::sleep(persist_interval); - - state.firehose_cursors.iter_sync(|relay, cursor| { - let seq = cursor.load(Ordering::SeqCst); - if seq > 0 - && let Err(e) = db::set_firehose_cursor(&state.db, relay, seq) - { - error!(relay = %relay, err = %e, "failed to save cursor"); - db::check_poisoned_report(&e); - } - true - }); - - let checkpoint_watermark = match state.db.checkpoint_count_deltas() { - Ok(watermark) => watermark, - Err(e) => { - error!(err = %e, "failed to checkpoint count deltas"); - db::check_poisoned_report(&e); - None - } - }; - - if let Err(e) = state.db.persist() { - error!(err = %e, "db persist failed"); - db::check_poisoned_report(&e); - } else { - let watermark = checkpoint_watermark - .map(Ok) - .unwrap_or_else(|| db::load_count_delta_watermark(&state.db)); - match watermark { - Ok(watermark) => state.db.mark_count_checkpoint_persisted(watermark), - Err(e) => { - error!(err = %e, "failed to load durable count checkpoint watermark"); - db::check_poisoned_report(&e); - } - } + move || { + while !state.stop.wait(persist_interval) { + persist(&state); } + persist(&state); } }); @@ -579,3 +547,42 @@ impl Hydrant { Ok(fut) } } + +/// saves the firehose cursors and count checkpoints, then fsyncs the db. +fn persist(state: &AppState) { + state.firehose_cursors.iter_sync(|relay, cursor| { + let seq = cursor.load(Ordering::SeqCst); + if seq > 0 + && let Err(e) = db::set_firehose_cursor(&state.db, relay, seq) + { + error!(relay = %relay, err = %e, "failed to save cursor"); + db::check_poisoned_report(&e); + } + true + }); + + let checkpoint_watermark = match state.db.checkpoint_count_deltas() { + Ok(watermark) => watermark, + Err(e) => { + error!(err = %e, "failed to checkpoint count deltas"); + db::check_poisoned_report(&e); + None + } + }; + + if let Err(e) = state.db.persist() { + error!(err = %e, "db persist failed"); + db::check_poisoned_report(&e); + } else { + let watermark = checkpoint_watermark + .map(Ok) + .unwrap_or_else(|| db::load_count_delta_watermark(&state.db)); + match watermark { + Ok(watermark) => state.db.mark_count_checkpoint_persisted(watermark), + Err(e) => { + error!(err = %e, "failed to load durable count checkpoint watermark"); + db::check_poisoned_report(&e); + } + } + } +} diff --git a/src/db/ephemeral.rs b/src/db/ephemeral.rs index 9f30397..dffd23d 100644 --- a/src/db/ephemeral.rs +++ b/src/db/ephemeral.rs @@ -11,8 +11,7 @@ use { #[cfg(feature = "indexer_stream")] pub fn ephemeral_ttl_worker(state: Arc) { info!("ephemeral TTL worker started"); - loop { - std::thread::sleep(Duration::from_secs(60)); + while !state.stop.wait(Duration::from_secs(60)) { if let Err(e) = ephemeral_ttl_tick(&state.db, &state.ephemeral_ttl) { error!(err = %e, "ephemeral TTL tick failed"); } @@ -26,8 +25,7 @@ pub fn ephemeral_ttl_worker(state: Arc) { #[cfg(feature = "relay")] pub fn relay_events_ttl_worker(state: Arc) { info!("relay events TTL worker started"); - loop { - std::thread::sleep(Duration::from_secs(60)); + while !state.stop.wait(Duration::from_secs(60)) { if let Err(e) = relay_events_ttl_tick(&state.db, &state.ephemeral_ttl) { error!(err = %e, "relay events TTL tick failed"); } diff --git a/src/ingest/indexer/shard.rs b/src/ingest/indexer/shard.rs index 75657a4..6e2c1f0 100644 --- a/src/ingest/indexer/shard.rs +++ b/src/ingest/indexer/shard.rs @@ -44,6 +44,11 @@ impl FirehoseWorker { let mut lifecycle_transitions = Vec::new(); while let Some(msg) = rx.blocking_recv() { + // dropping what's left is the same as a crash here, which a + // restart already recovers from + if state.stop.is_set() { + break; + } let txn = Txn::new(&state.db); lifecycle_transitions.clear(); diff --git a/src/ingest/relay/worker.rs b/src/ingest/relay/worker.rs index dcfdf2b..dba0d3c 100644 --- a/src/ingest/relay/worker.rs +++ b/src/ingest/relay/worker.rs @@ -122,6 +122,11 @@ impl RelayWorker { }; while let Some(msg) = rx.blocking_recv() { + // past this the runtime may be gone, and the firehose cursor + // hasn't moved for what's left, so it's read again next start + if state.stop.is_set() { + break; + } let message_started = StatsInstant::now(); let IngestMessage::Firehose { url, is_pds, msg } = msg; if let SubscribeReposMessage::Info(inf) = msg { diff --git a/src/state.rs b/src/state.rs index 76c897f..727c840 100644 --- a/src/state.rs +++ b/src/state.rs @@ -21,7 +21,7 @@ use crate::{ pds_meta::{PdsMeta, PdsMetaHandle, TierPolicy, new_handle as new_pds_handle}, resolver::Resolver, types::DocIdentity, - util::throttle::Throttler, + util::{stop::Stop, throttle::Throttler}, }; const IDENTITY_RECONCILE_COOL_OFF: Duration = Duration::from_secs(30); @@ -71,6 +71,7 @@ pub struct AppState { pub max_car_body_bytes: usize, pub get_repo_semaphore: Semaphore, pub maintenance_timeout: Duration, + pub(crate) stop: Stop, } impl AppState { @@ -171,6 +172,7 @@ impl AppState { max_car_body_bytes: config.max_car_body_bytes, get_repo_semaphore: Semaphore::new(config.get_repo_concurrency_limit), maintenance_timeout: config.maintenance_timeout, + stop: Stop::default(), // public backfill is direct-only; one socket pool must not inflate // per-PDS pacing as if proxies provided independent egress IPs. throttler: Throttler::new(1, config.per_pds_concurrency), diff --git a/src/util/mod.rs b/src/util/mod.rs index f4adb84..94548a9 100644 --- a/src/util/mod.rs +++ b/src/util/mod.rs @@ -12,6 +12,7 @@ use url::Url; use crate::{db::types::DidKey, types::RepoStatus}; +pub mod stop; pub mod throttle; #[allow(dead_code)] diff --git a/src/util/stop.rs b/src/util/stop.rs new file mode 100644 index 0000000..faab2ad --- /dev/null +++ b/src/util/stop.rs @@ -0,0 +1,72 @@ +use std::sync::atomic::{AtomicBool, Ordering}; +use std::time::{Duration, Instant}; + +use parking_lot::{Condvar, Mutex}; + +/// tells hydrant's own threads to wind down. they sleep on it instead of +/// `thread::sleep`, so a shutdown doesn't wait out a whole interval. +#[derive(Default)] +pub struct Stop { + // the ingest shards check this per message, so it stays a plain load + stopped: AtomicBool, + lock: Mutex<()>, + changed: Condvar, +} + +impl Stop { + pub fn trigger(&self) { + // set under the lock, or a waiter that just checked could miss the notify + let _lock = self.lock.lock(); + self.stopped.store(true, Ordering::Release); + self.changed.notify_all(); + } + + pub fn is_set(&self) -> bool { + self.stopped.load(Ordering::Acquire) + } + + /// sleeps for `timeout`, or less if stopped meanwhile. true means stop. + pub fn wait(&self, timeout: Duration) -> bool { + let deadline = Instant::now() + timeout; + let mut lock = self.lock.lock(); + while !self.is_set() { + if self.changed.wait_until(&mut lock, deadline).timed_out() { + break; + } + } + self.is_set() + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + + #[test] + fn wait_times_out_without_a_trigger() { + let stop = Stop::default(); + let started = Instant::now(); + assert!(!stop.wait(Duration::from_millis(20))); + assert!(started.elapsed() >= Duration::from_millis(20)); + } + + #[test] + fn trigger_wakes_a_waiting_thread_early() { + let stop = Arc::new(Stop::default()); + let waiter = std::thread::spawn({ + let stop = stop.clone(); + move || { + let started = Instant::now(); + (stop.wait(Duration::from_secs(60)), started.elapsed()) + } + }); + std::thread::sleep(Duration::from_millis(20)); + stop.trigger(); + let (stopped, waited) = waiter.join().unwrap(); + assert!(stopped); + assert!(waited < Duration::from_secs(5)); + // and stays set for anyone who checks later + assert!(stop.wait(Duration::from_secs(60))); + } +}