From c3754d12559474d5b268fe139b71049fc257020a Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Tue, 17 Feb 2026 18:29:29 -0500 Subject: [PATCH] refactor(server): reduce duplication in server, tls, pubsub, and main server.rs - extract `setup_tls_listener` to replace the 8-line TLS bind block that was copy-pasted verbatim into both `run` and `run_concurrent` pubsub.rs - replace `Vec` with byte-slice iteration in `glob_match_inner` and `match_char_class`. all metacharacters are ASCII so byte comparison is correct for Redis-compatible pub/sub patterns. eliminates two heap allocations per pattern match on the hot pub/sub path. - add doc comments explaining the backtracking algorithm tls.rs - extract `require_file(path, make_err)` to replace three near-identical "if !path.exists() { return Err(...) }" blocks main.rs - extract `exit_err(msg) -> !` to normalize the eprintln+exit(1) pattern used across all argument validation failures. call sites become one-liners. --- crates/ember-server/src/main.rs | 93 +++++++++++++------------------ crates/ember-server/src/pubsub.rs | 42 ++++++++------ crates/ember-server/src/server.rs | 40 ++++++------- crates/ember-server/src/tls.rs | 24 +++++--- 4 files changed, 99 insertions(+), 100 deletions(-) diff --git a/crates/ember-server/src/main.rs b/crates/ember-server/src/main.rs index 554feccd..f45ad13a 100644 --- a/crates/ember-server/src/main.rs +++ b/crates/ember-server/src/main.rs @@ -164,29 +164,39 @@ struct Args { cluster_node_timeout: u64, } +/// Prints `msg` to stderr and exits with code 1. +/// +/// Used throughout `main` to normalize the error-and-exit pattern for +/// argument validation failures. The `!` return type lets the compiler +/// verify that call sites don't need to produce a value after the call. +fn exit_err(msg: impl std::fmt::Display) -> ! { + eprintln!("{msg}"); + std::process::exit(1); +} + /// Resolves the password from either `--requirepass` or `--requirepass-file`. /// The two options are mutually exclusive. Exits on error. fn resolve_password(args: &mut Args) { if args.requirepass.is_some() && args.requirepass_file.is_some() { - eprintln!("error: --requirepass and --requirepass-file are mutually exclusive"); - std::process::exit(1); + exit_err("error: --requirepass and --requirepass-file are mutually exclusive"); } if let Some(ref path) = args.requirepass_file { match std::fs::read_to_string(path) { Ok(contents) => { let password = contents.trim_end().to_string(); if password.is_empty() { - eprintln!("error: --requirepass-file is empty: {}", path.display()); - std::process::exit(1); + exit_err(format!( + "error: --requirepass-file is empty: {}", + path.display() + )); } args.requirepass = Some(password); } Err(e) => { - eprintln!( + exit_err(format!( "error: failed to read --requirepass-file '{}': {e}", path.display() - ); - std::process::exit(1); + )); } } } @@ -198,11 +208,10 @@ fn parse_bind_addr(host: &str, port: u16, label: &str) -> SocketAddr { Ok(a) => a, Err(e) => { if label.is_empty() { - eprintln!("invalid bind address '{host}:{port}': {e}"); + exit_err(format!("invalid bind address '{host}:{port}': {e}")); } else { - eprintln!("invalid {label} bind address '{host}:{port}': {e}"); + exit_err(format!("invalid {label} bind address '{host}:{port}': {e}")); } - std::process::exit(1); } } } @@ -221,16 +230,13 @@ fn build_persistence_config( let data_dir = args.data_dir.take().unwrap_or_else(|| { if args.appendonly { - eprintln!("--data-dir is required when --appendonly is set"); - std::process::exit(1); + exit_err("--data-dir is required when --appendonly is set"); } PathBuf::from(".") }); - let fsync_policy = parse_fsync_policy(&args.appendfsync).unwrap_or_else(|e| { - eprintln!("invalid --appendfsync value: {e}"); - std::process::exit(1); - }); + let fsync_policy = parse_fsync_policy(&args.appendfsync) + .unwrap_or_else(|e| exit_err(format!("invalid --appendfsync value: {e}"))); Some(ShardPersistenceConfig { data_dir, @@ -257,16 +263,11 @@ async fn main() { let addr = parse_bind_addr(&args.host, args.port, ""); let max_memory = args.max_memory.as_deref().map(|s| { - parse_byte_size(s).unwrap_or_else(|e| { - eprintln!("invalid --max-memory value: {e}"); - std::process::exit(1); - }) + parse_byte_size(s).unwrap_or_else(|e| exit_err(format!("invalid --max-memory value: {e}"))) }); - let eviction_policy = parse_eviction_policy(&args.eviction_policy).unwrap_or_else(|e| { - eprintln!("invalid --eviction-policy value: {e}"); - std::process::exit(1); - }); + let eviction_policy = parse_eviction_policy(&args.eviction_policy) + .unwrap_or_else(|e| exit_err(format!("invalid --eviction-policy value: {e}"))); let shard_count = args.shards.unwrap_or_else(|| { std::thread::available_parallelism() @@ -275,8 +276,7 @@ async fn main() { }); if shard_count == 0 { - eprintln!("--shards must be at least 1"); - std::process::exit(1); + exit_err("--shards must be at least 1"); } // load encryption key if configured @@ -284,10 +284,7 @@ async fn main() { let encryption_key = if let Some(ref key_path) = args.encryption_key_file { match ember_persistence::encryption::EncryptionKey::from_file(key_path) { Ok(key) => Some(key), - Err(e) => { - eprintln!("failed to load encryption key: {e}"); - std::process::exit(1); - } + Err(e) => exit_err(format!("failed to load encryption key: {e}")), } } else { None @@ -295,8 +292,7 @@ async fn main() { #[cfg(feature = "encryption")] if encryption_key.is_some() && !args.appendonly && args.data_dir.is_none() { - eprintln!("--encryption-key-file requires --data-dir and --appendonly"); - std::process::exit(1); + exit_err("--encryption-key-file requires --data-dir and --appendonly"); } let persistence = build_persistence_config( @@ -348,8 +344,7 @@ async fn main() { if let Some(metrics_port) = args.metrics_port { let metrics_addr = parse_bind_addr(&args.host, metrics_port, "metrics"); if let Err(e) = metrics::install_exporter(metrics_addr) { - eprintln!("failed to start metrics exporter: {e}"); - std::process::exit(1); + exit_err(format!("failed to start metrics exporter: {e}")); } } @@ -371,22 +366,17 @@ async fn main() { // build TLS config if --tls-port is set let tls_config = if let Some(tls_port) = args.tls_port { - let cert_file = args.tls_cert_file.unwrap_or_else(|| { - eprintln!("--tls-port requires --tls-cert-file and --tls-key-file"); - std::process::exit(1); - }); - let key_file = args.tls_key_file.unwrap_or_else(|| { - eprintln!("--tls-port requires --tls-cert-file and --tls-key-file"); - std::process::exit(1); - }); + let cert_file = args + .tls_cert_file + .unwrap_or_else(|| exit_err("--tls-port requires --tls-cert-file and --tls-key-file")); + let key_file = args + .tls_key_file + .unwrap_or_else(|| exit_err("--tls-port requires --tls-cert-file and --tls-key-file")); let auth_clients = match args.tls_auth_clients.to_lowercase().as_str() { "yes" | "true" | "1" => true, "no" | "false" | "0" => false, - _ => { - eprintln!("--tls-auth-clients must be 'yes' or 'no'"); - std::process::exit(1); - } + _ => exit_err("--tls-auth-clients must be 'yes' or 'no'"), }; let tls_addr = parse_bind_addr(&args.host, tls_port, "TLS"); @@ -412,23 +402,20 @@ async fn main() { // validate cluster mode if args.cluster_enabled && args.concurrent { - eprintln!("error: --cluster-enabled and --concurrent are mutually exclusive"); - std::process::exit(1); + exit_err("error: --cluster-enabled and --concurrent are mutually exclusive"); } if args.cluster_bootstrap && !args.cluster_enabled { - eprintln!("error: --cluster-bootstrap requires --cluster-enabled"); - std::process::exit(1); + exit_err("error: --cluster-bootstrap requires --cluster-enabled"); } // build cluster coordinator if cluster mode is enabled let cluster: Option> = if args.cluster_enabled { if args.port.checked_add(args.cluster_port_offset).is_none() { - eprintln!( + exit_err(format!( "error: port {} + cluster-port-offset {} exceeds u16 range", args.port, args.cluster_port_offset - ); - std::process::exit(1); + )); } let local_id = NodeId::new(); diff --git a/crates/ember-server/src/pubsub.rs b/crates/ember-server/src/pubsub.rs index b589f5c9..f3cff8f1 100644 --- a/crates/ember-server/src/pubsub.rs +++ b/crates/ember-server/src/pubsub.rs @@ -193,19 +193,25 @@ impl PubSubManager { /// - `[abc]` matches any character in the set /// - `\x` escapes the next character /// -/// This matches Redis behavior for PSUBSCRIBE patterns. +/// Matching is byte-wise, which is correct for Redis-compatible pub/sub: +/// Redis treats patterns as raw byte sequences, and all metacharacters +/// (`*`, `?`, `[`, `\`) are ASCII so byte comparison is unambiguous. +/// This avoids allocating `Vec` on every match call. fn glob_match(pattern: &str, input: &str) -> bool { - let pat: Vec = pattern.chars().collect(); - let inp: Vec = input.chars().collect(); - glob_match_inner(&pat, &inp) + glob_match_inner(pattern.as_bytes(), input.as_bytes()) } -fn glob_match_inner(pat: &[char], inp: &[char]) -> bool { +/// Inner backtracking glob matcher operating on byte slices. +/// +/// The algorithm tracks the last `*` position in both the pattern and +/// input. On a mismatch it rewinds to that checkpoint and advances the +/// input by one byte — standard linear-time glob matching. +fn glob_match_inner(pat: &[u8], inp: &[u8]) -> bool { let (mut pi, mut ii) = (0, 0); let (mut star_pi, mut star_ii) = (usize::MAX, usize::MAX); while ii < inp.len() { - if pi < pat.len() && pat[pi] == '\\' && pi + 1 < pat.len() { + if pi < pat.len() && pat[pi] == b'\\' && pi + 1 < pat.len() { // escaped character — must match literally pi += 1; if inp[ii] == pat[pi] { @@ -213,16 +219,16 @@ fn glob_match_inner(pat: &[char], inp: &[char]) -> bool { ii += 1; continue; } - } else if pi < pat.len() && pat[pi] == '?' { + } else if pi < pat.len() && pat[pi] == b'?' { pi += 1; ii += 1; continue; - } else if pi < pat.len() && pat[pi] == '*' { + } else if pi < pat.len() && pat[pi] == b'*' { star_pi = pi; star_ii = ii; pi += 1; continue; - } else if pi < pat.len() && pat[pi] == '[' { + } else if pi < pat.len() && pat[pi] == b'[' { // character class if let Some((matched, end)) = match_char_class(&pat[pi..], inp[ii]) { if matched { @@ -249,22 +255,22 @@ fn glob_match_inner(pat: &[char], inp: &[char]) -> bool { } // consume trailing stars - while pi < pat.len() && pat[pi] == '*' { + while pi < pat.len() && pat[pi] == b'*' { pi += 1; } pi == pat.len() } -/// Matches a `[...]` character class. Returns (matched, chars_consumed) -/// if the bracket expression is valid. -fn match_char_class(pat: &[char], ch: char) -> Option<(bool, usize)> { - if pat.is_empty() || pat[0] != '[' { +/// Matches a `[...]` character class against a single byte. +/// Returns `(matched, bytes_consumed_from_pat)` when the bracket is valid. +fn match_char_class(pat: &[u8], ch: u8) -> Option<(bool, usize)> { + if pat.is_empty() || pat[0] != b'[' { return None; } let mut i = 1; - let negate = if i < pat.len() && pat[i] == '^' { + let negate = if i < pat.len() && pat[i] == b'^' { i += 1; true } else { @@ -272,8 +278,8 @@ fn match_char_class(pat: &[char], ch: char) -> Option<(bool, usize)> { }; let mut matched = false; - while i < pat.len() && pat[i] != ']' { - if i + 2 < pat.len() && pat[i + 1] == '-' { + while i < pat.len() && pat[i] != b']' { + if i + 2 < pat.len() && pat[i + 1] == b'-' { // range: [a-z] if ch >= pat[i] && ch <= pat[i + 2] { matched = true; @@ -287,7 +293,7 @@ fn match_char_class(pat: &[char], ch: char) -> Option<(bool, usize)> { } } - if i < pat.len() && pat[i] == ']' { + if i < pat.len() && pat[i] == b']' { Some((matched ^ negate, i + 1)) } else { None // unterminated bracket diff --git a/crates/ember-server/src/server.rs b/crates/ember-server/src/server.rs index a4fee6c7..c7cd1971 100644 --- a/crates/ember-server/src/server.rs +++ b/crates/ember-server/src/server.rs @@ -98,16 +98,7 @@ pub async fn run( let max_conn = max_connections.unwrap_or(DEFAULT_MAX_CONNECTIONS); let semaphore = Arc::new(Semaphore::new(max_conn)); - // set up TLS listener if configured - let tls_listener: Option<(TcpListener, TlsAcceptor)> = if let Some((tls_addr, tls_config)) = tls - { - let acceptor = crate::tls::load_tls_acceptor(&tls_config)?; - let tls_tcp = TcpListener::bind(tls_addr).await?; - info!("TLS listening on {tls_addr}"); - Some((tls_tcp, acceptor)) - } else { - None - }; + let tls_listener = setup_tls_listener(tls).await?; let ctx = Arc::new(ServerContext { start_time: Instant::now(), @@ -368,16 +359,7 @@ pub async fn run_concurrent( let max_conn = max_connections.unwrap_or(DEFAULT_MAX_CONNECTIONS); let semaphore = Arc::new(Semaphore::new(max_conn)); - // set up TLS listener if configured - let tls_listener: Option<(TcpListener, TlsAcceptor)> = if let Some((tls_addr, tls_config)) = tls - { - let acceptor = crate::tls::load_tls_acceptor(&tls_config)?; - let tls_tcp = TcpListener::bind(tls_addr).await?; - info!("TLS listening on {tls_addr}"); - Some((tls_tcp, acceptor)) - } else { - None - }; + let tls_listener = setup_tls_listener(tls).await?; let ctx = Arc::new(ServerContext { start_time: Instant::now(), @@ -516,6 +498,24 @@ pub async fn run_concurrent( Ok(()) } +/// Binds a TLS listener if TLS is configured, otherwise returns `None`. +/// +/// Extracted to avoid repeating the same bind + acceptor creation in both +/// `run` and `run_concurrent`. +async fn setup_tls_listener( + tls: Option<(SocketAddr, TlsConfig)>, +) -> Result, Box> { + match tls { + Some((tls_addr, tls_config)) => { + let acceptor = crate::tls::load_tls_acceptor(&tls_config)?; + let listener = TcpListener::bind(tls_addr).await?; + info!("TLS listening on {tls_addr}"); + Ok(Some((listener, acceptor))) + } + None => Ok(None), + } +} + /// Returns true if the connection should be rejected by protected mode. /// /// Protected mode activates when all three conditions hold: diff --git a/crates/ember-server/src/tls.rs b/crates/ember-server/src/tls.rs index 93f291fd..33d7776f 100644 --- a/crates/ember-server/src/tls.rs +++ b/crates/ember-server/src/tls.rs @@ -63,6 +63,18 @@ pub enum TlsError { VerifierError(String), } +/// Returns `Err` if `path` does not exist on disk. +/// +/// The `make_err` closure maps the path string to the appropriate error variant, +/// keeping the three existence checks in `load_tls_acceptor` as one-liners. +fn require_file(path: &Path, make_err: impl FnOnce(String) -> TlsError) -> Result<(), TlsError> { + if path.exists() { + Ok(()) + } else { + Err(make_err(path.to_string_lossy().into_owned())) + } +} + /// Loads TLS configuration and creates a `TlsAcceptor`. /// /// Reads certificates and private key from PEM files. If `ca_cert_file` is @@ -71,9 +83,7 @@ pub enum TlsError { pub fn load_tls_acceptor(config: &TlsConfig) -> Result { // load server certificates let cert_path = Path::new(&config.cert_file); - if !cert_path.exists() { - return Err(TlsError::CertFileNotFound(config.cert_file.clone())); - } + require_file(cert_path, TlsError::CertFileNotFound)?; let cert_file = File::open(cert_path).map_err(TlsError::CertReadError)?; let cert_reader = BufReader::new(cert_file); @@ -87,9 +97,7 @@ pub fn load_tls_acceptor(config: &TlsConfig) -> Result { // load private key let key_path = Path::new(&config.key_file); - if !key_path.exists() { - return Err(TlsError::KeyFileNotFound(config.key_file.clone())); - } + require_file(key_path, TlsError::KeyFileNotFound)?; let key_file = File::open(key_path).map_err(TlsError::KeyReadError)?; let key_reader = BufReader::new(key_file); @@ -100,9 +108,7 @@ pub fn load_tls_acceptor(config: &TlsConfig) -> Result { let server_config = if let Some(ref ca_path) = config.ca_cert_file { // load CA cert for client verification let ca_cert_path = Path::new(ca_path); - if !ca_cert_path.exists() { - return Err(TlsError::CaCertFileNotFound(ca_path.clone())); - } + require_file(ca_cert_path, TlsError::CaCertFileNotFound)?; let ca_file = File::open(ca_cert_path).map_err(TlsError::CaCertReadError)?; let ca_reader = BufReader::new(ca_file);