Skip to content

refactor(server): reduce duplication in server, tls, pubsub, and main #164

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
kacy merged 1 commit into from
Feb 17, 2026
Merged

refactor(server): reduce duplication in server, tls, pubsub, and main #164

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
93 changes: 40 additions & 53 deletions crates/ember-server/src/main.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
));
}
}
}
Expand All @@ -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);
}
}
}
Expand All @@ -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,
Expand All @@ -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()
Expand All @@ -275,28 +276,23 @@ 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
#[cfg(feature = "encryption")]
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
};

#[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(
Expand Down Expand Up @@ -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}"));
}
}

Expand All @@ -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");
Expand All @@ -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<Arc<ClusterCoordinator>> = 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();
Expand Down
42 changes: 24 additions & 18 deletions crates/ember-server/src/pubsub.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,36 +193,42 @@ 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<char>` on every match call.
fn glob_match(pattern: &str, input: &str) -> bool {
let pat: Vec<char> = pattern.chars().collect();
let inp: Vec<char> = 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] {
pi += 1;
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 {
Expand All @@ -249,31 +255,31 @@ 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 {
false
};

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;
Expand All @@ -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
Expand Down
40 changes: 20 additions & 20 deletions crates/ember-server/src/server.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand Down Expand Up @@ -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(),
Expand Down Expand Up @@ -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<Option<(TcpListener, TlsAcceptor)>, Box<dyn std::error::Error>> {
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:
Expand Down
Loading