diff --git a/Cargo.lock b/Cargo.lock index 153b2676..9fb9ad16 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -861,19 +861,28 @@ dependencies = [ [[package]] name = "extenddb" version = "0.1.2" +dependencies = [ + "anyhow", + "extenddb-app", + "extenddb-storage", + "extenddb-storage-postgres", +] + +[[package]] +name = "extenddb-app" +version = "0.1.2" dependencies = [ "anyhow", "base64 0.22.1", "clap", - "config", "daemonize", "extenddb-auth", "extenddb-cache", + "extenddb-config", "extenddb-core", "extenddb-engine", "extenddb-server", "extenddb-storage", - "extenddb-storage-postgres", "libc", "rcgen", "rustls", @@ -881,7 +890,6 @@ dependencies = [ "serde", "serde_json", "sqlx", - "syslog-tracing", "time", "tokio", "toml", @@ -920,6 +928,19 @@ dependencies = [ "tracing", ] +[[package]] +name = "extenddb-config" +version = "0.1.2" +dependencies = [ + "anyhow", + "config", + "extenddb-core", + "extenddb-storage", + "serde", + "toml", + "tracing", +] + [[package]] name = "extenddb-core" version = "0.1.2" @@ -966,21 +987,25 @@ dependencies = [ "crc32fast", "extenddb-auth", "extenddb-cache", + "extenddb-config", "extenddb-core", "extenddb-engine", "extenddb-storage", "futures", "hyper", + "libc", "metrics", "rand 0.9.4", "rustls", "serde", "serde_json", + "syslog-tracing", "time", "tokio", "tower", "tower-http", "tracing", + "tracing-subscriber", "uuid", ] @@ -996,12 +1021,12 @@ dependencies = [ "extenddb-auth", "extenddb-core", "futures", - "inventory", "rand 0.9.4", "serde_json", "thiserror", "time", "tokio", + "tokio-util", "toml", "tracing", "tracing-subscriber", @@ -1021,7 +1046,6 @@ dependencies = [ "extenddb-core", "extenddb-storage", "futures", - "inventory", "rand 0.9.4", "serde", "serde_json", @@ -1568,15 +1592,6 @@ dependencies = [ "generic-array", ] -[[package]] -name = "inventory" -version = "0.3.24" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a4f0c30c76f2f4ccee3fe55a2435f691ca00c0e4bd87abe4f4a851b1d4dac39b" -dependencies = [ - "rustversion", -] - [[package]] name = "is_terminal_polyfill" version = "1.70.2" diff --git a/Cargo.toml b/Cargo.toml index c357d69b..76f909b6 100755 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,9 +7,11 @@ members = [ "crates/cache", "crates/engine", "crates/storage", + "crates/config", "crates/storage-postgres", "crates/auth", "crates/server", + "crates/app", "crates/bin", ] @@ -25,9 +27,11 @@ extenddb-core = { path = "crates/core" } extenddb-cache = { path = "crates/cache" } extenddb-engine = { path = "crates/engine" } extenddb-storage = { path = "crates/storage" } +extenddb-config = { path = "crates/config" } extenddb-storage-postgres = { path = "crates/storage-postgres" } extenddb-auth = { path = "crates/auth" } extenddb-server = { path = "crates/server" } +extenddb-app = { path = "crates/app" } # Serialization serde = { version = "1", features = ["derive"] } @@ -40,6 +44,7 @@ anyhow = "1" # Async tokio = { version = "1", features = ["full"] } +tokio-util = { version = "0.7" } async-trait = "0.1" futures = "0.3" @@ -50,9 +55,6 @@ tower-http = { version = "0.6", features = ["compression-gzip", "cors", "set-hea hyper = { version = "1" } urlencoding = { version = "2.1" } -# Database plugin registry -inventory = "0.3" - # Caching moka = { version = "0.12", features = ["future"] } diff --git a/crates/app/Cargo.toml b/crates/app/Cargo.toml new file mode 100644 index 00000000..cdd0058e --- /dev/null +++ b/crates/app/Cargo.toml @@ -0,0 +1,33 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 +[package] +name = "extenddb-app" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true + +[dependencies] +extenddb-auth = { workspace = true } +extenddb-cache = { workspace = true } +extenddb-core = { workspace = true } +extenddb-engine = { workspace = true } +extenddb-storage = { workspace = true } +extenddb-config = { workspace = true } +extenddb-server = { workspace = true } +tokio = { workspace = true } +anyhow = { workspace = true } +clap = { workspace = true } +daemonize = { workspace = true } +serde = { workspace = true } +serde_json = { workspace = true } +sqlx = { workspace = true } +time = { workspace = true } +toml = { workspace = true } +tracing = { workspace = true } +tracing-subscriber = { workspace = true } +base64 = { workspace = true } +libc = { workspace = true } +rcgen = { workspace = true } +rustls = { workspace = true } +rustls-pemfile = { workspace = true } diff --git a/crates/bin/src/cmd_catalog_check.rs b/crates/app/src/cmd_catalog_check.rs similarity index 98% rename from crates/bin/src/cmd_catalog_check.rs rename to crates/app/src/cmd_catalog_check.rs index a93e6f88..91744c1d 100755 --- a/crates/bin/src/cmd_catalog_check.rs +++ b/crates/app/src/cmd_catalog_check.rs @@ -17,7 +17,7 @@ use std::collections::HashSet; use clap::Args; use sqlx::postgres::PgPoolOptions; -use crate::config; +use extenddb_config as config; #[derive(Args)] pub struct CatalogCheckArgs { @@ -47,7 +47,7 @@ pub async fn run(args: CatalogCheckArgs) -> anyhow::Result<()> { let run_dir = config::expand_tilde(&app_config.server.run_dir); // Refuse to run while server is up. - let pid_path = crate::serve_helpers::pid_file_path(&run_dir, port); + let pid_path = extenddb_config::pid_file_path(&run_dir, port); if let Ok(contents) = std::fs::read_to_string(&pid_path) && let Ok(pid) = contents.trim().parse::() && crate::util::is_process_alive(pid) diff --git a/crates/bin/src/cmd_destroy.rs b/crates/app/src/cmd_destroy.rs similarity index 89% rename from crates/bin/src/cmd_destroy.rs rename to crates/app/src/cmd_destroy.rs index 6a3e9bd5..ab4d99e9 100755 --- a/crates/bin/src/cmd_destroy.rs +++ b/crates/app/src/cmd_destroy.rs @@ -7,7 +7,7 @@ use clap::Args; -use crate::config; +use extenddb_config as config; #[derive(Args)] #[allow(clippy::doc_markdown)] // Clap help text, not rustdoc @@ -49,7 +49,7 @@ pub async fn run(args: DestroyArgs) -> anyhow::Result<()> { // Create bootstrap store for catalog queries and database teardown. let bootstrap = - extenddb_storage::bootstrapper::create_bootstrapper(backend, &args.config, &cli_args).await; + extenddb_storage::bootstrapper::create_bootstrapper(&args.config, &cli_args).await; let mut data_db = String::new(); @@ -96,16 +96,15 @@ pub async fn run(args: DestroyArgs) -> anyhow::Result<()> { // connects to the `postgres` database, so we can reuse it. if !data_db.is_empty() { // Defense-in-depth: validate even though this came from the catalog. - config::validate_identifier(backend, &data_db, "data database name")?; + config::validate_identifier(&data_db, "data database name")?; } // Reconnect as admin for DDL operations (the catalog pool must be dropped // before we can DROP DATABASE). drop(bootstrap); - let bootstrap = - extenddb_storage::bootstrapper::create_bootstrapper(backend, &args.config, &cli_args) - .await - .map_err(|e| anyhow::anyhow!("Cannot connect as admin: {e:?}"))?; + let bootstrap = extenddb_storage::bootstrapper::create_bootstrapper(&args.config, &cli_args) + .await + .map_err(|e| anyhow::anyhow!("Cannot connect as admin: {e:?}"))?; bootstrap .drop_databases(&data_db) diff --git a/crates/bin/src/cmd_init.rs b/crates/app/src/cmd_init.rs similarity index 97% rename from crates/bin/src/cmd_init.rs rename to crates/app/src/cmd_init.rs index 92118bb8..9e5ac10a 100755 --- a/crates/bin/src/cmd_init.rs +++ b/crates/app/src/cmd_init.rs @@ -10,8 +10,8 @@ use std::path::Path; use clap::Args; -use crate::config; use crate::init_helpers::{generate_config, generate_tls_cert_if_needed}; +use extenddb_config as config; #[derive(Args)] #[allow(clippy::doc_markdown)] // Clap help text, not rustdoc @@ -140,10 +140,9 @@ pub async fn run(args: InitArgs) -> anyhow::Result { let cli_args: Vec = std::env::args().collect(); // Create bootstrapper via registry (no hardcoded match!) - let bootstrapper = - extenddb_storage::bootstrapper::create_bootstrapper(&backend, &args.config, &cli_args) - .await - .map_err(|e| anyhow::anyhow!("{e:?}"))?; + let bootstrapper = extenddb_storage::bootstrapper::create_bootstrapper(&args.config, &cli_args) + .await + .map_err(|e| anyhow::anyhow!("{e:?}"))?; // Ensure application user exists. bootstrapper diff --git a/crates/bin/src/cmd_manage.rs b/crates/app/src/cmd_manage.rs similarity index 100% rename from crates/bin/src/cmd_manage.rs rename to crates/app/src/cmd_manage.rs diff --git a/crates/bin/src/cmd_migrate.rs b/crates/app/src/cmd_migrate.rs similarity index 90% rename from crates/bin/src/cmd_migrate.rs rename to crates/app/src/cmd_migrate.rs index 8221a80d..46c89c61 100755 --- a/crates/bin/src/cmd_migrate.rs +++ b/crates/app/src/cmd_migrate.rs @@ -7,7 +7,7 @@ use clap::Args; -use crate::config; +use extenddb_config as config; #[derive(Args)] pub struct MigrateArgs { @@ -36,8 +36,10 @@ pub async fn run(args: MigrateArgs) -> anyhow::Result<()> { args.config, ); } - let app_config = config::load(&args.config)?; - let backend = &app_config.storage.backend; + // Load the config for validation only: migrate drives the bootstrapper from + // the config path directly, but a malformed config should fail here rather + // than midway through a migration. + config::load(&args.config)?; println!("=== extenddb migrate ==="); println!("Config: {}", args.config); @@ -47,10 +49,9 @@ pub async fn run(args: MigrateArgs) -> anyhow::Result<()> { let cli_args: Vec = std::env::args().collect(); // Create bootstrapper via registry - let bootstrap = - extenddb_storage::bootstrapper::create_bootstrapper(backend, &args.config, &cli_args) - .await - .map_err(|e| anyhow::anyhow!("{e:?}"))?; + let bootstrap = extenddb_storage::bootstrapper::create_bootstrapper(&args.config, &cli_args) + .await + .map_err(|e| anyhow::anyhow!("{e:?}"))?; // Show current catalog version. println!("--- Checking current catalog version..."); diff --git a/crates/app/src/cmd_serve.rs b/crates/app/src/cmd_serve.rs new file mode 100755 index 00000000..714d2db1 --- /dev/null +++ b/crates/app/src/cmd_serve.rs @@ -0,0 +1,236 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! `extenddb serve` — start the Virtual `DynamoDB` server. + +use std::net::TcpListener; + +use clap::Args; +use daemonize::Daemonize; + +use crate::serve_helpers::{check_config_permissions, verify_daemon_started}; +use extenddb_config as config; +use extenddb_config::pid_file_path; +use extenddb_server::{BuildInfo, LogTarget, ServeParams}; + +#[derive(Args, Default)] +pub struct ServeArgs { + /// Path to configuration file + #[arg(short, long, default_value = "extenddb.toml")] + config: String, + + /// Override server port + #[arg(short, long)] + port: Option, + + /// Run in the foreground without daemonizing. + /// + /// Useful for running under a container or process supervisor (Docker, + /// Kubernetes, systemd Type=simple, runit, etc.). In foreground mode logs + /// are written to stderr instead of syslog so the supervisor can capture + /// them. + #[arg(long, alias = "no-daemon")] + foreground: bool, +} + +/// Bind the listening socket, daemonize, then start the tokio runtime. +/// Binding before forking ensures port conflicts are reported to stderr +/// before the parent process exits (D-4). +pub fn run(args: &ServeArgs, build: BuildInfo) -> anyhow::Result<()> { + // P50: Check config file permissions before loading. The config file may + // contain the encryption key (via `extenddb init`). Reject if more permissive + // than 0600 (owner read/write only). + if !std::path::Path::new(&args.config).exists() { + anyhow::bail!( + "Config file '{}' not found. Run 'extenddb init' to create one, \ + or use --config to specify a different location.", + args.config, + ); + } + check_config_permissions(&args.config)?; + + // Load config early so bind address is known before fork. + let app_config = config::load(&args.config)?; + + // D5: TLS is mandatory. Reject explicit opt-out. + if !app_config.server.tls.enabled { + anyhow::bail!("TLS is mandatory. Remove `tls.enabled = false` from your config file."); + } + + // D6: Auth is mandatory. Only "builtin" is supported. + if app_config.auth.provider == "none" { + anyhow::bail!( + "auth.provider = \"none\" is no longer supported. \ + Set auth.provider = \"builtin\" and run `extenddb init`." + ); + } + if app_config.auth.provider != "builtin" { + anyhow::bail!( + "Unknown auth provider '{}'. Only 'builtin' is supported.", + app_config.auth.provider + ); + } + + // Validate backend is supported and get catalog version (fail fast before binding port) + let backend = &app_config.storage.backend; + let catalog_version = extenddb_storage::operations::catalog_version()?; + + let port = args.port.unwrap_or(app_config.server.port); + let bind_addr = format!("{}:{}", app_config.server.bind_addr, port); + + // Bind in sync context — errors go to stderr before daemonizing. + let std_listener = TcpListener::bind(&bind_addr) + .map_err(|e| anyhow::anyhow!("Failed to bind {bind_addr}: {e}"))?; + std_listener + .set_nonblocking(true) + .map_err(|e| anyhow::anyhow!("Failed to set listener non-blocking: {e}"))?; + + // D-2: Print startup banner before daemonizing so the user gets + // confirmation the server is starting. P57 Bug 4 fix: say "starting" not + // "listening" — the server isn't actually accepting connections yet. + // + // In daemon mode the banner goes to stdout (the user invoking `extenddb + // serve` reads it before the parent exits). In foreground mode we route + // it to stderr so a process supervisor receives banner and tracing logs + // on the same stream — mixing stdout and stderr makes container log + // capture noisier than necessary. + let banner_line1 = format!( + "extenddb {} (catalog {}) starting on {}", + build.version, catalog_version, bind_addr, + ); + let banner_line2 = format!( + " storage: {} ({})", + backend, + config::redact_password(app_config.storage.connection_config()), + ); + if args.foreground { + eprintln!("{banner_line1}"); + eprintln!("{banner_line2}"); + } else { + println!("{banner_line1}"); + println!("{banner_line2}"); + } + + // D-3: Write PID file so `extenddb status` can report the daemon PID. + let run_dir = config::expand_tilde(&app_config.server.run_dir); + std::fs::create_dir_all(&run_dir) + .map_err(|e| anyhow::anyhow!("Failed to create run directory {run_dir}: {e}"))?; + let pid_file = pid_file_path(&run_dir, port); + + // P57 Bug 7 fix: Use execute() instead of start() so the parent can + // verify the daemon child is healthy before exiting. start() exits the + // parent immediately after fork, hiding child startup failures. + // + // When --foreground is set, skip daemonization entirely so the process + // can be supervised by Docker, Kubernetes, systemd Type=simple, etc. + // The PID file is still written below by `start_server`, and graceful + // shutdown on SIGINT/SIGTERM still works. + if !args.foreground { + let daemon = Daemonize::new().pid_file(&pid_file); + match daemon.execute() { + daemonize::Outcome::Parent(Ok(_)) => { + // Parent process: wait for the PID file to appear (written by + // the grandchild after the double-fork), then verify the daemon + // is still alive. This catches crashes during early startup + // (bad config, missing tables, TLS cert errors). + return verify_daemon_started(&pid_file, &bind_addr); + } + daemonize::Outcome::Parent(Err(e)) => { + return Err(anyhow::anyhow!("Failed to daemonize: {e}")); + } + daemonize::Outcome::Child(Ok(_)) => { + // Child (daemon) process: continue to start the server. + } + daemonize::Outcome::Child(Err(e)) => { + return Err(anyhow::anyhow!("Failed to daemonize (child): {e}")); + } + } + + // P57 Bug 3 fix: After daemonize, stderr is /dev/null. Install a panic + // hook that writes to syslog so panics are visible. Without this, the + // child process silently disappears on panic. Reuses the server crate's + // raw syslog writer so there is one implementation of it — tracing is + // unusable here because the subscriber is only set up inside `serve`. + std::panic::set_hook(Box::new(|info| { + extenddb_server::log_to_syslog_raw(&format!("extenddb panic: {info}")); + })); + } + + tokio::runtime::Builder::new_multi_thread() + .enable_all() + .build()? + .block_on(extenddb_server::serve( + ServeParams::new(app_config, std_listener, run_dir, build).with_log_target( + if args.foreground { + LogTarget::Stderr + } else { + LogTarget::Syslog + }, + ), + )) +} + +#[cfg(test)] +mod tests { + use super::ServeArgs; + use clap::Parser; + + /// Test wrapper so clap has a top-level `Parser` to drive `ServeArgs`. + #[derive(Parser)] + struct TestCli { + #[command(flatten)] + args: ServeArgs, + } + + fn parse(argv: &[&str]) -> ServeArgs { + TestCli::try_parse_from(argv) + .expect("ServeArgs should parse from valid argv") + .args + } + + #[test] + fn defaults_run_in_daemon_mode() { + // No --foreground flag preserves the historical daemon behavior so + // existing users and scripts are unaffected by the new flag. + let args = parse(&["extenddb-serve"]); + assert!(!args.foreground); + assert_eq!(args.config, "extenddb.toml"); + assert!(args.port.is_none()); + } + + #[test] + fn foreground_flag_is_recognized() { + let args = parse(&["extenddb-serve", "--foreground"]); + assert!(args.foreground); + } + + #[test] + fn no_daemon_alias_is_recognized() { + // The issue proposed either `--foreground` or `--no-daemon`; make + // sure the alias keeps working so users have a choice. + let args = parse(&["extenddb-serve", "--no-daemon"]); + assert!(args.foreground); + } + + #[test] + fn foreground_combines_with_other_flags() { + let args = parse(&[ + "extenddb-serve", + "--config", + "/etc/extenddb/extenddb.toml", + "--port", + "9000", + "--foreground", + ]); + assert!(args.foreground); + assert_eq!(args.config, "/etc/extenddb/extenddb.toml"); + assert_eq!(args.port, Some(9000)); + } + + #[test] + fn unknown_flag_is_rejected() { + // Guard against accidental future renames silently dropping the flag. + let result = TestCli::try_parse_from(["extenddb-serve", "--daemon-off"]); + assert!(result.is_err()); + } +} diff --git a/crates/bin/src/cmd_settings.rs b/crates/app/src/cmd_settings.rs similarity index 97% rename from crates/bin/src/cmd_settings.rs rename to crates/app/src/cmd_settings.rs index d1595cd7..88f3e775 100755 --- a/crates/bin/src/cmd_settings.rs +++ b/crates/app/src/cmd_settings.rs @@ -11,7 +11,7 @@ use clap::{Args, Subcommand}; use extenddb_storage::management_store::SettingsStore; -use crate::config; +use extenddb_config as config; // Re-use validation constants from the ops layer. use extenddb_server::management::ops_settings::{KNOWN_KEYS, READONLY_KEYS}; @@ -46,9 +46,7 @@ pub async fn run(args: SettingsArgs) -> anyhow::Result<()> { ); } let app_config = config::load(&args.config)?; - let backend = &app_config.storage.backend; let store = extenddb_storage::settings_store::create_settings_store( - backend, app_config.storage.connection_config(), ) .await diff --git a/crates/bin/src/cmd_status.rs b/crates/app/src/cmd_status.rs similarity index 93% rename from crates/bin/src/cmd_status.rs rename to crates/app/src/cmd_status.rs index 02c7dca5..e48cd9ca 100755 --- a/crates/bin/src/cmd_status.rs +++ b/crates/app/src/cmd_status.rs @@ -11,7 +11,7 @@ use std::net::TcpStream; use clap::Args; -use crate::config; +use extenddb_config as config; #[derive(Args)] pub struct StatusArgs { @@ -43,8 +43,8 @@ pub fn run(args: &StatusArgs) { // Validate the PID is alive to avoid reporting stale PIDs after unclean shutdown. // Try config-based run_dir first, fall back to default. let pid_file = config::load(&args.config).map_or_else( - |_| crate::serve_helpers::pid_file_path_default(port), - |c| crate::serve_helpers::pid_file_path(&config::expand_tilde(&c.server.run_dir), port), + |_| extenddb_config::pid_file_path_default(port), + |c| extenddb_config::pid_file_path(&config::expand_tilde(&c.server.run_dir), port), ); let pid_label = std::fs::read_to_string(&pid_file) .ok() diff --git a/crates/bin/src/cmd_stop.rs b/crates/app/src/cmd_stop.rs similarity index 94% rename from crates/bin/src/cmd_stop.rs rename to crates/app/src/cmd_stop.rs index 16621247..9308420f 100755 --- a/crates/bin/src/cmd_stop.rs +++ b/crates/app/src/cmd_stop.rs @@ -11,8 +11,8 @@ use std::time::{Duration, Instant}; use clap::Args; -use crate::config; use crate::util::is_process_alive; +use extenddb_config as config; /// Maximum time to wait for the process to exit after SIGTERM. const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(10); @@ -41,10 +41,8 @@ pub fn run(args: &StopArgs) { .unwrap_or(18443); let pid_file = match &app_config { - Some(c) => { - crate::serve_helpers::pid_file_path(&config::expand_tilde(&c.server.run_dir), port) - } - None => crate::serve_helpers::pid_file_path_default(port), + Some(c) => extenddb_config::pid_file_path(&config::expand_tilde(&c.server.run_dir), port), + None => extenddb_config::pid_file_path_default(port), }; let pid_str = match std::fs::read_to_string(&pid_file) { diff --git a/crates/bin/src/cmd_verify.rs b/crates/app/src/cmd_verify.rs similarity index 94% rename from crates/bin/src/cmd_verify.rs rename to crates/app/src/cmd_verify.rs index f94113ef..8e35a397 100755 --- a/crates/bin/src/cmd_verify.rs +++ b/crates/app/src/cmd_verify.rs @@ -13,7 +13,7 @@ use clap::Args; -use crate::config; +use extenddb_config as config; #[derive(Args)] pub struct VerifyArgs { @@ -31,13 +31,11 @@ pub async fn run(args: VerifyArgs) -> anyhow::Result<()> { ); } let app_config = config::load(&args.config)?; - let backend = &app_config.storage.backend; - let expected_version = extenddb_storage::operations::catalog_version(backend) - .unwrap_or_else(|_| "unknown".to_string()); + let expected_version = + extenddb_storage::operations::catalog_version().unwrap_or_else(|_| "unknown".to_string()); // Parse connection string to get database name for display let parts = extenddb_storage::operations::parse_connection_string( - backend, app_config.storage.connection_config(), ) .map_err(|e| anyhow::anyhow!("Failed to parse connection string: {e}"))?; @@ -52,7 +50,6 @@ pub async fn run(args: VerifyArgs) -> anyhow::Result<()> { // Create settings and diagnostics store println!("--- Checking catalog connection..."); let store = match extenddb_storage::settings_store::create_settings_store( - backend, app_config.storage.connection_config(), ) .await @@ -93,7 +90,6 @@ pub async fn run(args: VerifyArgs) -> anyhow::Result<()> { // Create diagnostics store (reuse for data DB test and table/index counts) let diag_store = extenddb_storage::diagnostics_store::create_diagnostics_store( - backend, app_config.storage.connection_config(), ) .await diff --git a/crates/bin/src/init_helpers.rs b/crates/app/src/init_helpers.rs similarity index 95% rename from crates/bin/src/init_helpers.rs rename to crates/app/src/init_helpers.rs index 3d0b7997..13ad0cce 100755 --- a/crates/bin/src/init_helpers.rs +++ b/crates/app/src/init_helpers.rs @@ -5,7 +5,7 @@ /// Generate a self-signed TLS certificate and key if they don't already exist. pub fn generate_tls_cert_if_needed(bind_addr: &str) -> anyhow::Result<()> { - let tls_dir = crate::config::expand_tilde("~/.extenddb/tls"); + let tls_dir = extenddb_config::expand_tilde("~/.extenddb/tls"); let cert_path = format!("{tls_dir}/cert.pem"); let key_path = format!("{tls_dir}/key.pem"); @@ -76,9 +76,9 @@ pub(crate) fn generate_config( let timestamp = time::OffsetDateTime::now_utc() .format(&time::format_description::well_known::Rfc3339) .unwrap_or_else(|_| "unknown".to_owned()); - let tls_cert = crate::config::expand_tilde("~/.extenddb/tls/cert.pem"); - let tls_key = crate::config::expand_tilde("~/.extenddb/tls/key.pem"); - let run_dir = crate::config::expand_tilde("~/.extenddb/run"); + let tls_cert = extenddb_config::expand_tilde("~/.extenddb/tls/cert.pem"); + let tls_key = extenddb_config::expand_tilde("~/.extenddb/tls/key.pem"); + let run_dir = extenddb_config::expand_tilde("~/.extenddb/run"); // Compute docs_dir line before the template so it lands in the top-level // TOML section (before any [section] header). diff --git a/crates/app/src/lib.rs b/crates/app/src/lib.rs new file mode 100644 index 00000000..6f04502f --- /dev/null +++ b/crates/app/src/lib.rs @@ -0,0 +1,161 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! ExtendDB application/CLI library. +//! +//! Owns the command-line interface (`serve`, `init`, `destroy`, `verify`, +//! `migrate`, `status`, `stop`, `settings`, `manage`, `catalog-check`) and the +//! subcommand dispatch. It is backend-agnostic: a backend's thin `main` +//! installs its backend with +//! [`set_backend`](extenddb_storage::set_backend) and then calls [`run`]: +//! +//! ```ignore +//! fn main() -> anyhow::Result<()> { +//! extenddb_storage::set_backend(my_backend::backend())?; +//! extenddb_app::run(extenddb_app::BuildInfo { +//! version: env!("CARGO_PKG_VERSION"), +//! git_hash: env!("MY_GIT_HASH"), +//! build_time: env!("MY_BUILD_TIME"), +//! }) +//! } +//! ``` + +mod cmd_catalog_check; +mod cmd_destroy; +mod cmd_init; +mod cmd_manage; +mod cmd_migrate; +mod cmd_serve; +mod cmd_settings; +mod cmd_status; +mod cmd_stop; +mod cmd_verify; +mod init_helpers; +mod manage_http; +mod manage_types; +mod serve_helpers; +mod util; + +use clap::{Parser, Subcommand}; + +/// Build provenance supplied by the deployed binary. +/// +/// Defined by `extenddb-server` (which consumes it for the startup banner and +/// console version string) and re-exported here so a thin `main` only needs the +/// `extenddb-app` dependency. +pub use extenddb_server::BuildInfo; + +#[derive(Parser)] +#[command(name = "extenddb", about = "ExtendDB — DynamoDB-compatible API server")] +struct Cli { + /// Print version and exit + #[arg(short = 'V', long)] + version: bool, + + #[command(subcommand)] + command: Option, +} + +#[derive(Subcommand)] +enum Command { + /// Start the extenddb server + Serve(cmd_serve::ServeArgs), + /// Initialize a new extenddb deployment + Init(cmd_init::InitArgs), + /// Tear down a extenddb deployment + Destroy(cmd_destroy::DestroyArgs), + /// Validate a extenddb deployment + Verify(cmd_verify::VerifyArgs), + /// Apply catalog schema migrations + Migrate(cmd_migrate::MigrateArgs), + /// Check if the extenddb server is running + Status(cmd_status::StatusArgs), + /// Stop the running extenddb server + Stop(cmd_stop::StopArgs), + /// Read or write runtime settings + Settings(cmd_settings::SettingsArgs), + /// Manage admin users and accounts via the management API + Manage(cmd_manage::ManageArgs), + /// Check catalog and data database integrity + CatalogCheck(cmd_catalog_check::CatalogCheckArgs), + /// Print version, catalog version, git commit, and build timestamp + Version, +} + +/// Parse the command line and dispatch the selected subcommand. +/// +/// The backend must already be installed via +/// [`extenddb_storage::set_backend`] before this is called. +/// +/// # Errors +/// +/// Returns any error produced by the selected subcommand. +pub fn run(build: BuildInfo) -> anyhow::Result<()> { + let cli = Cli::parse(); + + if cli.version { + print_version(build); + return Ok(()); + } + + match cli.command.unwrap_or(Command::Version) { + Command::Serve(args) => cmd_serve::run(&args, build), + Command::Init(args) => { + let code = run_interactive(cmd_init::run(args))?; + if code != 0 { + std::process::exit(i32::from(code)); + } + Ok(()) + } + Command::Destroy(args) => run_interactive(cmd_destroy::run(args)), + Command::Verify(args) => run_interactive(cmd_verify::run(args)), + Command::Migrate(args) => run_interactive(cmd_migrate::run(args)), + Command::Status(args) => { + cmd_status::run(&args); + Ok(()) + } + Command::Stop(args) => { + cmd_stop::run(&args); + Ok(()) + } + Command::Settings(args) => run_interactive(cmd_settings::run(args)), + Command::Manage(args) => run_interactive(cmd_manage::run(args)), + Command::CatalogCheck(args) => run_interactive(cmd_catalog_check::run(args)), + Command::Version => { + print_version(build); + Ok(()) + } + } +} + +/// Print version, catalog version, git commit hash, and build timestamp. +fn print_version(build: BuildInfo) { + println!("extenddb {}", build.version); + + // One backend is compiled into this binary; report its catalog version. + match ( + extenddb_storage::backend_name(), + extenddb_storage::operations::catalog_version(), + ) { + (Some(backend), Ok(version)) => println!("catalog {version} ({backend})"), + (Some(backend), Err(_)) => println!("catalog unknown ({backend})"), + (None, _) => println!("catalog unknown (no backend installed)"), + } + + println!("commit {}", build.git_hash); + println!("built {}", build.build_time); +} + +/// Run an async subcommand with a single-threaded tokio runtime and stderr logging. +/// All non-serve subcommands are interactive (D-24). +fn run_interactive( + future: impl std::future::Future>, +) -> anyhow::Result { + tracing_subscriber::fmt() + .try_init() + .unwrap_or_else(|e| eprintln!("Warning: logging init failed: {e}")); + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build()? + .block_on(future) +} diff --git a/crates/bin/src/manage_http.rs b/crates/app/src/manage_http.rs similarity index 99% rename from crates/bin/src/manage_http.rs rename to crates/app/src/manage_http.rs index ec52891c..5a8bb1a0 100755 --- a/crates/bin/src/manage_http.rs +++ b/crates/app/src/manage_http.rs @@ -12,8 +12,8 @@ use std::sync::Arc; use base64::Engine; -use crate::config; use crate::manage_types::{CacheAction, CacheInvalidateScope, ManageCommand}; +use extenddb_config as config; /// Resolve the management API endpoint from CLI args or config file. /// diff --git a/crates/bin/src/manage_types.rs b/crates/app/src/manage_types.rs similarity index 100% rename from crates/bin/src/manage_types.rs rename to crates/app/src/manage_types.rs diff --git a/crates/bin/src/serve_helpers.rs b/crates/app/src/serve_helpers.rs similarity index 78% rename from crates/bin/src/serve_helpers.rs rename to crates/app/src/serve_helpers.rs index b8c705f1..d83949de 100755 --- a/crates/bin/src/serve_helpers.rs +++ b/crates/app/src/serve_helpers.rs @@ -6,26 +6,6 @@ use std::path::PathBuf; -use crate::config; - -/// P57 Bug 7: Best-effort raw syslog write for fatal errors. Used when the -/// tracing subscriber may not be initialized (e.g., errors during early -/// startup before syslog tracing is configured). -pub fn log_to_syslog_raw(msg: &str) { - // SAFETY: openlog/syslog are POSIX-standard C functions. The ident - // string is a static C string literal with 'static lifetime. - unsafe { - libc::openlog( - c"extenddb".as_ptr(), - libc::LOG_PID | libc::LOG_NDELAY, - libc::LOG_DAEMON, - ); - if let Ok(cmsg) = std::ffi::CString::new(msg.to_owned()) { - libc::syslog(libc::LOG_CRIT, c"%s".as_ptr(), cmsg.as_ptr()); - } - } -} - /// Platform-appropriate hint for viewing syslog output. fn syslog_hint() -> &'static str { if cfg!(target_os = "macos") { @@ -83,19 +63,6 @@ pub fn verify_daemon_started(pid_file: &PathBuf, bind_addr: &str) -> anyhow::Res Ok(()) } -/// PID file path for a given port and run directory. -/// Used by `serve` (write) and `status` (read). -pub fn pid_file_path(run_dir: &str, port: u16) -> PathBuf { - PathBuf::from(format!("{run_dir}/extenddb-{port}.pid")) -} - -/// PID file path using the default run directory. Used by `status` when -/// no config file is loaded. -pub fn pid_file_path_default(port: u16) -> PathBuf { - let run_dir = config::ServerConfig::default().run_dir; - pid_file_path(&run_dir, port) -} - /// Check that the config file has permissions no more permissive than `0600`. /// /// The config file may contain the encryption key for credential storage. diff --git a/crates/bin/src/util.rs b/crates/app/src/util.rs similarity index 100% rename from crates/bin/src/util.rs rename to crates/app/src/util.rs diff --git a/crates/bin/Cargo.toml b/crates/bin/Cargo.toml index 9b6755b8..d5b148ab 100755 --- a/crates/bin/Cargo.toml +++ b/crates/bin/Cargo.toml @@ -11,33 +11,8 @@ license.workspace = true name = "extenddb" path = "src/main.rs" -[features] -default = ["postgres"] -postgres = ["extenddb-storage-postgres"] - [dependencies] -extenddb-auth = { workspace = true } -extenddb-cache = { workspace = true } -extenddb-core = { workspace = true } -extenddb-engine = { workspace = true } +extenddb-app = { workspace = true } extenddb-storage = { workspace = true } -extenddb-storage-postgres = { workspace = true, optional = true } -extenddb-server = { workspace = true } -tokio = { workspace = true } +extenddb-storage-postgres = { workspace = true } anyhow = { workspace = true } -clap = { workspace = true } -config = { workspace = true } -daemonize = { workspace = true } -serde = { workspace = true } -serde_json = { workspace = true } -sqlx = { workspace = true } -time = { workspace = true } -toml = { workspace = true } -tracing = { workspace = true } -tracing-subscriber = { workspace = true } -syslog-tracing = { workspace = true } -base64 = { workspace = true } -libc = { workspace = true } -rcgen = { workspace = true } -rustls = { workspace = true } -rustls-pemfile = { workspace = true } diff --git a/crates/bin/src/cmd_serve.rs b/crates/bin/src/cmd_serve.rs deleted file mode 100755 index 3918b7cf..00000000 --- a/crates/bin/src/cmd_serve.rs +++ /dev/null @@ -1,683 +0,0 @@ -// Copyright 2026 ExtendDB contributors -// SPDX-License-Identifier: Apache-2.0 - -//! `extenddb serve` — start the Virtual `DynamoDB` server. - -use std::net::TcpListener; -use std::sync::Arc; - -use clap::Args; -use daemonize::Daemonize; -use extenddb_server::AppState; -use syslog_tracing::{Facility, Options, Syslog}; -use tracing_subscriber::{ - EnvFilter, Layer, fmt, fmt::writer::BoxMakeWriter, layer::SubscriberExt, reload, - util::SubscriberInitExt, -}; - -use crate::config; -use crate::serve_helpers::{ - check_config_permissions, log_to_syslog_raw, pid_file_path, verify_daemon_started, -}; -use crate::workers; - -#[derive(Args, Default)] -pub struct ServeArgs { - /// Path to configuration file - #[arg(short, long, default_value = "extenddb.toml")] - config: String, - - /// Override server port - #[arg(short, long)] - port: Option, - - /// Run in the foreground without daemonizing. - /// - /// Useful for running under a container or process supervisor (Docker, - /// Kubernetes, systemd Type=simple, runit, etc.). In foreground mode logs - /// are written to stderr instead of syslog so the supervisor can capture - /// them. - #[arg(long, alias = "no-daemon")] - foreground: bool, -} - -/// Bind the listening socket, daemonize, then start the tokio runtime. -/// Binding before forking ensures port conflicts are reported to stderr -/// before the parent process exits (D-4). -pub fn run(args: &ServeArgs) -> anyhow::Result<()> { - // P50: Check config file permissions before loading. The config file may - // contain the encryption key (via `extenddb init`). Reject if more permissive - // than 0600 (owner read/write only). - if !std::path::Path::new(&args.config).exists() { - anyhow::bail!( - "Config file '{}' not found. Run 'extenddb init' to create one, \ - or use --config to specify a different location.", - args.config, - ); - } - check_config_permissions(&args.config)?; - - // Load config early so bind address is known before fork. - let app_config = config::load(&args.config)?; - - // D5: TLS is mandatory. Reject explicit opt-out. - if !app_config.server.tls.enabled { - anyhow::bail!("TLS is mandatory. Remove `tls.enabled = false` from your config file."); - } - - // D6: Auth is mandatory. Only "builtin" is supported. - if app_config.auth.provider == "none" { - anyhow::bail!( - "auth.provider = \"none\" is no longer supported. \ - Set auth.provider = \"builtin\" and run `extenddb init`." - ); - } - if app_config.auth.provider != "builtin" { - anyhow::bail!( - "Unknown auth provider '{}'. Only 'builtin' is supported.", - app_config.auth.provider - ); - } - - // Validate backend is supported and get catalog version (fail fast before binding port) - let backend = &app_config.storage.backend; - let catalog_version = extenddb_storage::operations::catalog_version(backend)?; - - let port = args.port.unwrap_or(app_config.server.port); - let bind_addr = format!("{}:{}", app_config.server.bind_addr, port); - - // Bind in sync context — errors go to stderr before daemonizing. - let std_listener = TcpListener::bind(&bind_addr) - .map_err(|e| anyhow::anyhow!("Failed to bind {bind_addr}: {e}"))?; - std_listener - .set_nonblocking(true) - .map_err(|e| anyhow::anyhow!("Failed to set listener non-blocking: {e}"))?; - - // D-2: Print startup banner before daemonizing so the user gets - // confirmation the server is starting. P57 Bug 4 fix: say "starting" not - // "listening" — the server isn't actually accepting connections yet. - // - // In daemon mode the banner goes to stdout (the user invoking `extenddb - // serve` reads it before the parent exits). In foreground mode we route - // it to stderr so a process supervisor receives banner and tracing logs - // on the same stream — mixing stdout and stderr makes container log - // capture noisier than necessary. - let banner_line1 = format!( - "extenddb {} (catalog {}) starting on {}", - env!("CARGO_PKG_VERSION"), - catalog_version, - bind_addr, - ); - let banner_line2 = format!( - " storage: {} ({})", - backend, - config::redact_password(backend, app_config.storage.connection_config()), - ); - if args.foreground { - eprintln!("{banner_line1}"); - eprintln!("{banner_line2}"); - } else { - println!("{banner_line1}"); - println!("{banner_line2}"); - } - - // D-3: Write PID file so `extenddb status` can report the daemon PID. - let run_dir = config::expand_tilde(&app_config.server.run_dir); - std::fs::create_dir_all(&run_dir) - .map_err(|e| anyhow::anyhow!("Failed to create run directory {run_dir}: {e}"))?; - let pid_file = pid_file_path(&run_dir, port); - - // P57 Bug 7 fix: Use execute() instead of start() so the parent can - // verify the daemon child is healthy before exiting. start() exits the - // parent immediately after fork, hiding child startup failures. - // - // When --foreground is set, skip daemonization entirely so the process - // can be supervised by Docker, Kubernetes, systemd Type=simple, etc. - // The PID file is still written below by `start_server`, and graceful - // shutdown on SIGINT/SIGTERM still works. - if !args.foreground { - let daemon = Daemonize::new().pid_file(&pid_file); - match daemon.execute() { - daemonize::Outcome::Parent(Ok(_)) => { - // Parent process: wait for the PID file to appear (written by - // the grandchild after the double-fork), then verify the daemon - // is still alive. This catches crashes during early startup - // (bad config, missing tables, TLS cert errors). - return verify_daemon_started(&pid_file, &bind_addr); - } - daemonize::Outcome::Parent(Err(e)) => { - return Err(anyhow::anyhow!("Failed to daemonize: {e}")); - } - daemonize::Outcome::Child(Ok(_)) => { - // Child (daemon) process: continue to start the server. - } - daemonize::Outcome::Child(Err(e)) => { - return Err(anyhow::anyhow!("Failed to daemonize (child): {e}")); - } - } - - // P57 Bug 3 fix: After daemonize, stderr is /dev/null. Install a panic - // hook that writes to syslog so panics are visible. Without this, the - // child process silently disappears on panic. - std::panic::set_hook(Box::new(|info| { - // Best-effort syslog write. We can't use tracing here because the - // subscriber may not be initialized yet (it's set up in serve_inner). - let msg = format!("extenddb panic: {info}"); - // SAFETY: openlog/syslog are POSIX-standard C functions. The ident - // string is a static C string literal with 'static lifetime. - unsafe { - libc::openlog( - c"extenddb".as_ptr(), - libc::LOG_PID | libc::LOG_NDELAY, - libc::LOG_DAEMON, - ); - // Use CString to ensure null-termination for the format arg. - if let Ok(cmsg) = std::ffi::CString::new(msg) { - libc::syslog(libc::LOG_CRIT, c"%s".as_ptr(), cmsg.as_ptr()); - } - } - })); - } - - tokio::runtime::Builder::new_multi_thread() - .enable_all() - .build()? - .block_on(serve( - app_config, - std_listener, - port, - run_dir, - args.foreground, - )) -} - -/// Async entry point: initializes syslog logging, storage, and auth, then -/// starts the HTTP server on the pre-bound listener. -async fn serve( - app_config: config::AppConfig, - std_listener: TcpListener, - port: u16, - run_dir: String, - foreground: bool, -) -> anyhow::Result<()> { - // CB-27: Clean up PID file if serve() fails before reaching the HTTP - // server (e.g., Postgres connection failure). The PID file was already - // written by Daemonize in run(). - let pid_path = pid_file_path(&run_dir, port); - let backend = app_config.storage.backend.clone(); - let result = serve_inner(app_config, std_listener, port, run_dir, backend, foreground).await; - if let Err(ref e) = result { - let _ = std::fs::remove_file(&pid_path); - // P57 Bug 7: Log fatal errors to syslog. After daemonize, stderr is - // /dev/null so anyhow's error display is lost. Use tracing if - // available, fall back to raw syslog if tracing isn't initialized yet. - // In foreground mode, also echo to stderr since the supervisor - // captures stderr rather than syslog. - tracing::error!("extenddb fatal: {e:#}"); - if foreground { - eprintln!("extenddb fatal: {e:#}"); - } else { - log_to_syslog_raw(&format!("extenddb fatal: {e:#}")); - } - } - result -} - -/// Inner serve function — separated so the outer `serve` can clean up the PID -/// file on any error path. -async fn serve_inner( - app_config: config::AppConfig, - std_listener: TcpListener, - port: u16, - run_dir: String, - backend: String, - foreground: bool, -) -> anyhow::Result<()> { - let catalog_version = extenddb_storage::operations::catalog_version(&backend) - .unwrap_or_else(|_| "unknown".to_string()); - - // In foreground mode, daemonize was skipped so the PID file was never - // written. Write it now so `extenddb status`/`stop` and `start_server`'s - // graceful shutdown cleanup still work. The grandchild PID written by - // daemonize matches `std::process::id()` post-fork, so this stays - // consistent with daemon mode. - if foreground { - let pid_file = pid_file_path(&run_dir, port); - std::fs::write(&pid_file, format!("{}\n", std::process::id())) - .map_err(|e| anyhow::anyhow!("Failed to write PID file {}: {e}", pid_file.display()))?; - } - - // Init logging (REQ-LOG-003, REQ-LOG-006) — syslog in daemon mode, stderr - // in foreground mode so a container/process supervisor can capture logs. - // D-3: sqlx messages are controlled by an independent `sqlx_log_level` - // runtime setting (default: warn). Both extenddb and sqlx messages use the - // `extenddb` syslog identifier (POSIX syslog supports only one identity per - // process). sqlx messages are identifiable by their `sqlx::query` target. - // Filter with: `journalctl -t extenddb | grep -v sqlx` (exclude) or - // `journalctl -t extenddb | grep sqlx` (include only). - // - // The EnvFilter encodes both levels: `{app_level},sqlx={sqlx_level}`. - // The poll_log_level worker reloads the filter when either setting changes. - let filter_str = format!("{},sqlx=warn", app_config.logging.level); - // CB-29: Always use the config file log level, never RUST_LOG. The runtime - // settings poller handles dynamic level changes. RUST_LOG silently - // overriding the config is an operational surprise. - let filter = EnvFilter::new(&filter_str); - let (filter_layer, reload_handle) = reload::Layer::new(filter); - - // Pick the writer first (foreground → stderr, daemon → syslog), then the - // format (text vs json). syslog supplies its own timestamps, so we strip - // them with `.without_time()` only on the syslog path. - let (writer, with_time): (BoxMakeWriter, bool) = if foreground { - (BoxMakeWriter::new(std::io::stderr), true) - } else { - let syslog = Syslog::new( - c"extenddb", - Options::LOG_PID | Options::LOG_NDELAY, - Facility::Daemon, - ) - .ok_or_else(|| { - anyhow::anyhow!( - "Failed to initialize syslog — another syslog logger may already be active" - ) - })?; - (BoxMakeWriter::new(syslog), false) - }; - - let fmt_layer = match (with_time, app_config.logging.format == "json") { - (true, true) => fmt::layer().json().with_writer(writer).boxed(), - (true, false) => fmt::layer().with_writer(writer).boxed(), - (false, true) => fmt::layer() - .json() - .without_time() - .with_writer(writer) - .boxed(), - (false, false) => fmt::layer().without_time().with_writer(writer).boxed(), - }; - - tracing_subscriber::registry() - .with(filter_layer) - .with(fmt_layer) - .try_init() - .map_err(|e| anyhow::anyhow!("Failed to initialize tracing: {e}"))?; - - // Create server components via factory pattern - let components = extenddb_storage::create_server_components( - &backend, - app_config.storage.as_trait(), - &app_config.server.region, - ) - .await?; - - let storage = components.engine; - let catalog_store = components.catalog_store; - let cred_store = components.credential_store; - let runtime_hooks = components.runtime_hooks; - - // Build SwrCacheConfig values from the [auth.cache] TOML section. - let cache_cfg = &app_config.auth.cache; - let cache_enabled = cache_cfg.enabled; - let make_cache_cfg = |name: &'static str| -> extenddb_cache::SwrCacheConfig { - extenddb_cache::SwrCacheConfig { - ttl: std::time::Duration::from_secs(cache_cfg.ttl_seconds), - soft_ttl: std::time::Duration::from_secs(cache_cfg.soft_ttl_seconds), - negative_ttl: std::time::Duration::from_secs(cache_cfg.negative_ttl_seconds), - max_entries: cache_cfg.max_entries, - name, - } - }; - // Validate config eagerly so misconfiguration fails fast at startup. - // Today every named subcache shares the same TTL/max_entries shape (only - // `name` differs), so a single `validate()` check suffices. If per-cache - // tuning is ever added, validate every constructed config here. - if let Err(e) = make_cache_cfg("__validate__").validate() { - anyhow::bail!( - "Invalid [auth.cache] configuration: {e}. Check ttl_seconds, \ - soft_ttl_seconds, negative_ttl_seconds, max_entries." - ); - } - if !cache_enabled { - tracing::warn!( - "auth.cache.enabled = false — auth/authz caches are in pass-through mode \ - (every lookup hits the catalog directly)" - ); - } - - // Phase 2: Wrap the raw credential store. In pass-through mode the - // wrapper bypasses the cache and forwards every lookup to the inner - // store; otherwise it caches per the TOML config. - let cached_cred_store = Arc::new(if cache_enabled { - extenddb_auth::CachedCredentialStore::with_arc(cred_store, make_cache_cfg("credential")) - } else { - extenddb_auth::CachedCredentialStore::pass_through_arc( - cred_store, - make_cache_cfg("credential"), - ) - }); - let auth: Arc = Arc::new( - extenddb_auth::BuiltinAuthProvider::new((*cached_cred_store).clone()), - ); - - // Phase 3: Build the authorization cache. - let authz_cache: Arc = { - let store: Arc = - catalog_store.clone(); - let cfg = extenddb_server::AuthzCacheConfig { - identity_policies: make_cache_cfg("identity_policies"), - group_policies: make_cache_cfg("group_policies"), - boundary: make_cache_cfg("boundary"), - principal_tags: make_cache_cfg("principal_tags"), - resource_tags: make_cache_cfg("resource_tags"), - session_data: make_cache_cfg("session_data"), - }; - Arc::new(if cache_enabled { - extenddb_server::CachedAuthzStore::new(store, cfg) - } else { - extenddb_server::CachedAuthzStore::pass_through(store, cfg) - }) - }; - - // Phase 4: Build the TableKeyInfo cache. - let table_key_info_cache: Arc = - Arc::new(if cache_enabled { - extenddb_server::CachedTableKeyInfoStore::new( - storage.clone(), - make_cache_cfg("table_key_info"), - ) - } else { - extenddb_server::CachedTableKeyInfoStore::pass_through( - storage.clone(), - make_cache_cfg("table_key_info"), - ) - }); - - // Assemble the cache registry threaded into AppState for write-through - // invalidations from the management API. - let auth_cache = - extenddb_auth::AuthCacheRegistry::empty() - .with_credential(cached_cred_store) - .with_authz_invalidator( - authz_cache.clone() as Arc - ) - .with_table_key_info_invalidator(table_key_info_cache.clone() - as Arc); - - let data_db_info = runtime_hooks - .as_ref() - .and_then(|h| h.backend_info()) - .unwrap_or_else(|| "(unknown)".to_owned()); - - // REQ-LOG-001: Startup banner with effective configuration. - // REQ-LOG-002: Connection strings redact passwords. - let log_output = if foreground { "stderr" } else { "syslog" }; - tracing::info!( - "extenddb {} (catalog {}) starting — bind={}:{}, region={}, auth={}, catalog_db={}, data_db={}, log_output={}, log_level={}", - env!("CARGO_PKG_VERSION"), - catalog_version, - app_config.server.bind_addr, - port, - app_config.server.region, - app_config.auth.provider, - config::redact_password(&backend, app_config.storage.connection_config()), - data_db_info, - log_output, - app_config.logging.level, - ); - - // Convert pre-bound std listener to tokio (D-4: bind before fork). - let listener = tokio::net::TcpListener::from_std(std_listener)?; - - // P120e: Create metrics collector early so workers can record health. - let metrics = Arc::new(extenddb_core::metrics::MetricsCollector::new()); - - let tls_enabled = app_config.server.tls.enabled; - - // P53: Resolve import and export path lists. Supports both the new - // [import]/[export] sections and the deprecated import_export_root. - let resolve_paths = |raw_paths: &[String], - label: &str| - -> anyhow::Result>> { - let mut resolved = Vec::new(); - for raw in raw_paths { - let expanded = config::expand_tilde(raw); - let path = std::path::PathBuf::from(&expanded); - if !path.exists() { - std::fs::create_dir_all(&path) - .map_err(|e| anyhow::anyhow!("Cannot create {label} path {expanded}: {e}"))?; - } - let canonical = path - .canonicalize() - .map_err(|e| anyhow::anyhow!("Cannot canonicalize {label} path {expanded}: {e}"))?; - resolved.push(Arc::new(canonical)); - } - Ok(resolved) - }; - - // Build effective path lists: new config takes precedence over deprecated. - let mut import_paths_raw = app_config.import_config.paths.clone(); - let mut export_paths_raw = app_config.export_config.paths.clone(); - if let Some(ref legacy) = app_config.import_export_root { - if import_paths_raw.is_empty() { - import_paths_raw.push(legacy.clone()); - } - if export_paths_raw.is_empty() { - export_paths_raw.push(legacy.clone()); - } - if !app_config.import_config.paths.is_empty() && !app_config.export_config.paths.is_empty() - { - tracing::warn!( - "Both import_export_root and [import]/[export] sections configured; import_export_root is ignored" - ); - } - } - - let import_paths: Arc<[Arc]> = - Arc::from(resolve_paths(&import_paths_raw, "import")?); - let export_paths: Arc<[Arc]> = - Arc::from(resolve_paths(&export_paths_raw, "export")?); - - if import_paths.is_empty() { - tracing::info!("Import disabled (no [import] paths configured)"); - } else { - for p in import_paths.iter() { - tracing::info!("Import enabled, path: {}", p.display()); - } - } - if export_paths.is_empty() { - tracing::info!("Export disabled (no [export] paths configured)"); - } else { - for p in export_paths.iter() { - tracing::info!("Export enabled, path: {}", p.display()); - } - } - - // D9: Build static config entries for the console settings page. - // Must be called before `app_config.limits` is moved. - let config_entries = config::build_config_entries(&app_config); - - // AI-1: Load runtime documentation from docs_dir if configured. - let docs_store = app_config.docs_dir.as_ref().and_then(|raw| { - let expanded = config::expand_tilde(raw); - let path = std::path::PathBuf::from(&expanded); - match extenddb_server::console::docs_embed::DocsStore::load(&path) { - Ok(store) => { - tracing::info!("Documentation loaded from {}", path.display()); - Some(store) - } - Err(e) => { - tracing::warn!("Documentation unavailable: {e}"); - None - } - } - }); - - let limits = Arc::new({ - let mut limits = app_config.limits; - if let Some(max_bytes) = app_config.max_import_bytes { - limits.max_import_file_bytes = max_bytes; - } - limits - }); - - let config_throttling = app_config.server.throttling_enabled.unwrap_or(false); - let initial_throttling = catalog_store - .get_setting("throttling_enabled") - .await - .ok() - .flatten() - .map_or(config_throttling, |v| v == "true"); - - let throttle = Arc::new(extenddb_core::throttle::ThrottleManager::new( - limits.per_account_max_rcu, - limits.per_account_max_wcu, - initial_throttling, - )); - - let state = AppState { - storage, - auth, - limits, - region: Arc::from(app_config.server.region.as_str()), - server_addr: format!("localhost:{port}"), - catalog_store: Some(catalog_store.clone()), - version_info: Arc::from( - format!( - "{} · catalog {} · {}", - env!("CARGO_PKG_VERSION"), - catalog_version, - env!("EXTENDDB_GIT_HASH"), - ) - .as_str(), - ), - metrics: metrics.clone(), - tls_enabled, - import_paths, - export_paths, - throttle: throttle.clone(), - auth_cache, - authz_cache, - table_key_info_cache, - config_entries, - docs_store, - }; - - // D-22: Spawn background task to poll log_level from settings table. - tokio::spawn(workers::poll_log_level( - catalog_store.clone(), - reload_handle.clone(), - app_config.logging.level.clone(), - )); - // Poll throttling_enabled runtime setting. - tokio::spawn(workers::poll_throttling_enabled( - catalog_store.clone(), - throttle, - config_throttling, - )); - // Spawn background tasks for metrics pruning and flushing. - tokio::spawn(workers::metrics_prune_worker(metrics.clone())); - tokio::spawn(workers::metrics_flush_worker( - metrics.clone(), - catalog_store.clone(), - )); - // Spawn background task to clean up old login attempt records. - tokio::spawn(workers::login_attempt_cleanup_worker(catalog_store.clone())); - // Phase 11a: Spawn background task to warn about approximate consumed capacity. - tokio::spawn(workers::capacity_warning_worker()); - - // Spawn backend-specific workers via runtime hooks - if let Some(hooks) = runtime_hooks { - let worker_ctx = extenddb_storage::WorkerContext { - metrics: metrics.clone(), - catalog_store: catalog_store.clone(), - reload_handle: reload_handle.clone(), - config_log_level: app_config.logging.level.clone(), - }; - hooks.spawn_workers(&worker_ctx).await; - } - - let tls_config = if tls_enabled { - let cert_path = crate::config::expand_tilde(&app_config.server.tls.cert_path); - let key_path = crate::config::expand_tilde(&app_config.server.tls.key_path); - Some(extenddb_server::ServerTlsConfig { - cert_path: std::path::PathBuf::from(cert_path), - key_path: std::path::PathBuf::from(key_path), - }) - } else { - None - }; - - extenddb_server::start_server( - listener, - state, - Some(pid_file_path(&run_dir, port)), - tls_config, - ) - .await?; - - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::ServeArgs; - use clap::Parser; - - /// Test wrapper so clap has a top-level `Parser` to drive `ServeArgs`. - #[derive(Parser)] - struct TestCli { - #[command(flatten)] - args: ServeArgs, - } - - fn parse(argv: &[&str]) -> ServeArgs { - TestCli::try_parse_from(argv) - .expect("ServeArgs should parse from valid argv") - .args - } - - #[test] - fn defaults_run_in_daemon_mode() { - // No --foreground flag preserves the historical daemon behavior so - // existing users and scripts are unaffected by the new flag. - let args = parse(&["extenddb-serve"]); - assert!(!args.foreground); - assert_eq!(args.config, "extenddb.toml"); - assert!(args.port.is_none()); - } - - #[test] - fn foreground_flag_is_recognized() { - let args = parse(&["extenddb-serve", "--foreground"]); - assert!(args.foreground); - } - - #[test] - fn no_daemon_alias_is_recognized() { - // The issue proposed either `--foreground` or `--no-daemon`; make - // sure the alias keeps working so users have a choice. - let args = parse(&["extenddb-serve", "--no-daemon"]); - assert!(args.foreground); - } - - #[test] - fn foreground_combines_with_other_flags() { - let args = parse(&[ - "extenddb-serve", - "--config", - "/etc/extenddb/extenddb.toml", - "--port", - "9000", - "--foreground", - ]); - assert!(args.foreground); - assert_eq!(args.config, "/etc/extenddb/extenddb.toml"); - assert_eq!(args.port, Some(9000)); - } - - #[test] - fn unknown_flag_is_rejected() { - // Guard against accidental future renames silently dropping the flag. - let result = TestCli::try_parse_from(["extenddb-serve", "--daemon-off"]); - assert!(result.is_err()); - } -} diff --git a/crates/bin/src/main.rs b/crates/bin/src/main.rs index b4f61bf0..5d71c9e7 100755 --- a/crates/bin/src/main.rs +++ b/crates/bin/src/main.rs @@ -1,137 +1,25 @@ // Copyright 2026 ExtendDB contributors // SPDX-License-Identifier: Apache-2.0 -//! extenddb binary — entry point for the Virtual `DynamoDB` server. +//! extenddb — the PostgreSQL-backed ExtendDB server binary. //! -//! Provides subcommands for server operation and lifecycle management: -//! `serve`, `init`, `destroy`, `verify`, `migrate`, `status`, `stop`, `settings`. -//! Running with no subcommand prints version information. - -mod cmd_catalog_check; -mod cmd_destroy; -mod cmd_init; -mod cmd_manage; -mod cmd_migrate; -mod cmd_serve; -mod cmd_settings; -mod cmd_status; -mod cmd_stop; -mod cmd_verify; -mod config; -mod init_helpers; -mod manage_http; -mod manage_types; -mod serve_helpers; -mod util; -mod workers; - -use clap::{Parser, Subcommand}; - -#[derive(Parser)] -#[command(name = "extenddb", about = "ExtendDB — DynamoDB-compatible API server")] -struct Cli { - /// Print version and exit - #[arg(short = 'V', long)] - version: bool, - - #[command(subcommand)] - command: Option, -} - -#[derive(Subcommand)] -enum Command { - /// Start the extenddb server - Serve(cmd_serve::ServeArgs), - /// Initialize a new extenddb deployment - Init(cmd_init::InitArgs), - /// Tear down a extenddb deployment - Destroy(cmd_destroy::DestroyArgs), - /// Validate a extenddb deployment - Verify(cmd_verify::VerifyArgs), - /// Apply catalog schema migrations - Migrate(cmd_migrate::MigrateArgs), - /// Check if the extenddb server is running - Status(cmd_status::StatusArgs), - /// Stop the running extenddb server - Stop(cmd_stop::StopArgs), - /// Read or write runtime settings - Settings(cmd_settings::SettingsArgs), - /// Manage admin users and accounts via the management API - Manage(cmd_manage::ManageArgs), - /// Check catalog and data database integrity - CatalogCheck(cmd_catalog_check::CatalogCheckArgs), - /// Print version, catalog version, git commit, and build timestamp - Version, -} +//! This is the reference thin bin for the per-backend packaging model: it +//! installs exactly one backend and hands off to the shared `extenddb-app` CLI. +//! A third-party backend author copies this file, swaps the `backend()` call for +//! their crate, and ships their own `extenddb-` image — with no edits to +//! any ExtendDB core crate. fn main() -> anyhow::Result<()> { - let cli = Cli::parse(); - - if cli.version { - print_version(); - return Ok(()); - } - - match cli.command.unwrap_or(Command::Version) { - Command::Serve(args) => cmd_serve::run(&args), - Command::Init(args) => { - let code = run_interactive(cmd_init::run(args))?; - if code != 0 { - std::process::exit(i32::from(code)); - } - Ok(()) - } - Command::Destroy(args) => run_interactive(cmd_destroy::run(args)), - Command::Verify(args) => run_interactive(cmd_verify::run(args)), - Command::Migrate(args) => run_interactive(cmd_migrate::run(args)), - Command::Status(args) => { - cmd_status::run(&args); - Ok(()) - } - Command::Stop(args) => { - cmd_stop::run(&args); - Ok(()) - } - Command::Settings(args) => run_interactive(cmd_settings::run(args)), - Command::Manage(args) => run_interactive(cmd_manage::run(args)), - Command::CatalogCheck(args) => run_interactive(cmd_catalog_check::run(args)), - Command::Version => { - print_version(); - Ok(()) - } - } -} - -/// Print version, catalog version, git commit hash, and build timestamp. -fn print_version() { - println!("extenddb {}", env!("CARGO_PKG_VERSION")); - - // Report catalog version(s) for all registered backend(s) - let backends = extenddb_storage::operations::list_operations_backends(); - if backends.is_empty() { - println!("catalog unknown (no backends registered)"); - } else { - for backend in backends { - let version = extenddb_storage::operations::catalog_version(backend) - .unwrap_or_else(|_| "unknown".to_string()); - println!("catalog {version} ({backend})"); - } - } - - println!("commit {}", env!("EXTENDDB_GIT_HASH")); - println!("built {}", env!("EXTENDDB_BUILD_TIME")); -} - -/// Run an async subcommand with a single-threaded tokio runtime and stderr logging. -/// All non-serve subcommands are interactive (D-24). -fn run_interactive( - future: impl std::future::Future>, -) -> anyhow::Result { - tracing_subscriber::fmt() - .try_init() - .unwrap_or_else(|e| eprintln!("Warning: logging init failed: {e}")); - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build()? - .block_on(future) + // Install the compiled-in backend before dispatch. The compiler checks this + // call; there is no link-time auto-registration and no name to resolve, so a + // missing or mistyped backend cannot become a runtime error. + extenddb_storage::set_backend(extenddb_storage_postgres::backend())?; + + extenddb_app::run(extenddb_app::BuildInfo { + // Read from the bin crate so the reported version is the deployed + // artifact's, not a library crate's. + version: env!("CARGO_PKG_VERSION"), + git_hash: env!("EXTENDDB_GIT_HASH"), + build_time: env!("EXTENDDB_BUILD_TIME"), + }) } diff --git a/crates/config/Cargo.toml b/crates/config/Cargo.toml new file mode 100644 index 00000000..43f3215d --- /dev/null +++ b/crates/config/Cargo.toml @@ -0,0 +1,17 @@ +# Copyright 2026 ExtendDB contributors +# SPDX-License-Identifier: Apache-2.0 +[package] +name = "extenddb-config" +version.workspace = true +edition.workspace = true +rust-version.workspace = true +license.workspace = true + +[dependencies] +extenddb-core = { workspace = true } +extenddb-storage = { workspace = true } +serde = { workspace = true } +toml = { workspace = true } +config = { workspace = true } +anyhow = { workspace = true } +tracing = { workspace = true } diff --git a/crates/config/src/display.rs b/crates/config/src/display.rs new file mode 100644 index 00000000..06c78a83 --- /dev/null +++ b/crates/config/src/display.rs @@ -0,0 +1,131 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 +//! Configuration display and redaction helpers for the console settings page. + +use crate::AppConfig; + +/// Keys whose values must be redacted in configuration displays. +/// +/// The single source of truth for redaction patterns. Consumers (including the +/// console settings page) call [`should_redact`] rather than keeping their own +/// copy of this list. +const REDACTED_CONFIG_KEYS: &[&str] = &[ + "connection_string", + "encryption_key", + "password", + "secret", + "token", +]; + +/// Return `true` if a configuration or settings key's value must be redacted +/// before it is displayed. +/// +/// Matching is case-insensitive and substring-based, so `DATA_DB_PASSWORD` and +/// `storage.connection_string` both redact. +#[must_use] +pub fn should_redact(key: &str) -> bool { + let lower = key.to_lowercase(); + REDACTED_CONFIG_KEYS.iter().any(|p| lower.contains(p)) +} + +/// Return `"••••••••"` if `key` matches a redaction pattern, else `val`. +fn redact_if_sensitive(key: &str, val: &str) -> String { + if should_redact(key) { + "••••••••".to_owned() + } else { + val.to_owned() + } +} + +/// D9: Build static configuration entries for the console settings page. +/// +/// Extracts key-value pairs from the parsed `AppConfig` and pre-redacts +/// sensitive values (connection strings, passwords, keys). +#[must_use] +pub fn build_config_entries(cfg: &AppConfig) -> Vec<(String, String)> { + let r = redact_if_sensitive; + let backend = &cfg.storage.backend; + let mut entries = vec![ + ("server.bind_addr".into(), cfg.server.bind_addr.clone()), + ("server.port".into(), cfg.server.port.to_string()), + ("server.region".into(), cfg.server.region.clone()), + ("server.run_dir".into(), cfg.server.run_dir.clone()), + ( + "server.tls.enabled".into(), + cfg.server.tls.enabled.to_string(), + ), + ( + "server.tls.cert_path".into(), + cfg.server.tls.cert_path.clone(), + ), + ( + "server.tls.key_path".into(), + cfg.server.tls.key_path.clone(), + ), + ( + "server.throttling_enabled".into(), + cfg.server + .throttling_enabled + .map_or("none".into(), |b| b.to_string()), + ), + ( + format!("storage.{backend}.connection_string"), + r("connection_string", cfg.storage.connection_config()), + ), + ( + format!("storage.{backend}.pool_size"), + cfg.storage.max_connections().to_string(), + ), + ( + format!("storage.{backend}.catalog_pool_size"), + cfg.storage.max_catalog_connections().to_string(), + ), + ("auth.provider".into(), cfg.auth.provider.clone()), + ("logging.level".into(), cfg.logging.level.clone()), + ("logging.format".into(), cfg.logging.format.clone()), + ("docs_dir".into(), cfg.docs_dir.clone().unwrap_or_default()), + ( + "import.paths".into(), + if cfg.import_config.paths.is_empty() { + "none".into() + } else { + cfg.import_config.paths.join(", ") + }, + ), + ( + "export.paths".into(), + if cfg.export_config.paths.is_empty() { + "none".into() + } else { + cfg.export_config.paths.join(", ") + }, + ), + ]; + + // Commonly adjusted limits (full list in [limits] section of extenddb.sample.toml). + let lim = &cfg.limits; + entries.extend([ + ( + "limits.max_item_size_bytes".into(), + lim.max_item_size_bytes.to_string(), + ), + ( + "limits.max_tables_per_account".into(), + lim.max_tables_per_account.to_string(), + ), + ( + "limits.max_gsis_per_table".into(), + lim.max_gsis_per_table.to_string(), + ), + ( + "limits.allow_multipart_table_keys".into(), + lim.allow_multipart_table_keys.to_string(), + ), + ( + "limits.max_import_file_bytes".into(), + lim.max_import_file_bytes.to_string(), + ), + ]); + + entries +} diff --git a/crates/bin/src/config.rs b/crates/config/src/lib.rs old mode 100755 new mode 100644 similarity index 72% rename from crates/bin/src/config.rs rename to crates/config/src/lib.rs index e9c6cb60..5a693ecf --- a/crates/bin/src/config.rs +++ b/crates/config/src/lib.rs @@ -1,7 +1,16 @@ // Copyright 2026 ExtendDB contributors // SPDX-License-Identifier: Apache-2.0 -//! Shared configuration types for the extenddb binary. +//! Configuration types and loading for ExtendDB. +//! +//! Owns [`AppConfig`] and its subsections, config-file loading, redaction +//! helpers, and runtime-path helpers (PID file). This crate is +//! backend-agnostic: it depends only on the `extenddb-storage` trait surface +//! (never on a concrete backend), so both `extenddb-server` (for `serve`) and +//! the CLI/app layer can depend on it without pulling in a backend or forming +//! a dependency cycle. + +use std::path::PathBuf; use extenddb_core::limits::LimitsConfig; use serde::Deserialize; @@ -153,36 +162,43 @@ impl<'de> serde::Deserialize<'de> for StorageConfig { // Deserialize into a raw TOML value first let value: toml::Value = toml::Value::deserialize(deserializer)?; - // Extract the backend field - let backend = value - .get("backend") - .and_then(|v| v.as_str()) - .unwrap_or("postgres") - .to_string(); + // The installed backend is authoritative: it supplies the name used to + // locate its `[storage.]` section. The file's `backend` key is + // therefore optional, and when present it is validated against the + // compiled-in backend rather than used to choose one. A binary contains + // exactly one backend, so a mismatch is an operator error worth naming + // instead of silently ignoring. + let backend = extenddb_storage::backend_name().ok_or_else(|| { + D::Error::custom("no storage backend installed (set_backend was not called)") + })?; + + if let Some(requested) = value.get("backend").and_then(|v| v.as_str()) + && requested != backend + { + return Err(D::Error::custom(format!( + "[storage] backend = \"{requested}\" does not match this binary's \ + compiled-in backend \"{backend}\". This binary can only serve \ + \"{backend}\"; either remove the key or install the \ + extenddb-{requested} binary." + ))); + } // Get the backend-specific table (e.g., [storage.postgres]) let backend_table: &toml::Table = value - .get(&backend) + .get(backend) .and_then(|v| v.as_table()) .ok_or_else(|| { D::Error::custom(format!("Missing [storage.{backend}] section in config")) })?; - // Use the registry to deserialize the backend config - let config = extenddb_storage::config::deserialize_storage_config(&backend, backend_table) + // Hand the section to the installed backend's deserializer. + let config = extenddb_storage::config::deserialize_storage_config(backend_table) .map_err(D::Error::custom)?; - Ok(StorageConfig { backend, config }) - } -} - -#[cfg(feature = "postgres")] -impl Default for StorageConfig { - fn default() -> Self { - Self { - backend: default_backend(), - config: Box::new(extenddb_storage_postgres::PostgresStorageConfig::default()), - } + Ok(StorageConfig { + backend: backend.to_owned(), + config, + }) } } @@ -303,6 +319,7 @@ fn default_run_dir() -> String { /// Expand a leading `~` in a path to `$HOME`. Returns the input unchanged /// if `$HOME` is unset or the path does not start with `~`. +#[must_use] pub fn expand_tilde(path: &str) -> String { if let Some(rest) = path.strip_prefix('~') && (rest.is_empty() || rest.starts_with('/')) @@ -312,10 +329,7 @@ pub fn expand_tilde(path: &str) -> String { } path.to_owned() } -#[cfg(feature = "postgres")] -fn default_backend() -> String { - "postgres".to_owned() -} + fn default_tls_enabled() -> bool { true } @@ -359,12 +373,13 @@ pub fn load(config_path: &str) -> anyhow::Result { /// /// Uses the backend-specific operations engine to handle different connection /// string formats (`PostgreSQL`). -pub fn redact_password(backend: &str, conn: &str) -> String { - extenddb_storage::operations::redact_connection_string(backend, conn) - .unwrap_or_else(|_| conn.to_owned()) +#[must_use] +pub fn redact_password(conn: &str) -> String { + extenddb_storage::operations::redact_connection_string(conn).unwrap_or_else(|_| conn.to_owned()) } /// Return the current OS username, falling back to given default username: e.g. `"postgres"`. +#[must_use] pub fn whoami(default: &str) -> String { std::env::var("USER").unwrap_or_else(|_| default.to_owned()) } @@ -378,125 +393,29 @@ pub fn whoami(default: &str) -> String { /// # Errors /// /// Returns an error describing the invalid character found. -pub fn validate_identifier(backend: &str, name: &str, label: &str) -> anyhow::Result<()> { - extenddb_storage::operations::validate_identifier(backend, name, label) +pub fn validate_identifier(name: &str, label: &str) -> anyhow::Result<()> { + extenddb_storage::operations::validate_identifier(name, label) .map_err(|e| anyhow::anyhow!("{e:?}")) } -/// Keys whose values must be redacted in configuration displays. +/// PID file path for a given port and run directory. /// -/// Canonical list — keep in sync with `REDACTED_KEYS` in -/// `crates/server/src/console/pages/settings_pages.rs`. -const REDACTED_CONFIG_KEYS: &[&str] = &[ - "connection_string", - "encryption_key", - "password", - "secret", - "token", -]; - -/// Return `"••••••••"` if `key` matches a redaction pattern, else `val`. -fn redact_if_sensitive(key: &str, val: &str) -> String { - let lower = key.to_lowercase(); - if REDACTED_CONFIG_KEYS.iter().any(|p| lower.contains(p)) { - "••••••••".to_owned() - } else { - val.to_owned() - } +/// Used by `serve` (write) and `status`/`stop` (read). +#[must_use] +pub fn pid_file_path(run_dir: &str, port: u16) -> PathBuf { + PathBuf::from(format!("{run_dir}/extenddb-{port}.pid")) } -/// D9: Build static configuration entries for the console settings page. -/// -/// Extracts key-value pairs from the parsed `AppConfig` and pre-redacts -/// sensitive values (connection strings, passwords, keys). -pub fn build_config_entries(cfg: &AppConfig) -> Vec<(String, String)> { - let r = redact_if_sensitive; - let backend = &cfg.storage.backend; - let mut entries = vec![ - ("server.bind_addr".into(), cfg.server.bind_addr.clone()), - ("server.port".into(), cfg.server.port.to_string()), - ("server.region".into(), cfg.server.region.clone()), - ("server.run_dir".into(), cfg.server.run_dir.clone()), - ( - "server.tls.enabled".into(), - cfg.server.tls.enabled.to_string(), - ), - ( - "server.tls.cert_path".into(), - cfg.server.tls.cert_path.clone(), - ), - ( - "server.tls.key_path".into(), - cfg.server.tls.key_path.clone(), - ), - ( - "server.throttling_enabled".into(), - cfg.server - .throttling_enabled - .map_or("none".into(), |b| b.to_string()), - ), - ( - format!("storage.{backend}.connection_string"), - r("connection_string", cfg.storage.connection_config()), - ), - ( - format!("storage.{backend}.pool_size"), - cfg.storage.max_connections().to_string(), - ), - ( - format!("storage.{backend}.catalog_pool_size"), - cfg.storage.max_catalog_connections().to_string(), - ), - ("auth.provider".into(), cfg.auth.provider.clone()), - ("logging.level".into(), cfg.logging.level.clone()), - ("logging.format".into(), cfg.logging.format.clone()), - ("docs_dir".into(), cfg.docs_dir.clone().unwrap_or_default()), - ( - "import.paths".into(), - if cfg.import_config.paths.is_empty() { - "none".into() - } else { - cfg.import_config.paths.join(", ") - }, - ), - ( - "export.paths".into(), - if cfg.export_config.paths.is_empty() { - "none".into() - } else { - cfg.export_config.paths.join(", ") - }, - ), - ]; - - // Commonly adjusted limits (full list in [limits] section of extenddb.sample.toml). - let lim = &cfg.limits; - entries.extend([ - ( - "limits.max_item_size_bytes".into(), - lim.max_item_size_bytes.to_string(), - ), - ( - "limits.max_tables_per_account".into(), - lim.max_tables_per_account.to_string(), - ), - ( - "limits.max_gsis_per_table".into(), - lim.max_gsis_per_table.to_string(), - ), - ( - "limits.allow_multipart_table_keys".into(), - lim.allow_multipart_table_keys.to_string(), - ), - ( - "limits.max_import_file_bytes".into(), - lim.max_import_file_bytes.to_string(), - ), - ]); - - entries +/// PID file path using the default run directory. Used by `status` when +/// no config file is loaded. +#[must_use] +pub fn pid_file_path_default(port: u16) -> PathBuf { + pid_file_path(&ServerConfig::default().run_dir, port) } +mod display; +pub use display::{build_config_entries, should_redact}; + #[cfg(test)] mod tests { use super::*; @@ -519,4 +438,12 @@ mod tests { // ~user should NOT be expanded (we only handle ~/...) assert_eq!(expand_tilde("~user/foo"), "~user/foo"); } + + #[test] + fn pid_file_path_formats_port() { + assert_eq!( + pid_file_path("/run/extenddb", 18443), + PathBuf::from("/run/extenddb/extenddb-18443.pid") + ); + } } diff --git a/crates/server/Cargo.toml b/crates/server/Cargo.toml index 726b40e3..e8495b8a 100755 --- a/crates/server/Cargo.toml +++ b/crates/server/Cargo.toml @@ -10,6 +10,7 @@ license.workspace = true [dependencies] extenddb-cache = { workspace = true } extenddb-core = { workspace = true } +extenddb-config = { workspace = true } extenddb-engine = { workspace = true } extenddb-storage = { workspace = true } extenddb-auth = { workspace = true } @@ -34,6 +35,9 @@ aes-gcm = { workspace = true } rand = { workspace = true } axum-server = { workspace = true } rustls = { workspace = true } +tracing-subscriber = { workspace = true } +syslog-tracing = { workspace = true } +libc = { workspace = true } [dev-dependencies] extenddb-cache = { workspace = true, features = ["test-util"] } diff --git a/crates/server/src/console/pages/settings_pages.rs b/crates/server/src/console/pages/settings_pages.rs index 7d0091f3..e2efa500 100755 --- a/crates/server/src/console/pages/settings_pages.rs +++ b/crates/server/src/console/pages/settings_pages.rs @@ -16,24 +16,12 @@ use axum::response::{Html, IntoResponse, Response}; use crate::console::ConsoleState; use crate::console::html; +// Redaction patterns live in `extenddb-config` so the console and the static +// config display cannot drift apart. +use extenddb_config::should_redact; use super::{identity_label, is_admin, require_session}; -/// Keys whose values must be redacted in the settings display. -const REDACTED_KEYS: &[&str] = &[ - "connection_string", - "encryption_key", - "password", - "secret", - "token", -]; - -/// Check if a settings key should have its value redacted. -fn should_redact(key: &str) -> bool { - let lower = key.to_lowercase(); - REDACTED_KEYS.iter().any(|&pattern| lower.contains(pattern)) -} - /// Known runtime settings with their default values. /// These are the settings that can be changed via `extenddb settings set`. const RUNTIME_DEFAULTS: &[(&str, &str)] = &[ diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 2c4d9132..b7d0dc47 100755 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -18,7 +18,11 @@ mod metrics_endpoint; pub mod rate_limit; mod request_helpers; mod response; +mod serve; mod throttle_helpers; +mod workers; + +pub use serve::{BuildInfo, LogTarget, ServeParams, log_to_syslog_raw, serve}; use std::path::PathBuf; use std::sync::Arc; diff --git a/crates/server/src/serve.rs b/crates/server/src/serve.rs new file mode 100644 index 00000000..76f72a79 --- /dev/null +++ b/crates/server/src/serve.rs @@ -0,0 +1,618 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The `serve` library entrypoint. +//! +//! A backend's thin `main` loads config, binds the listening socket, and then +//! calls [`serve`] to run the server. There is no backend selection: `serve` +//! assembles server components from the single backend installed via +//! [`set_backend`](extenddb_storage::set_backend), wires the auth/authz/table-key +//! caches and [`AppState`](crate::AppState), spawns the generic + backend workers, +//! and serves until shutdown. Daemonization, PID-file creation, and CLI argument +//! handling stay in the app/CLI layer that calls this function. + +use std::net::TcpListener; +use std::sync::Arc; +use std::time::Duration; + +use extenddb_config as config; +use extenddb_config::pid_file_path; +use extenddb_storage::CancellationToken; +use syslog_tracing::{Facility, Options, Syslog}; +use tracing_subscriber::{ + EnvFilter, Layer, fmt, fmt::writer::BoxMakeWriter, layer::SubscriberExt, reload, + util::SubscriberInitExt, +}; + +use crate::AppState; +use crate::workers; + +/// Build provenance of the deployed binary. +/// +/// The library crates cannot read the bin's `build.rs` environment variables or +/// its package version, so the thin `main` passes them in. Surfaced by +/// `extenddb version`, the startup banner, and the console version string. +/// +/// All fields are `&'static str` because every value originates from a compile +/// time `env!` and is baked into the binary. Declaring the true lifetime up +/// front means the values can later be stored beyond the call (in a struct, a +/// metrics label, a spawned task) without a breaking signature change. +#[derive(Debug, Clone, Copy)] +pub struct BuildInfo { + /// Package version of the deployed binary (e.g. `env!("CARGO_PKG_VERSION")` + /// from the bin crate — not from a library crate, whose version may drift). + pub version: &'static str, + /// Short git commit hash of the build (e.g. `env!("EXTENDDB_GIT_HASH")`). + pub git_hash: &'static str, + /// Build timestamp (e.g. `env!("EXTENDDB_BUILD_TIME")`). + pub build_time: &'static str, +} + +/// Where the server writes its log output. +/// +/// This is the library's whole view of the deployment model: it decides where +/// logs go and nothing else. Whether the process daemonized, runs under a +/// container, or is supervised by systemd is the caller's concern. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum LogTarget { + /// Write logs to the POSIX syslog. Used by the daemonized deployment, where + /// stderr is `/dev/null` after the double fork. + Syslog, + /// Write logs to stderr, for a container or process supervisor that + /// captures the process's own streams. + Stderr, +} + +impl LogTarget { + /// Short label for the startup banner. + const fn label(self) -> &'static str { + match self { + Self::Syslog => "syslog", + Self::Stderr => "stderr", + } + } +} + +/// Everything [`serve`] needs to run a server. +/// +/// Marked `#[non_exhaustive]`, so construct it with [`ServeParams::new`] and the +/// `with_*` methods rather than a struct literal — new fields can then be added +/// without breaking third-party backends. +#[non_exhaustive] +pub struct ServeParams { + /// Parsed application configuration. + pub app_config: config::AppConfig, + /// Already-bound listening socket. The caller binds before daemonizing so + /// port conflicts surface on stderr before the parent exits. The listening + /// port is read from this socket, so it cannot disagree with it. + pub listener: TcpListener, + /// Directory holding the PID file. + pub run_dir: String, + /// Where log output is written. + pub log_target: LogTarget, + /// Build provenance of the deployed binary. + pub build: BuildInfo, +} + +impl ServeParams { + /// Create parameters that log to syslog (the daemon default). + #[must_use] + pub fn new( + app_config: config::AppConfig, + listener: TcpListener, + run_dir: String, + build: BuildInfo, + ) -> Self { + Self { + app_config, + listener, + run_dir, + log_target: LogTarget::Syslog, + build, + } + } + + /// Send log output to `target` instead of the syslog default. + #[must_use] + pub fn with_log_target(mut self, target: LogTarget) -> Self { + self.log_target = target; + self + } +} + +/// Run the ExtendDB server on the pre-bound listener in `params` until shutdown. +/// +/// On any error before the HTTP server starts, the PID file is removed and a +/// fatal message is written to the configured [`LogTarget`]. +/// +/// # Errors +/// +/// Returns an error if logging init, backend component creation, cache +/// configuration, path resolution, or the HTTP server fails. +pub async fn serve(params: ServeParams) -> anyhow::Result<()> { + // The listener is the single source of truth for the port: it is already + // bound, so reading it back cannot disagree with the caller's intent (and + // resolves port 0 to the kernel-assigned port). + let port = params + .listener + .local_addr() + .map_err(|e| anyhow::anyhow!("Failed to read listener address: {e}"))? + .port(); + + // CB-27: Clean up PID file if serve fails before reaching the HTTP server + // (e.g., backend connection failure). The PID file was already written by + // the daemonize step in the caller. + let pid_path = pid_file_path(¶ms.run_dir, port); + let log_target = params.log_target; + let result = serve_inner(params, port).await; + if let Err(ref e) = result { + let _ = std::fs::remove_file(&pid_path); + // P57 Bug 7: Log fatal errors where the operator will see them. After + // daemonize, stderr is /dev/null so anyhow's error display is lost. Use + // tracing if available, fall back to the raw writer if tracing isn't + // initialized yet. + tracing::error!("extenddb fatal: {e:#}"); + match log_target { + LogTarget::Stderr => eprintln!("extenddb fatal: {e:#}"), + LogTarget::Syslog => log_to_syslog_raw(&format!("extenddb fatal: {e:#}")), + } + } + result +} + +/// Inner serve function — separated so [`serve`] can clean up the PID file on +/// any error path. +async fn serve_inner(params: ServeParams, port: u16) -> anyhow::Result<()> { + let ServeParams { + app_config, + listener: std_listener, + run_dir, + log_target, + build, + } = params; + let catalog_version = + extenddb_storage::operations::catalog_version().unwrap_or_else(|_| "unknown".to_string()); + + // Write the PID file so `extenddb status`/`stop` and `start_server`'s + // graceful shutdown cleanup work in every deployment. When the caller + // daemonized, the grandchild PID that `daemonize` wrote is the same value + // as `std::process::id()` post-fork, so this is a consistent rewrite rather + // than a conflicting one. + let pid_file = pid_file_path(&run_dir, port); + std::fs::write(&pid_file, format!("{}\n", std::process::id())) + .map_err(|e| anyhow::anyhow!("Failed to write PID file {}: {e}", pid_file.display()))?; + + // Init logging (REQ-LOG-003, REQ-LOG-006) — the caller chose the target via + // [`LogTarget`]; a supervised/container deployment picks stderr so the + // supervisor can capture logs. + // D-3: sqlx messages are controlled by an independent `sqlx_log_level` + // runtime setting (default: warn). Both extenddb and sqlx messages use the + // `extenddb` syslog identifier (POSIX syslog supports only one identity per + // process). sqlx messages are identifiable by their `sqlx::query` target. + // Filter with: `journalctl -t extenddb | grep -v sqlx` (exclude) or + // `journalctl -t extenddb | grep sqlx` (include only). + // + // The EnvFilter encodes both levels: `{app_level},sqlx={sqlx_level}`. + // The poll_log_level worker reloads the filter when either setting changes. + let filter_str = format!("{},sqlx=warn", app_config.logging.level); + // CB-29: Always use the config file log level, never RUST_LOG. The runtime + // settings poller handles dynamic level changes. RUST_LOG silently + // overriding the config is an operational surprise. + let filter = EnvFilter::new(&filter_str); + let (filter_layer, reload_handle) = reload::Layer::new(filter); + + // Pick the writer first (stderr vs syslog), then the format (text vs json). + // syslog supplies its own timestamps, so we strip them with + // `.without_time()` only on the syslog path. + let (writer, with_time): (BoxMakeWriter, bool) = match log_target { + LogTarget::Stderr => (BoxMakeWriter::new(std::io::stderr), true), + LogTarget::Syslog => { + let syslog = Syslog::new( + c"extenddb", + Options::LOG_PID | Options::LOG_NDELAY, + Facility::Daemon, + ) + .ok_or_else(|| { + anyhow::anyhow!( + "Failed to initialize syslog — another syslog logger may already be active" + ) + })?; + (BoxMakeWriter::new(syslog), false) + } + }; + + let fmt_layer = match (with_time, app_config.logging.format == "json") { + (true, true) => fmt::layer().json().with_writer(writer).boxed(), + (true, false) => fmt::layer().with_writer(writer).boxed(), + (false, true) => fmt::layer() + .json() + .without_time() + .with_writer(writer) + .boxed(), + (false, false) => fmt::layer().without_time().with_writer(writer).boxed(), + }; + + tracing_subscriber::registry() + .with(filter_layer) + .with(fmt_layer) + .try_init() + .map_err(|e| anyhow::anyhow!("Failed to initialize tracing: {e}"))?; + + // Create server components via factory pattern + let components = extenddb_storage::create_server_components( + app_config.storage.as_trait(), + &app_config.server.region, + ) + .await?; + + let storage = components.engine; + let catalog_store = components.catalog_store; + let cred_store = components.credential_store; + let runtime_hooks = components.runtime_hooks; + + // Build SwrCacheConfig values from the [auth.cache] TOML section. + let cache_cfg = &app_config.auth.cache; + let cache_enabled = cache_cfg.enabled; + let make_cache_cfg = |name: &'static str| -> extenddb_cache::SwrCacheConfig { + extenddb_cache::SwrCacheConfig { + ttl: std::time::Duration::from_secs(cache_cfg.ttl_seconds), + soft_ttl: std::time::Duration::from_secs(cache_cfg.soft_ttl_seconds), + negative_ttl: std::time::Duration::from_secs(cache_cfg.negative_ttl_seconds), + max_entries: cache_cfg.max_entries, + name, + } + }; + // Validate config eagerly so misconfiguration fails fast at startup. + // Today every named subcache shares the same TTL/max_entries shape (only + // `name` differs), so a single `validate()` check suffices. If per-cache + // tuning is ever added, validate every constructed config here. + if let Err(e) = make_cache_cfg("__validate__").validate() { + anyhow::bail!( + "Invalid [auth.cache] configuration: {e}. Check ttl_seconds, \ + soft_ttl_seconds, negative_ttl_seconds, max_entries." + ); + } + if !cache_enabled { + tracing::warn!( + "auth.cache.enabled = false — auth/authz caches are in pass-through mode \ + (every lookup hits the catalog directly)" + ); + } + + // Phase 2: Wrap the raw credential store. In pass-through mode the + // wrapper bypasses the cache and forwards every lookup to the inner + // store; otherwise it caches per the TOML config. + let cached_cred_store = Arc::new(if cache_enabled { + extenddb_auth::CachedCredentialStore::with_arc(cred_store, make_cache_cfg("credential")) + } else { + extenddb_auth::CachedCredentialStore::pass_through_arc( + cred_store, + make_cache_cfg("credential"), + ) + }); + let auth: Arc = Arc::new( + extenddb_auth::BuiltinAuthProvider::new((*cached_cred_store).clone()), + ); + + // Phase 3: Build the authorization cache. + let authz_cache: Arc = { + let store: Arc = + catalog_store.clone(); + let cfg = crate::AuthzCacheConfig { + identity_policies: make_cache_cfg("identity_policies"), + group_policies: make_cache_cfg("group_policies"), + boundary: make_cache_cfg("boundary"), + principal_tags: make_cache_cfg("principal_tags"), + resource_tags: make_cache_cfg("resource_tags"), + session_data: make_cache_cfg("session_data"), + }; + Arc::new(if cache_enabled { + crate::CachedAuthzStore::new(store, cfg) + } else { + crate::CachedAuthzStore::pass_through(store, cfg) + }) + }; + + // Phase 4: Build the TableKeyInfo cache. + let table_key_info_cache: Arc = Arc::new(if cache_enabled { + crate::CachedTableKeyInfoStore::new(storage.clone(), make_cache_cfg("table_key_info")) + } else { + crate::CachedTableKeyInfoStore::pass_through( + storage.clone(), + make_cache_cfg("table_key_info"), + ) + }); + + // Assemble the cache registry threaded into AppState for write-through + // invalidations from the management API. + let auth_cache = + extenddb_auth::AuthCacheRegistry::empty() + .with_credential(cached_cred_store) + .with_authz_invalidator( + authz_cache.clone() as Arc + ) + .with_table_key_info_invalidator(table_key_info_cache.clone() + as Arc); + + let data_db_info = runtime_hooks + .as_ref() + .and_then(|h| h.backend_info()) + .unwrap_or_else(|| "(unknown)".to_owned()); + + // REQ-LOG-001: Startup banner with effective configuration. + // REQ-LOG-002: Connection strings redact passwords. + tracing::info!( + "extenddb {} (catalog {}) starting — bind={}:{}, region={}, auth={}, catalog_db={}, data_db={}, log_output={}, log_level={}", + build.version, + catalog_version, + app_config.server.bind_addr, + port, + app_config.server.region, + app_config.auth.provider, + config::redact_password(app_config.storage.connection_config()), + data_db_info, + log_target.label(), + app_config.logging.level, + ); + + // Convert pre-bound std listener to tokio (D-4: bind before fork). + let listener = tokio::net::TcpListener::from_std(std_listener)?; + + // P120e: Create metrics collector early so workers can record health. + let metrics = Arc::new(extenddb_core::metrics::MetricsCollector::new()); + + let tls_enabled = app_config.server.tls.enabled; + + // P53: Resolve import and export path lists. Supports both the new + // [import]/[export] sections and the deprecated import_export_root. + let resolve_paths = |raw_paths: &[String], + label: &str| + -> anyhow::Result>> { + let mut resolved = Vec::new(); + for raw in raw_paths { + let expanded = config::expand_tilde(raw); + let path = std::path::PathBuf::from(&expanded); + if !path.exists() { + std::fs::create_dir_all(&path) + .map_err(|e| anyhow::anyhow!("Cannot create {label} path {expanded}: {e}"))?; + } + let canonical = path + .canonicalize() + .map_err(|e| anyhow::anyhow!("Cannot canonicalize {label} path {expanded}: {e}"))?; + resolved.push(Arc::new(canonical)); + } + Ok(resolved) + }; + + // Build effective path lists: new config takes precedence over deprecated. + let mut import_paths_raw = app_config.import_config.paths.clone(); + let mut export_paths_raw = app_config.export_config.paths.clone(); + if let Some(ref legacy) = app_config.import_export_root { + if import_paths_raw.is_empty() { + import_paths_raw.push(legacy.clone()); + } + if export_paths_raw.is_empty() { + export_paths_raw.push(legacy.clone()); + } + if !app_config.import_config.paths.is_empty() && !app_config.export_config.paths.is_empty() + { + tracing::warn!( + "Both import_export_root and [import]/[export] sections configured; import_export_root is ignored" + ); + } + } + + let import_paths: Arc<[Arc]> = + Arc::from(resolve_paths(&import_paths_raw, "import")?); + let export_paths: Arc<[Arc]> = + Arc::from(resolve_paths(&export_paths_raw, "export")?); + + if import_paths.is_empty() { + tracing::info!("Import disabled (no [import] paths configured)"); + } else { + for p in import_paths.iter() { + tracing::info!("Import enabled, path: {}", p.display()); + } + } + if export_paths.is_empty() { + tracing::info!("Export disabled (no [export] paths configured)"); + } else { + for p in export_paths.iter() { + tracing::info!("Export enabled, path: {}", p.display()); + } + } + + // D9: Build static config entries for the console settings page. + // Must be called before `app_config.limits` is moved. + let config_entries = config::build_config_entries(&app_config); + + // AI-1: Load runtime documentation from docs_dir if configured. + let docs_store = app_config.docs_dir.as_ref().and_then(|raw| { + let expanded = config::expand_tilde(raw); + let path = std::path::PathBuf::from(&expanded); + match crate::console::docs_embed::DocsStore::load(&path) { + Ok(store) => { + tracing::info!("Documentation loaded from {}", path.display()); + Some(store) + } + Err(e) => { + tracing::warn!("Documentation unavailable: {e}"); + None + } + } + }); + + let limits = Arc::new({ + let mut limits = app_config.limits; + if let Some(max_bytes) = app_config.max_import_bytes { + limits.max_import_file_bytes = max_bytes; + } + limits + }); + + let config_throttling = app_config.server.throttling_enabled.unwrap_or(false); + let initial_throttling = catalog_store + .get_setting("throttling_enabled") + .await + .ok() + .flatten() + .map_or(config_throttling, |v| v == "true"); + + let throttle = Arc::new(extenddb_core::throttle::ThrottleManager::new( + limits.per_account_max_rcu, + limits.per_account_max_wcu, + initial_throttling, + )); + + let state = AppState { + storage, + auth, + limits, + region: Arc::from(app_config.server.region.as_str()), + server_addr: format!("localhost:{port}"), + catalog_store: Some(catalog_store.clone()), + version_info: Arc::from( + format!( + "{} · catalog {} · {}", + build.version, catalog_version, build.git_hash, + ) + .as_str(), + ), + metrics: metrics.clone(), + tls_enabled, + import_paths, + export_paths, + throttle: throttle.clone(), + auth_cache, + authz_cache, + table_key_info_cache, + config_entries, + docs_store, + }; + + // Workers run until `shutdown` is cancelled, which happens after the HTTP + // server stops accepting. Handles are collected so shutdown can drain them + // (the metrics flush worker persists its final bucket on the way out) + // instead of leaving the work to a runtime drop. + let shutdown = CancellationToken::new(); + let mut worker_handles = vec![ + // D-22: Poll log_level from the settings table. + tokio::spawn(workers::poll_log_level( + catalog_store.clone(), + reload_handle.clone(), + app_config.logging.level.clone(), + shutdown.clone(), + )), + // Poll the throttling_enabled runtime setting. + tokio::spawn(workers::poll_throttling_enabled( + catalog_store.clone(), + throttle, + config_throttling, + shutdown.clone(), + )), + // Metrics pruning and flushing. + tokio::spawn(workers::metrics_prune_worker( + metrics.clone(), + shutdown.clone(), + )), + tokio::spawn(workers::metrics_flush_worker( + metrics.clone(), + catalog_store.clone(), + shutdown.clone(), + )), + // Clean up old login attempt records. + tokio::spawn(workers::login_attempt_cleanup_worker( + catalog_store.clone(), + shutdown.clone(), + )), + // Phase 11a: Warn about approximate consumed capacity. + tokio::spawn(workers::capacity_warning_worker(shutdown.clone())), + ]; + + // Spawn backend-specific workers via runtime hooks + if let Some(hooks) = runtime_hooks { + let worker_ctx = extenddb_storage::WorkerContext { + metrics: metrics.clone(), + catalog_store: catalog_store.clone(), + reload_handle: reload_handle.clone(), + config_log_level: app_config.logging.level.clone(), + shutdown: shutdown.clone(), + }; + worker_handles.extend(hooks.spawn_workers(&worker_ctx).await); + } + + let tls_config = if tls_enabled { + let cert_path = config::expand_tilde(&app_config.server.tls.cert_path); + let key_path = config::expand_tilde(&app_config.server.tls.key_path); + Some(crate::ServerTlsConfig { + cert_path: std::path::PathBuf::from(cert_path), + key_path: std::path::PathBuf::from(key_path), + }) + } else { + None + }; + + let server_result = crate::start_server( + listener, + state, + Some(pid_file_path(&run_dir, port)), + tls_config, + ) + .await; + + // The HTTP server has stopped accepting; drain the workers so in-flight + // cycles finish and the metrics flush worker writes its final bucket. + drain_workers(&shutdown, worker_handles).await; + + server_result?; + + Ok(()) +} + +/// Cancel the shutdown token and wait for every worker to return. +/// +/// Bounded by `DRAIN_TIMEOUT` so a worker stuck inside a backend call cannot +/// hold the process open; a timeout is logged and the remaining tasks are +/// dropped, which is the pre-drain behavior. +async fn drain_workers(shutdown: &CancellationToken, handles: Vec>) { + const DRAIN_TIMEOUT: Duration = Duration::from_secs(5); + + shutdown.cancel(); + let count = handles.len(); + let drain = futures::future::join_all(handles); + match tokio::time::timeout(DRAIN_TIMEOUT, drain).await { + Ok(results) => { + let panicked = results.iter().filter(|r| r.is_err()).count(); + if panicked > 0 { + tracing::warn!("{panicked} of {count} background worker(s) ended abnormally"); + } else { + tracing::info!("{count} background worker(s) drained"); + } + } + Err(_) => tracing::warn!( + "Background workers did not drain within {}s; shutting down anyway", + DRAIN_TIMEOUT.as_secs() + ), + } +} + +/// P57 Bug 7: Best-effort raw syslog write for fatal errors and panics. +/// +/// Used when the tracing subscriber may not be initialized — during early +/// startup before syslog tracing is configured, and from the caller's panic +/// hook after daemonizing (stderr is `/dev/null` there, so a panic would +/// otherwise be invisible). +pub fn log_to_syslog_raw(msg: &str) { + // SAFETY: openlog/syslog are POSIX-standard C functions. The ident + // string is a static C string literal with 'static lifetime. + unsafe { + libc::openlog( + c"extenddb".as_ptr(), + libc::LOG_PID | libc::LOG_NDELAY, + libc::LOG_DAEMON, + ); + if let Ok(cmsg) = std::ffi::CString::new(msg.to_owned()) { + libc::syslog(libc::LOG_CRIT, c"%s".as_ptr(), cmsg.as_ptr()); + } + } +} diff --git a/crates/bin/src/workers.rs b/crates/server/src/workers.rs similarity index 53% rename from crates/bin/src/workers.rs rename to crates/server/src/workers.rs index c06188af..372fdb45 100755 --- a/crates/bin/src/workers.rs +++ b/crates/server/src/workers.rs @@ -8,34 +8,39 @@ //! TTL cleanup, table size refresh, stream record expiry, idempotency token //! cleanup, capacity warning, and metrics pruning. //! +//! Every worker takes a [`CancellationToken`] and stops at its next tick after +//! the token is cancelled, so `serve` can drain them on shutdown instead of +//! relying on the runtime dropping mid-flight tasks. `metrics_flush_worker` +//! performs a final full flush on cancellation so the last partial bucket is +//! persisted rather than discarded. +//! //! Workers are generic over storage traits so they are decoupled from the //! concrete `PostgresEngine` / `PostgresCatalogStore` types. use std::sync::Arc; +use std::time::Duration; use extenddb_core::throttle::ThrottleManager; use extenddb_storage::management_store::{MetricsStore, RateLimitStore, SettingsStore}; +use extenddb_storage::{CancellationToken, sleep_or_shutdown as tick}; use tracing_subscriber::{EnvFilter, reload}; /// Poll the `log_level` and `sqlx_log_level` settings from the database /// and reload the tracing filter when either changes (D-22, D-3). /// The combined filter is `{log_level},sqlx={sqlx_log_level}`. /// Falls back to `config_level` when `log_level` is absent from the DB. -/// Runs until the process exits. +/// Runs until `token` is cancelled. pub(crate) async fn poll_log_level( store: Arc, handle: reload::Handle, config_level: String, + token: CancellationToken, ) { - use std::time::Duration; - const POLL_INTERVAL: Duration = Duration::from_secs(30); let mut current_level = config_level; let mut current_sqlx_level = String::from("warn"); - loop { - tokio::time::sleep(POLL_INTERVAL).await; - + while tick(&token, POLL_INTERVAL).await { let (log_result, sqlx_result) = tokio::join!( store.get_setting("log_level"), store.get_setting("sqlx_log_level"), @@ -99,15 +104,12 @@ pub(crate) async fn poll_throttling_enabled( store: Arc, throttle: Arc, config_enabled: bool, + token: CancellationToken, ) { - use std::time::Duration; - const POLL_INTERVAL: Duration = Duration::from_secs(30); let mut current = config_enabled; - loop { - tokio::time::sleep(POLL_INTERVAL).await; - + while tick(&token, POLL_INTERVAL).await { let new_enabled = match store.get_setting("throttling_enabled").await { Ok(Some(v)) => v == "true", Ok(None) => config_enabled, @@ -134,15 +136,12 @@ pub(crate) async fn poll_throttling_enabled( /// Phase 11a: `ConsumedCapacity` returns plausible stubs, not real values. /// This worker reads and resets the counter on a fixed interval and emits /// a single log line summarizing usage since the last tick. -pub(crate) async fn capacity_warning_worker() { +pub(crate) async fn capacity_warning_worker(token: CancellationToken) { use extenddb_engine::capacity_helpers::CAPACITY_REQUEST_COUNT; - use std::time::Duration; const WARNING_INTERVAL: Duration = Duration::from_secs(3600); - loop { - tokio::time::sleep(WARNING_INTERVAL).await; - + while tick(&token, WARNING_INTERVAL).await { let count = CAPACITY_REQUEST_COUNT.swap(0, std::sync::atomic::Ordering::Relaxed); if count > 0 { tracing::warn!( @@ -154,12 +153,14 @@ pub(crate) async fn capacity_warning_worker() { } /// Periodically prune metrics data points older than 1 day. -pub(crate) async fn metrics_prune_worker(metrics: Arc) { +pub(crate) async fn metrics_prune_worker( + metrics: Arc, + token: CancellationToken, +) { use extenddb_core::metrics::QuerySource; - const PRUNE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(300); - loop { - tokio::time::sleep(PRUNE_INTERVAL).await; + const PRUNE_INTERVAL: Duration = Duration::from_secs(300); + while tick(&token, PRUNE_INTERVAL).await { let cycle_start = std::time::Instant::now(); metrics.prune(); #[allow(clippy::cast_precision_loss)] @@ -173,19 +174,31 @@ pub(crate) async fn metrics_prune_worker(metrics: Arc, store: Arc, + token: CancellationToken, ) { use extenddb_core::metrics::QuerySource; use extenddb_storage::management_store::MetricsRow; - const FLUSH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60); - const RETENTION: std::time::Duration = std::time::Duration::from_secs(86400); + const FLUSH_INTERVAL: Duration = Duration::from_secs(60); + const RETENTION: Duration = Duration::from_secs(86400); loop { - tokio::time::sleep(FLUSH_INTERVAL).await; + let running = tick(&token, FLUSH_INTERVAL).await; + // Final flush drains everything; steady-state flushes leave points + // younger than one interval to accumulate into a complete bucket. + let drain_age = if running { + FLUSH_INTERVAL + } else { + Duration::ZERO + }; let cycle_start = std::time::Instant::now(); - let buckets = metrics.drain(FLUSH_INTERVAL); + let buckets = metrics.drain(drain_age); if !buckets.is_empty() { let rows: Vec = buckets .iter() @@ -238,19 +251,184 @@ pub(crate) async fn metrics_flush_worker( let cycle_us = cycle_start.elapsed().as_micros() as f64; metrics.record_worker_success(QuerySource::MetricsFlush, cycle_us); } + if !running { + break; + } } } /// Background worker that deletes old login attempt records. -pub(crate) async fn login_attempt_cleanup_worker(store: Arc) { - use std::time::Duration; - +pub(crate) async fn login_attempt_cleanup_worker( + store: Arc, + token: CancellationToken, +) { const CLEANUP_INTERVAL: Duration = Duration::from_secs(3600); // Keep records for 24 hours for audit purposes. const MAX_AGE_SECONDS: i64 = 86400; - loop { - tokio::time::sleep(CLEANUP_INTERVAL).await; + while tick(&token, CLEANUP_INTERVAL).await { store.cleanup_old_attempts(MAX_AGE_SECONDS).await; } } + +#[cfg(test)] +mod tests { + use super::{login_attempt_cleanup_worker, metrics_flush_worker}; + use extenddb_core::metrics::MetricsCollector; + use extenddb_storage::CancellationToken; + use extenddb_storage::management_store::{MetricsRow, MetricsStore, OpResult, RateLimitStore}; + use futures::future::BoxFuture; + use std::sync::Arc; + use std::sync::Mutex; + use std::time::Duration; + + /// Records every row handed to `insert_metrics` so a test can assert what + /// the flush worker persisted. + #[derive(Default)] + struct RecordingMetricsStore { + inserted: Mutex>, + } + + impl RecordingMetricsStore { + fn inserted_count(&self) -> usize { + self.inserted.lock().expect("lock poisoned").len() + } + } + + impl MetricsStore for RecordingMetricsStore { + fn insert_metrics(&self, rows: &[MetricsRow]) -> BoxFuture<'_, OpResult<()>> { + self.inserted + .lock() + .expect("lock poisoned") + .extend_from_slice(rows); + Box::pin(async { Ok(()) }) + } + + fn query_metrics( + &self, + _start: time::OffsetDateTime, + _end: time::OffsetDateTime, + _table_name: Option<&str>, + _metric: Option<&str>, + ) -> BoxFuture<'_, OpResult>> { + Box::pin(async { Ok(Vec::new()) }) + } + + fn prune_metrics(&self, _retention: Duration) -> BoxFuture<'_, OpResult<()>> { + Box::pin(async { Ok(()) }) + } + } + + /// Counts cleanup sweeps so a test can assert the worker did not run one. + #[derive(Default)] + struct CountingRateLimitStore { + sweeps: Mutex, + } + + impl RateLimitStore for CountingRateLimitStore { + fn count_principal_failures( + &self, + _principal: &str, + _window_seconds: i64, + ) -> BoxFuture<'_, OpResult> { + Box::pin(async { Ok(0) }) + } + + fn count_ip_failures( + &self, + _source_ip: &str, + _window_seconds: i64, + ) -> BoxFuture<'_, OpResult> { + Box::pin(async { Ok(0) }) + } + + fn record_failed_login( + &self, + _principal: &str, + _source_ip: Option<&str>, + ) -> BoxFuture<'_, ()> { + Box::pin(async {}) + } + + fn cleanup_old_attempts(&self, _max_age_seconds: i64) -> BoxFuture<'_, ()> { + *self.sweeps.lock().expect("lock poisoned") += 1; + Box::pin(async {}) + } + } + + /// A worker must keep running while the token is live. Without this the + /// cancellation test below would also pass for a worker that returns + /// immediately and never does any work at all. + #[tokio::test] + async fn worker_keeps_running_until_cancelled() { + let store = Arc::new(CountingRateLimitStore::default()); + let token = CancellationToken::new(); + let handle = tokio::spawn(login_attempt_cleanup_worker(store.clone(), token.clone())); + + // The cleanup interval is an hour, so the worker must still be sleeping. + let outcome = tokio::time::timeout(Duration::from_millis(200), handle).await; + assert!( + outcome.is_err(), + "worker returned before it was cancelled — its loop is not waiting on the interval" + ); + assert_eq!( + *store.sweeps.lock().expect("lock poisoned"), + 0, + "worker swept before its first interval elapsed" + ); + token.cancel(); + } + + /// Cancelling the token must stop the worker at its next tick rather than + /// leaving it to be dropped when the runtime shuts down. + #[tokio::test] + async fn cancellation_stops_a_sleeping_worker() { + let store = Arc::new(CountingRateLimitStore::default()); + let token = CancellationToken::new(); + let handle = tokio::spawn(login_attempt_cleanup_worker(store, token.clone())); + + token.cancel(); + + // The worker is mid-`sleep` on an hour-long interval; it may only return + // promptly because it also selects on the token. + tokio::time::timeout(Duration::from_secs(2), handle) + .await + .expect("worker did not return within 2s of cancellation") + .expect("worker panicked"); + } + + /// On shutdown the flush worker must persist the partial bucket it is + /// holding. The flush interval is 60s, so a row appearing at all proves it + /// came from the cancellation path and not from a periodic tick. + #[tokio::test] + async fn cancellation_flushes_the_final_partial_bucket() { + let metrics = Arc::new(MetricsCollector::new()); + let store = Arc::new(RecordingMetricsStore::default()); + let token = CancellationToken::new(); + + // Buffer a data point younger than one flush interval — a steady-state + // flush would deliberately leave this to accumulate. + metrics.record_latency(Some("test-table"), "PutItem", 1_234.0); + + let handle = tokio::spawn(metrics_flush_worker(metrics, store.clone(), token.clone())); + + // Nothing may be written before shutdown begins. + tokio::time::sleep(Duration::from_millis(100)).await; + assert_eq!( + store.inserted_count(), + 0, + "worker flushed before its first interval elapsed" + ); + + token.cancel(); + tokio::time::timeout(Duration::from_secs(2), handle) + .await + .expect("flush worker did not return within 2s of cancellation") + .expect("flush worker panicked"); + + assert!( + store.inserted_count() > 0, + "final flush lost the in-flight bucket: nothing was persisted on shutdown" + ); + } +} diff --git a/crates/storage-postgres/Cargo.toml b/crates/storage-postgres/Cargo.toml index e8e54e36..9231a0e3 100755 --- a/crates/storage-postgres/Cargo.toml +++ b/crates/storage-postgres/Cargo.toml @@ -13,7 +13,6 @@ extenddb-core = { workspace = true } extenddb-storage = { workspace = true } extenddb-auth = { workspace = true } futures = { workspace = true } -inventory = { workspace = true } sqlx = { workspace = true } tokio = { workspace = true, features = ["sync"] } serde = { workspace = true } diff --git a/crates/storage-postgres/src/lib.rs b/crates/storage-postgres/src/lib.rs index e5e8c78c..a8c77579 100755 --- a/crates/storage-postgres/src/lib.rs +++ b/crates/storage-postgres/src/lib.rs @@ -38,68 +38,61 @@ pub use config::PostgresStorageConfig; pub use config::parse_connection_string; pub use credential_store::DbCredentialStore; -// Auto-register the Postgres backend at compile time -inventory::submit! { - extenddb_storage::bootstrapper::BackendRegistration { +/// The `PostgreSQL` storage backend. +/// +/// A thin `main` installs it before dispatching any subcommand: +/// +/// ```ignore +/// extenddb_storage::set_backend(extenddb_storage_postgres::backend())?; +/// ``` +pub fn backend() -> extenddb_storage::Backend { + extenddb_storage::Backend { name: "postgres", - factory: |config_path, cli_args| { + bootstrapper: |config_path, cli_args| { Box::pin(async move { let store = PostgresBootstrapper::from_config(&config_path, &cli_args).await?; Ok(Box::new(store) as Box) }) - } - } -} - -// Auto-register PostgreSQL operations engine -inventory::submit! { - extenddb_storage::operations::OperationsEngineRegistration { - name: "postgres", + }, operations: &operations::PostgresOperationsEngine, - } -} - -// Auto-register PostgreSQL config deserializer -inventory::submit! { - extenddb_storage::config::StorageConfigRegistration { - backend: "postgres", - deserializer: |table| { - let config: PostgresStorageConfig = table.clone().try_into() + storage_config: |table| { + let config: PostgresStorageConfig = table + .clone() + .try_into() .map_err(|e: toml::de::Error| format!("Failed to parse postgres config: {e}"))?; Ok(Box::new(config) as Box) }, - } -} - -// Auto-register PostgreSQL settings store factory -inventory::submit! { - extenddb_storage::settings_store::SettingsStoreRegistration { - backend: "postgres", - factory: |connection_string| { + settings_store: |connection_string| { let connection_string = connection_string.to_string(); Box::pin(async move { let pool = sqlx::PgPool::connect(&connection_string) .await - .map_err(|e| extenddb_storage::settings_store::SettingsStoreError::ConnectionFailed(e.to_string()))?; - Ok(Box::new(PostgresCatalogStore::new(pool)) as Box) + .map_err(|e| { + extenddb_storage::settings_store::SettingsStoreError::ConnectionFailed( + e.to_string(), + ) + })?; + Ok(Box::new(PostgresCatalogStore::new(pool)) + as Box< + dyn extenddb_storage::management_store::SettingsStore, + >) }) }, - } -} - -// Auto-register PostgreSQL diagnostics store factory -inventory::submit! { - extenddb_storage::diagnostics_store::DiagnosticsStoreRegistration { - backend: "postgres", - factory: |connection_string| { + diagnostics_store: |connection_string| { let connection_string = connection_string.to_string(); Box::pin(async move { let pool = sqlx::PgPool::connect(&connection_string) .await - .map_err(|e| extenddb_storage::diagnostics_store::DiagnosticsStoreError::ConnectionFailed(e.to_string()))?; - Ok(Box::new(PostgresCatalogStore::new(pool)) as Box) + .map_err(|e| { + extenddb_storage::diagnostics_store::DiagnosticsStoreError::ConnectionFailed( + e.to_string(), + ) + })?; + Ok(Box::new(PostgresCatalogStore::new(pool)) + as Box) }) }, + server_components: server_components_factory, } } @@ -336,9 +329,7 @@ impl PostgresEngine { use extenddb_auth::CredentialStore; use extenddb_storage::hooks::{ServerRuntimeHooks, WorkerContext}; -use extenddb_storage::server_components::{ - BackendError, ServerComponents, ServerComponentsRegistration, -}; +use extenddb_storage::server_components::{BackendError, ServerComponents}; /// Backend-specific runtime hooks for `PostgreSQL`. struct PostgresRuntimeHooks { @@ -350,55 +341,83 @@ struct PostgresRuntimeHooks { #[async_trait::async_trait] impl ServerRuntimeHooks for PostgresRuntimeHooks { - async fn spawn_workers(&self, ctx: &WorkerContext) { - // Backend-specific workers that need PostgreSQL internals + async fn spawn_workers(&self, ctx: &WorkerContext) -> Vec> { + // Backend-specific workers that need PostgreSQL internals. Each takes + // the shutdown token and returns at its next tick after cancellation; + // the handles are returned so `serve` can drain them. // 1. Control plane transitions poller let storage_for_poller = self.engine.clone(); let cp_notify = self.control_plane_notify.clone(); let catalog_store = ctx.catalog_store.clone(); - tokio::spawn(async move { - workers::poll_control_plane_transitions(storage_for_poller, cp_notify, catalog_store) - .await; + let token = ctx.shutdown.clone(); + let control_plane = tokio::spawn(async move { + workers::poll_control_plane_transitions( + storage_for_poller, + cp_notify, + catalog_store, + token, + ) + .await; }); // 2. Table size refresh worker let storage_for_size = self.engine.clone(); - tokio::spawn(async move { workers::table_size_refresh_worker(storage_for_size).await }); + let token = ctx.shutdown.clone(); + let table_size = tokio::spawn(async move { + workers::table_size_refresh_worker(storage_for_size, token).await + }); // 3. Stream record cleanup worker let storage_for_stream = self.engine.clone(); let metrics = ctx.metrics.clone(); - tokio::spawn(async move { - workers::stream_record_cleanup_worker(storage_for_stream, metrics).await; + let token = ctx.shutdown.clone(); + let stream_cleanup = tokio::spawn(async move { + workers::stream_record_cleanup_worker(storage_for_stream, metrics, token).await; }); // 4. Idempotency token cleanup worker let storage_for_token = self.engine.clone(); let metrics = ctx.metrics.clone(); - tokio::spawn(async move { - workers::idempotency_token_cleanup_worker(storage_for_token, metrics).await; + let token = ctx.shutdown.clone(); + let idempotency_cleanup = tokio::spawn(async move { + workers::idempotency_token_cleanup_worker(storage_for_token, metrics, token).await; }); // 5. TTL cleanup worker let storage_for_ttl = self.engine.clone(); let metrics = ctx.metrics.clone(); - tokio::spawn(async move { ttl_worker::ttl_cleanup_worker(storage_for_ttl, metrics).await }); + let token = ctx.shutdown.clone(); + let ttl = tokio::spawn(async move { + ttl_worker::ttl_cleanup_worker(storage_for_ttl, metrics, token).await; + }); // 6. Pool metrics worker - needs both catalog and data pools let catalog_pool = self.engine.pool.clone(); let data_pool = self.engine.data_pool().clone(); let metrics = ctx.metrics.clone(); - tokio::spawn(async move { - workers::pool_metrics_worker(catalog_pool, data_pool, metrics).await; + let token = ctx.shutdown.clone(); + let pool_metrics = tokio::spawn(async move { + workers::pool_metrics_worker(catalog_pool, data_pool, metrics, token).await; }); // 7. GSI delay poller let catalog_store_for_gsi = ctx.catalog_store.clone(); let gsi_delay = self.gsi_default_delay_ms.clone(); - tokio::spawn( - async move { workers::poll_gsi_delay(catalog_store_for_gsi, gsi_delay).await }, - ); + let token = ctx.shutdown.clone(); + let gsi_poller = tokio::spawn(async move { + workers::poll_gsi_delay(catalog_store_for_gsi, gsi_delay, token).await; + }); + + vec![ + control_plane, + table_size, + stream_cleanup, + idempotency_cleanup, + ttl, + pool_metrics, + gsi_poller, + ] } fn backend_info(&self) -> Option { @@ -406,125 +425,129 @@ impl ServerRuntimeHooks for PostgresRuntimeHooks { } } -// Register the PostgreSQL backend factory -inventory::submit! { - ServerComponentsRegistration { - backend: "postgres", - factory: |config, region| { - let connection_string = config.connection_config().to_string(); - let max_connections = config.max_connections(); - let max_catalog_connections = config.max_catalog_connections(); - let region = region.to_string(); - Box::pin(async move { - // Build PostgresConfig from extracted values - let pg_config = PostgresConfig { - connection_string: connection_string.clone(), - pool_size: max_connections, - max_item_size_bytes: 400_000, - }; - - // Create PostgresEngine - let engine = PostgresEngine::new(&pg_config, ®ion) - .await - .map_err(|e| BackendError::ConnectionFailed { - backend: "postgres".to_string(), - details: e.to_string(), - })?; - - // Check catalog version - engine.check_catalog_version().await.map_err(|e| match e { - StorageError::CatalogVersionMismatch { expected, found } => { - BackendError::CatalogVersionMismatch { expected, found } - } - _ => BackendError::InitializationFailed(e.to_string()), - })?; +/// Build server components for the Postgres backend (registered in [`register`]). +fn server_components_factory( + config: &dyn extenddb_storage::config::StorageConfig, + region: &str, +) -> std::pin::Pin< + Box> + Send>, +> { + let connection_string = config.connection_config().to_string(); + let max_connections = config.max_connections(); + let max_catalog_connections = config.max_catalog_connections(); + let region = region.to_string(); + Box::pin(async move { + // Build PostgresConfig from extracted values + let pg_config = PostgresConfig { + connection_string: connection_string.clone(), + pool_size: max_connections, + max_item_size_bytes: 400_000, + }; - // Recover control plane transitions (ignore errors) - match engine.process_control_plane_transitions().await { - Ok(ref t) if t.is_empty() => {} - Ok(transitions) => { - for (name, transition) in &transitions { - tracing::info!("Recovered table '{name}': {transition}"); - } - } - Err(e) => tracing::error!("Failed to recover control plane transitions: {e}"), + // Create PostgresEngine + let engine = PostgresEngine::new(&pg_config, ®ion) + .await + .map_err(|e| BackendError::ConnectionFailed { + backend: "postgres".to_string(), + details: e.to_string(), + })?; + + // Check catalog version + engine.check_catalog_version().await.map_err(|e| match e { + StorageError::CatalogVersionMismatch { expected, found } => { + BackendError::CatalogVersionMismatch { expected, found } + } + _ => BackendError::InitializationFailed(e.to_string()), + })?; + + // Recover control plane transitions (ignore errors) + match engine.process_control_plane_transitions().await { + Ok(ref t) if t.is_empty() => {} + Ok(transitions) => { + for (name, transition) in &transitions { + tracing::info!("Recovered table '{name}': {transition}"); } + } + Err(e) => tracing::error!("Failed to recover control plane transitions: {e}"), + } - // Start GSI workers - let engine = engine.start_gsi_workers(); + // Start GSI workers + let engine = engine.start_gsi_workers(); - // Get data database name for logging (before wrapping in Arc) - let data_db_name = engine - .get_data_database_info() - .await - .unwrap_or_else(|_| "(query failed)".to_owned()); - - // Get references to fields we need before wrapping - let control_plane_notify = engine.control_plane_notify.clone(); - let gsi_default_delay_ms = engine.gsi_default_delay_ms.clone(); - - // Wrap engine in Arc - let engine = Arc::new(engine); - - // Create catalog store. Honors storage.postgres.catalog_pool_size, - // defaulting to pool_size when unset. Clamped to the same minimum - // as the engine pool. - let catalog_pool_size = if max_catalog_connections < MIN_POOL_SIZE { - tracing::warn!( - "storage.postgres.catalog_pool_size = {} is below the minimum of {}; clamping to {}", - max_catalog_connections, - MIN_POOL_SIZE, - MIN_POOL_SIZE - ); - MIN_POOL_SIZE - } else { - max_catalog_connections - }; - let catalog_pool = PgPoolOptions::new() - .max_connections(catalog_pool_size) - .min_connections(catalog_pool_size.min(2)) - .test_before_acquire(false) - .max_lifetime(std::time::Duration::from_secs(1800)) - .connect(&connection_string) - .await - .map_err(|e| BackendError::ConnectionFailed { - backend: "postgres".to_string(), - details: format!("Failed to create catalog pool: {e}"), - })?; + // Get data database name for logging (before wrapping in Arc) + let data_db_name = engine + .get_data_database_info() + .await + .unwrap_or_else(|_| "(query failed)".to_owned()); - // Load encryption key - let enc_key: Option = - sqlx::query_scalar("SELECT value FROM settings WHERE key = 'encryption_key'") - .fetch_optional(&catalog_pool) - .await - .map_err(|e| BackendError::InitializationFailed(format!("Failed to fetch encryption key: {e}")))?; - - let catalog_store = Arc::new(match enc_key { - Some(k) => PostgresCatalogStore::with_encryption_key(catalog_pool.clone(), k), - None => return Err(BackendError::MissingEncryptionKey), - }) as Arc; - - // Create auth provider - let enc_key = extenddb_storage::CatalogStore::cached_encryption_key(&*catalog_store) - .ok_or(BackendError::MissingEncryptionKey)?; - let cred_store: Arc = - Arc::new(DbCredentialStore::new(catalog_pool.clone(), enc_key)); - - // Create runtime hooks - let runtime_hooks = Box::new(PostgresRuntimeHooks { - engine: engine.clone(), - control_plane_notify, - gsi_default_delay_ms, - data_db_name, - }); - - Ok(ServerComponents { - engine, - catalog_store, - credential_store: cred_store, - runtime_hooks: Some(runtime_hooks), - }) - }) - }, - } + // Get references to fields we need before wrapping + let control_plane_notify = engine.control_plane_notify.clone(); + let gsi_default_delay_ms = engine.gsi_default_delay_ms.clone(); + + // Wrap engine in Arc + let engine = Arc::new(engine); + + // Create catalog store. Honors storage.postgres.catalog_pool_size, + // defaulting to pool_size when unset. Clamped to the same minimum + // as the engine pool. + let catalog_pool_size = if max_catalog_connections < MIN_POOL_SIZE { + tracing::warn!( + "storage.postgres.catalog_pool_size = {} is below the minimum of {}; clamping to {}", + max_catalog_connections, + MIN_POOL_SIZE, + MIN_POOL_SIZE + ); + MIN_POOL_SIZE + } else { + max_catalog_connections + }; + let catalog_pool = PgPoolOptions::new() + .max_connections(catalog_pool_size) + .min_connections(catalog_pool_size.min(2)) + .test_before_acquire(false) + .max_lifetime(std::time::Duration::from_secs(1800)) + .connect(&connection_string) + .await + .map_err(|e| BackendError::ConnectionFailed { + backend: "postgres".to_string(), + details: format!("Failed to create catalog pool: {e}"), + })?; + + // Load encryption key + let enc_key: Option = + sqlx::query_scalar("SELECT value FROM settings WHERE key = 'encryption_key'") + .fetch_optional(&catalog_pool) + .await + .map_err(|e| { + BackendError::InitializationFailed(format!( + "Failed to fetch encryption key: {e}" + )) + })?; + + let catalog_store = Arc::new(match enc_key { + Some(k) => PostgresCatalogStore::with_encryption_key(catalog_pool.clone(), k), + None => return Err(BackendError::MissingEncryptionKey), + }) as Arc; + + // Create auth provider + let enc_key = extenddb_storage::CatalogStore::cached_encryption_key(&*catalog_store) + .ok_or(BackendError::MissingEncryptionKey)?; + let cred_store: Arc = + Arc::new(DbCredentialStore::new(catalog_pool.clone(), enc_key)); + + // Create runtime hooks + let runtime_hooks = Box::new(PostgresRuntimeHooks { + engine: engine.clone(), + control_plane_notify, + gsi_default_delay_ms, + data_db_name, + }); + + Ok(ServerComponents { + engine, + catalog_store, + credential_store: cred_store, + runtime_hooks: Some(runtime_hooks), + }) + }) } diff --git a/crates/storage-postgres/src/ttl_worker.rs b/crates/storage-postgres/src/ttl_worker.rs index 8c87a9bc..c40a59b5 100644 --- a/crates/storage-postgres/src/ttl_worker.rs +++ b/crates/storage-postgres/src/ttl_worker.rs @@ -9,7 +9,9 @@ use std::time::Duration; use extenddb_core::metrics::MetricsCollector; use extenddb_core::types::UserIdentity; use extenddb_storage::error::StorageError; -use extenddb_storage::{DataEngine, MetadataEngine, TableEngine}; +use extenddb_storage::{ + CancellationToken, DataEngine, MetadataEngine, TableEngine, sleep_or_shutdown, +}; use crate::PostgresEngine; @@ -20,11 +22,11 @@ const BATCH_SIZE: usize = 100; pub(crate) async fn ttl_cleanup_worker( storage: Arc, metrics: Arc, + token: CancellationToken, ) { let region_arc: Arc = Arc::from(storage.region.as_str()); - loop { - tokio::time::sleep(SCAN_INTERVAL).await; + while sleep_or_shutdown(&token, SCAN_INTERVAL).await { retry_pending_indexes(&storage).await; sweep_expired_items(&storage, &metrics, ®ion_arc).await; } diff --git a/crates/storage-postgres/src/workers.rs b/crates/storage-postgres/src/workers.rs index 3f7a7d8a..e7388f88 100644 --- a/crates/storage-postgres/src/workers.rs +++ b/crates/storage-postgres/src/workers.rs @@ -9,7 +9,8 @@ use std::time::Duration; use extenddb_core::metrics::MetricsCollector; use extenddb_storage::management_store::SettingsStore; -use extenddb_storage::{DataEngine, MetadataEngine, StreamEngine}; +use extenddb_storage::sleep_or_shutdown as tick; +use extenddb_storage::{CancellationToken, DataEngine, MetadataEngine, StreamEngine}; use sqlx::PgPool; use crate::PostgresEngine; @@ -18,14 +19,22 @@ pub(crate) async fn poll_control_plane_transitions( storage: Arc, notify: Arc, settings: Arc, + token: CancellationToken, ) { const ACTIVE_POLL: Duration = Duration::from_secs(1); const IDLE_TIMEOUT: Duration = Duration::from_secs(60); const MARGIN_SECS: f64 = 5.0; loop { - // Idle: wait for a wake signal or timeout (defensive sweep) - let _ = tokio::time::timeout(IDLE_TIMEOUT, notify.notified()).await; + // Idle: wait for a wake signal, an idle timeout (defensive sweep), or + // shutdown. + let shutting_down = tokio::select! { + () = token.cancelled() => true, + _ = tokio::time::timeout(IDLE_TIMEOUT, notify.notified()) => false, + }; + if shutting_down { + return; + } // Read control_plane_delay_seconds from settings to compute active window let delay_secs = read_control_plane_delay(&*settings).await; @@ -49,7 +58,9 @@ pub(crate) async fn poll_control_plane_transitions( if tokio::time::Instant::now() >= deadline { break; } - tokio::time::sleep(ACTIVE_POLL).await; + if !tick(&token, ACTIVE_POLL).await { + return; + } } } } @@ -65,12 +76,13 @@ async fn read_control_plane_delay(store: &S) -> f64 { .unwrap_or(0.25) } -pub(crate) async fn table_size_refresh_worker(storage: Arc) { +pub(crate) async fn table_size_refresh_worker( + storage: Arc, + token: CancellationToken, +) { const REFRESH_INTERVAL: Duration = Duration::from_secs(300); - loop { - tokio::time::sleep(REFRESH_INTERVAL).await; - + while tick(&token, REFRESH_INTERVAL).await { let tables = match MetadataEngine::all_active_tables(&*storage).await { Ok(t) => t, Err(e) => { @@ -92,14 +104,14 @@ pub(crate) async fn table_size_refresh_worker(storage: Arc) { pub(crate) async fn stream_record_cleanup_worker( storage: Arc, metrics: Arc, + token: CancellationToken, ) { use extenddb_core::metrics::QuerySource; const CLEANUP_INTERVAL: Duration = Duration::from_secs(3600); const RETENTION_HOURS: i64 = 24; - loop { - tokio::time::sleep(CLEANUP_INTERVAL).await; + while tick(&token, CLEANUP_INTERVAL).await { let cycle_start = std::time::Instant::now(); match StreamEngine::cleanup_expired_stream_records(&*storage, RETENTION_HOURS).await { @@ -125,14 +137,14 @@ pub(crate) async fn stream_record_cleanup_worker( pub(crate) async fn idempotency_token_cleanup_worker( storage: Arc, metrics: Arc, + token: CancellationToken, ) { use extenddb_core::metrics::QuerySource; const CLEANUP_INTERVAL: Duration = Duration::from_secs(600); const MAX_AGE_SECONDS: i64 = 600; - loop { - tokio::time::sleep(CLEANUP_INTERVAL).await; + while tick(&token, CLEANUP_INTERVAL).await { let cycle_start = std::time::Instant::now(); match DataEngine::cleanup_expired_idempotency_tokens(&*storage, MAX_AGE_SECONDS).await { @@ -158,12 +170,11 @@ pub(crate) async fn idempotency_token_cleanup_worker( pub(crate) async fn poll_gsi_delay( store: Arc, gsi_delay: Arc, + token: CancellationToken, ) { const POLL_INTERVAL: Duration = Duration::from_secs(30); - loop { - tokio::time::sleep(POLL_INTERVAL).await; - + while tick(&token, POLL_INTERVAL).await { match store.get_setting("gsi_propagation_delay_ms").await { Ok(Some(val)) => { if let Ok(ms) = val.parse::() { @@ -185,12 +196,11 @@ pub(crate) async fn pool_metrics_worker( catalog_pool: PgPool, data_pool: PgPool, metrics: Arc, + token: CancellationToken, ) { const SAMPLE_INTERVAL: Duration = Duration::from_secs(5); - loop { - tokio::time::sleep(SAMPLE_INTERVAL).await; - + while tick(&token, SAMPLE_INTERVAL).await { let catalog_size = catalog_pool.size() as usize; let catalog_idle = catalog_pool.num_idle(); let data_size = data_pool.size() as usize; diff --git a/crates/storage/Cargo.toml b/crates/storage/Cargo.toml index 4e7cabee..44224b03 100755 --- a/crates/storage/Cargo.toml +++ b/crates/storage/Cargo.toml @@ -16,12 +16,12 @@ bigdecimal = { workspace = true } extenddb-auth = { workspace = true } extenddb-core = { workspace = true } futures = { workspace = true } -inventory = { workspace = true } rand = { workspace = true } serde_json = { workspace = true } thiserror = { workspace = true } time = { workspace = true } tokio = { workspace = true } +tokio-util = { workspace = true } toml = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } diff --git a/crates/storage/src/backend.rs b/crates/storage/src/backend.rs new file mode 100644 index 00000000..8eca6603 --- /dev/null +++ b/crates/storage/src/backend.rs @@ -0,0 +1,122 @@ +// Copyright 2026 ExtendDB contributors +// SPDX-License-Identifier: Apache-2.0 + +//! The compiled-in storage backend. +//! +//! A binary is built for exactly one backend. The thin `main` installs it once +//! with [`set_backend`] before dispatching any subcommand: +//! +//! ```ignore +//! fn main() -> anyhow::Result<()> { +//! extenddb_storage::set_backend(extenddb_storage_postgres::backend())?; +//! extenddb_app::run(extenddb_app::BuildInfo { .. }) +//! } +//! ``` +//! +//! This replaces both the previous `inventory`-based auto-registration and the +//! name-keyed registry that succeeded it. Auto-registration relied on the linker +//! preserving `submit!` statics, which only happened if the binary referenced the +//! backend crate — an invisible, compiles-fine failure mode. A name-keyed +//! registry fixed that but kept string dispatch, which allows a class of runtime +//! error that cannot exist here: there is nothing to look up, so a mistyped or +//! absent backend name can no longer produce an "unknown backend" failure after +//! startup. +//! +//! The backend carries its own [`Backend::name`], so configuration can locate +//! its `[storage.]` section without taking that name from the config file. + +use std::sync::OnceLock; + +use crate::bootstrapper::BootstrapperFactory; +use crate::config::StorageConfigDeserializer; +use crate::diagnostics_store::DiagnosticsStoreFactory; +use crate::operations::OperationsEngine; +use crate::server_components::ServerComponentsFactory; +use crate::settings_store::SettingsStoreFactory; + +/// The complete set of factories a storage backend provides. +/// +/// A backend crate exposes one constructor returning this value (by convention +/// `backend()`), which the thin bin hands to [`set_backend`]. +pub struct Backend { + /// Backend name, used for the `[storage.]` config section, the startup + /// banner, and diagnostics. This is the authoritative name: the config + /// file's `backend` key is validated against it rather than driving dispatch. + pub name: &'static str, + /// Creates the deployment bootstrapper (`init`, `destroy`, `migrate`). + pub bootstrapper: BootstrapperFactory, + /// Deserializes the backend's `[storage.]` config section. + pub storage_config: StorageConfigDeserializer, + /// Backend operations engine (catalog version, connection redaction). + pub operations: &'static dyn OperationsEngine, + /// Creates the runtime settings store. + pub settings_store: SettingsStoreFactory, + /// Creates the diagnostics store (`catalog-check`, `verify`). + pub diagnostics_store: DiagnosticsStoreFactory, + /// Creates the assembled server components for `serve`. + pub server_components: ServerComponentsFactory, +} + +static BACKEND: OnceLock = OnceLock::new(); + +/// Error returned by [`set_backend`] when a backend was already installed. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub struct BackendAlreadySet; + +impl std::fmt::Display for BackendAlreadySet { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "storage backend already installed") + } +} + +impl std::error::Error for BackendAlreadySet {} + +/// Install the process-wide storage backend. +/// +/// Call exactly once, from `main`, before dispatching any subcommand. +/// +/// # Errors +/// +/// Returns [`BackendAlreadySet`] if a backend was already installed; the first +/// one wins and the argument is dropped. +pub fn set_backend(backend: Backend) -> Result<(), BackendAlreadySet> { + BACKEND.set(backend).map_err(|_| BackendAlreadySet) +} + +/// Borrow the installed backend, if one has been installed. +/// +/// Returns `None` before [`set_backend`] runs. Callers treat that the same way +/// they treated a missing registry: a clear runtime error rather than a panic. +#[must_use] +pub fn try_backend() -> Option<&'static Backend> { + BACKEND.get() +} + +/// Name of the installed backend, or `None` before one is installed. +#[must_use] +pub fn backend_name() -> Option<&'static str> { + try_backend().map(|b| b.name) +} + +#[cfg(test)] +mod tests { + use super::{BackendAlreadySet, backend_name, try_backend}; + + /// Before `set_backend` runs there is no backend, and the lookup helpers say + /// so rather than panicking. (`set_backend` installs into a process-wide + /// `OnceLock`, so a positive test would leak into every other test in this + /// binary and is covered by the integration suite instead.) + #[test] + fn no_backend_is_installed_by_default() { + assert!(try_backend().is_none()); + assert_eq!(backend_name(), None); + } + + #[test] + fn already_set_error_renders() { + assert_eq!( + BackendAlreadySet.to_string(), + "storage backend already installed" + ); + } +} diff --git a/crates/storage/src/bootstrapper.rs b/crates/storage/src/bootstrapper.rs index 9fefd361..88299ebe 100755 --- a/crates/storage/src/bootstrapper.rs +++ b/crates/storage/src/bootstrapper.rs @@ -145,54 +145,19 @@ pub type BootstrapperFactory = Vec, ) -> Pin, StorageError>> + Send>>; -/// Backend bootstrapper registration entry. +/// Create a bootstrapper using the installed backend. /// -/// Backend crates submit instances of this struct using `inventory::submit!` -/// to register their bootstrappers at compile time. -pub struct BackendRegistration { - pub name: &'static str, - pub factory: BootstrapperFactory, -} - -inventory::collect!(BackendRegistration); - -/// Create a bootstrapper for the given backend. -/// -/// Looks up the backend in the compile-time registry and calls its bootstrapper factory. +/// Calls the bootstrapper factory of the [`Backend`](crate::Backend) installed +/// via [`set_backend`](crate::set_backend). pub async fn create_bootstrapper( - backend: &str, config_path: &str, cli_args: &[String], ) -> Result, StorageError> { - for registration in inventory::iter:: { - if registration.name == backend { - tracing::info!("Found registered backend: {}", backend); - return (registration.factory)(config_path.to_string(), cli_args.to_vec()).await; - } - } - - let available: Vec<&str> = inventory::iter::() - .map(|r| r.name) - .collect(); - - tracing::error!( - "Unknown backend: {}. Available: {}", - backend, - available.join(", ") - ); - - Err(StorageError::Internal(format!( - "Unknown backend: {backend}. Available backends: {}", - available.join(", ") - ))) -} - -/// List all registered backends. -#[must_use] -pub fn list_backends() -> Vec<&'static str> { - inventory::iter::() - .map(|r| r.name) - .collect() + let backend = crate::backend::try_backend().ok_or_else(|| { + StorageError::Internal("no storage backend installed (set_backend was not called)".into()) + })?; + tracing::info!("Using compiled-in backend: {}", backend.name); + (backend.bootstrapper)(config_path.to_string(), cli_args.to_vec()).await } /// Helper functions for bootstrapper implementations. diff --git a/crates/storage/src/config.rs b/crates/storage/src/config.rs index a6b260ab..f1286dfa 100644 --- a/crates/storage/src/config.rs +++ b/crates/storage/src/config.rs @@ -42,26 +42,13 @@ impl Clone for Box { /// Takes a TOML table and returns a boxed `StorageConfig` trait object. pub type StorageConfigDeserializer = fn(&toml::Table) -> Result, String>; -/// Registration entry for a storage config deserializer. -pub struct StorageConfigRegistration { - pub backend: &'static str, - pub deserializer: StorageConfigDeserializer, -} - -inventory::collect!(StorageConfigRegistration); - /// Deserialize a storage configuration from a TOML table. /// -/// Looks up the registered deserializer for the given backend name -/// and invokes it with the provided TOML table. -pub fn deserialize_storage_config( - backend: &str, - table: &toml::Table, -) -> Result, String> { - for reg in inventory::iter:: { - if reg.backend == backend { - return (reg.deserializer)(table); - } - } - Err(format!("Unknown backend: {backend}")) +/// Uses the deserializer of the [`Backend`](crate::Backend) installed via +/// [`set_backend`](crate::set_backend), invoking it with the provided TOML +/// table. +pub fn deserialize_storage_config(table: &toml::Table) -> Result, String> { + let backend = crate::backend::try_backend() + .ok_or_else(|| "no storage backend installed (set_backend was not called)".to_owned())?; + (backend.storage_config)(table) } diff --git a/crates/storage/src/diagnostics_store.rs b/crates/storage/src/diagnostics_store.rs index 23dbb211..bea75a05 100644 --- a/crates/storage/src/diagnostics_store.rs +++ b/crates/storage/src/diagnostics_store.rs @@ -9,17 +9,18 @@ use futures::future::BoxFuture; /// Error type for diagnostics store creation. #[derive(Debug)] pub enum DiagnosticsStoreError { - BackendNotFound(String), + /// No storage backend has been installed (set_backend was not called). + BackendNotInstalled, ConnectionFailed(String), } impl std::fmt::Display for DiagnosticsStoreError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::BackendNotFound(backend) => { + Self::BackendNotInstalled => { write!( f, - "No diagnostics store factory registered for backend '{backend}'" + "no storage backend installed (set_backend was not called)" ) } Self::ConnectionFailed(msg) => write!(f, "Failed to connect: {msg}"), @@ -33,23 +34,11 @@ impl std::error::Error for DiagnosticsStoreError {} pub type DiagnosticsStoreFactory = fn(&str) -> BoxFuture<'static, Result, DiagnosticsStoreError>>; -/// Registration entry for a diagnostics store factory. -pub struct DiagnosticsStoreRegistration { - pub backend: &'static str, - pub factory: DiagnosticsStoreFactory, -} - -inventory::collect!(DiagnosticsStoreRegistration); - -/// Create a diagnostics store for the given backend and connection string. +/// Create a diagnostics store for the installed backend. pub async fn create_diagnostics_store( - backend: &str, connection_string: &str, ) -> Result, DiagnosticsStoreError> { - for registration in inventory::iter:: { - if registration.backend == backend { - return (registration.factory)(connection_string).await; - } - } - Err(DiagnosticsStoreError::BackendNotFound(backend.to_string())) + let backend = + crate::backend::try_backend().ok_or(DiagnosticsStoreError::BackendNotInstalled)?; + (backend.diagnostics_store)(connection_string).await } diff --git a/crates/storage/src/hooks.rs b/crates/storage/src/hooks.rs index ee166e36..82259e7e 100644 --- a/crates/storage/src/hooks.rs +++ b/crates/storage/src/hooks.rs @@ -6,6 +6,7 @@ use std::sync::Arc; use async_trait::async_trait; +pub use tokio_util::sync::CancellationToken; use tracing_subscriber::{EnvFilter, Registry, reload}; /// Context passed to `ServerRuntimeHooks::spawn_workers`. @@ -16,6 +17,25 @@ pub struct WorkerContext { pub catalog_store: Arc, pub reload_handle: reload::Handle, pub config_log_level: String, + /// Cancelled when the server begins shutting down. Backend workers should + /// select on [`CancellationToken::cancelled`] alongside their own timer and + /// return once it fires, so shutdown drains them instead of dropping them + /// mid-cycle. [`sleep_or_shutdown`] does this for the common + /// `loop { sleep(interval); work(); }` shape. Re-exported here so a backend + /// crate does not need its own `tokio-util` dependency. + pub shutdown: CancellationToken, +} + +/// Sleep for `interval`, returning early if the server is shutting down. +/// +/// Returns `true` when the interval elapsed (run another cycle) and `false` +/// when `token` was cancelled (stop looping), which makes the canonical worker +/// loop `while sleep_or_shutdown(&token, INTERVAL).await { work().await; }`. +pub async fn sleep_or_shutdown(token: &CancellationToken, interval: std::time::Duration) -> bool { + tokio::select! { + () = token.cancelled() => false, + () = tokio::time::sleep(interval) => true, + } } /// Backend-specific runtime hooks for worker spawning and initialization. @@ -30,7 +50,12 @@ pub trait ServerRuntimeHooks: Send + Sync { /// Called after server components are created but before the HTTP server /// starts. Backends can spawn workers that need access to backend-specific /// state (connection pools, notify handles, etc.). - async fn spawn_workers(&self, ctx: &WorkerContext); + /// + /// Return the spawned tasks' join handles so the server can await them + /// during shutdown after cancelling [`WorkerContext::shutdown`]. Returning + /// an empty vector opts out of the drain (the tasks are then dropped when + /// the runtime shuts down). + async fn spawn_workers(&self, ctx: &WorkerContext) -> Vec>; /// Get backend-specific info for logging (optional). /// diff --git a/crates/storage/src/lib.rs b/crates/storage/src/lib.rs index bdc43b25..205adf09 100755 --- a/crates/storage/src/lib.rs +++ b/crates/storage/src/lib.rs @@ -8,6 +8,7 @@ //! methods receive `account_id` from the authenticated identity. pub mod authorization_store; +pub mod backend; pub mod bootstrapper; pub mod config; pub mod diagnostics; @@ -20,14 +21,15 @@ pub mod server_components; pub mod settings_store; pub mod transact; +pub use backend::{Backend, BackendAlreadySet, backend_name, set_backend, try_backend}; + pub use transact::{IdempotencyKey, TransactGetOp, TransactWriteOp}; pub use server_components::{ - BackendError, ServerComponents, ServerComponentsFactory, ServerComponentsRegistration, - create_server_components, + BackendError, ServerComponents, ServerComponentsFactory, create_server_components, }; -pub use hooks::{ServerRuntimeHooks, WorkerContext}; +pub use hooks::{CancellationToken, ServerRuntimeHooks, WorkerContext, sleep_or_shutdown}; /// Pluggable lookup for `TableKeyInfo`. /// diff --git a/crates/storage/src/operations.rs b/crates/storage/src/operations.rs index d9a632b8..be9576b3 100644 --- a/crates/storage/src/operations.rs +++ b/crates/storage/src/operations.rs @@ -45,63 +45,44 @@ pub struct ConnectionParts { pub database: String, } -/// Registration entry for backend operations. -pub struct OperationsEngineRegistration { - pub name: &'static str, - pub operations: &'static dyn OperationsEngine, -} - -inventory::collect!(OperationsEngineRegistration); - -/// Get the operations engine for a backend by name. -pub fn get_operations_engine(backend: &str) -> Result<&'static dyn OperationsEngine, StorageError> { - for reg in inventory::iter:: { - if reg.name == backend { - return Ok(reg.operations); - } - } - - let available: Vec<&str> = inventory::iter::() - .map(|r| r.name) - .collect(); - - Err(StorageError::Internal(format!( - "Unknown backend: {backend}. Available backends: {}", - available.join(", ") - ))) -} - -/// List all registered backend names. -#[must_use] -pub fn list_operations_backends() -> Vec<&'static str> { - inventory::iter::() - .map(|r| r.name) - .collect() +/// Get the operations engine of the installed backend. +/// +/// # Errors +/// +/// Returns an error if no backend has been installed. +pub fn get_operations_engine() -> Result<&'static dyn OperationsEngine, StorageError> { + crate::backend::try_backend() + .map(|b| b.operations) + .ok_or_else(|| { + StorageError::Internal( + "no storage backend installed (set_backend was not called)".into(), + ) + }) } // Convenience functions that delegate to the operations engine /// Get the catalog version for a backend. -pub fn catalog_version(backend: &str) -> Result { - get_operations_engine(backend).map(OperationsEngine::catalog_version) +pub fn catalog_version() -> Result { + get_operations_engine().map(OperationsEngine::catalog_version) } /// Redact sensitive information from a connection string. -pub fn redact_connection_string(backend: &str, s: &str) -> Result { - get_operations_engine(backend).map(|ops| ops.redact_connection_string(s)) +pub fn redact_connection_string(s: &str) -> Result { + get_operations_engine().map(|ops| ops.redact_connection_string(s)) } /// Parse a connection string into components. -pub fn parse_connection_string(backend: &str, s: &str) -> Result { - get_operations_engine(backend)?.parse_connection_string(s) +pub fn parse_connection_string(s: &str) -> Result { + get_operations_engine()?.parse_connection_string(s) } /// Validate an identifier for DDL safety. -pub fn validate_identifier(backend: &str, name: &str, label: &str) -> Result<(), StorageError> { - get_operations_engine(backend)?.validate_identifier(name, label) +pub fn validate_identifier(name: &str, label: &str) -> Result<(), StorageError> { + get_operations_engine()?.validate_identifier(name, label) } /// Check if a configuration key contains sensitive data. -pub fn is_sensitive_key(backend: &str, key: &str) -> Result { - get_operations_engine(backend).map(|ops| ops.is_sensitive_key(key)) +pub fn is_sensitive_key(key: &str) -> Result { + get_operations_engine().map(|ops| ops.is_sensitive_key(key)) } diff --git a/crates/storage/src/server_components.rs b/crates/storage/src/server_components.rs index 441f61e8..4035e95c 100644 --- a/crates/storage/src/server_components.rs +++ b/crates/storage/src/server_components.rs @@ -40,8 +40,8 @@ pub struct ServerComponents { /// Errors that can occur during backend initialization. #[derive(Debug)] pub enum BackendError { - /// Backend name not registered - UnknownBackend(String), + /// No storage backend has been installed (set_backend was not called). + BackendNotInstalled, /// Failed to connect to backend database ConnectionFailed { backend: String, details: String }, @@ -59,8 +59,11 @@ pub enum BackendError { impl std::fmt::Display for BackendError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::UnknownBackend(b) => { - write!(f, "Unknown backend '{b}'. Available backends: postgres") + Self::BackendNotInstalled => { + write!( + f, + "no storage backend installed (set_backend was not called)" + ) } Self::ConnectionFailed { backend, details } => { write!(f, "Failed to connect to {backend}: {details}") @@ -90,32 +93,15 @@ pub type ServerComponentsFactory = &str, ) -> Pin> + Send>>; -/// Registration for backend server components factory. +/// Create server components using the installed backend. /// -/// Backends submit this via `inventory::submit`! to register themselves. -pub struct ServerComponentsRegistration { - /// Backend name (e.g., "postgres") - pub backend: &'static str, - - /// Factory function that creates the backend components - pub factory: ServerComponentsFactory, -} - -inventory::collect!(ServerComponentsRegistration); - -/// Create server components for the specified backend. -/// -/// Searches registered backends via inventory and calls the matching factory. -/// Returns `UnknownBackend` error if the backend is not registered. +/// Calls the factory of the [`Backend`](crate::Backend) installed via +/// [`set_backend`](crate::set_backend). Returns `BackendNotInstalled` if no +/// backend has been installed. pub async fn create_server_components( - backend: &str, config: &dyn StorageConfig, region: &str, ) -> Result { - for reg in inventory::iter:: { - if reg.backend == backend { - return (reg.factory)(config, region).await; - } - } - Err(BackendError::UnknownBackend(backend.to_string())) + let backend = crate::backend::try_backend().ok_or(BackendError::BackendNotInstalled)?; + (backend.server_components)(config, region).await } diff --git a/crates/storage/src/settings_store.rs b/crates/storage/src/settings_store.rs index 8a6032c9..0f406a1d 100644 --- a/crates/storage/src/settings_store.rs +++ b/crates/storage/src/settings_store.rs @@ -9,17 +9,18 @@ use futures::future::BoxFuture; /// Error type for settings store creation. #[derive(Debug)] pub enum SettingsStoreError { - BackendNotFound(String), + /// No storage backend has been installed (set_backend was not called). + BackendNotInstalled, ConnectionFailed(String), } impl std::fmt::Display for SettingsStoreError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - Self::BackendNotFound(backend) => { + Self::BackendNotInstalled => { write!( f, - "No settings store factory registered for backend '{backend}'" + "no storage backend installed (set_backend was not called)" ) } Self::ConnectionFailed(msg) => write!(f, "Failed to connect: {msg}"), @@ -33,23 +34,10 @@ impl std::error::Error for SettingsStoreError {} pub type SettingsStoreFactory = fn(&str) -> BoxFuture<'static, Result, SettingsStoreError>>; -/// Registration entry for a settings store factory. -pub struct SettingsStoreRegistration { - pub backend: &'static str, - pub factory: SettingsStoreFactory, -} - -inventory::collect!(SettingsStoreRegistration); - -/// Create a settings store for the given backend and connection string. +/// Create a settings store for the installed backend. pub async fn create_settings_store( - backend: &str, connection_string: &str, ) -> Result, SettingsStoreError> { - for registration in inventory::iter:: { - if registration.backend == backend { - return (registration.factory)(connection_string).await; - } - } - Err(SettingsStoreError::BackendNotFound(backend.to_string())) + let backend = crate::backend::try_backend().ok_or(SettingsStoreError::BackendNotInstalled)?; + (backend.settings_store)(connection_string).await }