diff --git a/.config/make/help.mak b/.config/make/help.mak index 044bf1b8a..a1ad37353 100644 --- a/.config/make/help.mak +++ b/.config/make/help.mak @@ -4,7 +4,7 @@ .PHONY: help help: ## Shows This Help Menu echo -e "$$HEADER" - grep -E '(^[a-zA-Z0-9_-]+:.*?## .*$$)|(^## )' $(MAKEFILE_LIST) | sed 's/^[^:]*://g' | awk 'BEGIN {FS = ":.*?## | #"} ; {printf "${cyan}%-30s${reset} ${white}%s${reset} ${green}%s${reset}\n", $$1, $$2, $$3}' | sed -e 's/\[36m##/\n[32m##/' + grep -E '(^[a-zA-Z0-9_-]+:.*?## .*$$)|(^## )' $(MAKEFILE_LIST) | sed 's/^[^:]*://g' | awk 'BEGIN {FS = ":.*?## | #"} /^## / {printf "\n${green}%s${reset}\n", $$0; next} {printf "${cyan}%-30s${reset} ${white}%s${reset} ${green}%s${reset}\n", $$1, $$2, $$3}' .PHONY: help-build help-build: ## Shows RustFS build help diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index dd4e4c53e..8b41f25ed 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -112,11 +112,20 @@ fn resolve_transition_worker_count() -> (i64, i64, i64) { .filter(|value| *value > 0) .unwrap_or(fallback); let mut effective = configured; - let absolute_max = get_env_i64(ENV_TRANSITION_WORKERS_ABSOLUTE_MAX, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX); + let absolute_max = resolve_transition_workers_absolute_max(); effective = std::cmp::min(effective, absolute_max); (configured, absolute_max, effective) } +fn resolve_transition_workers_absolute_max() -> i64 { + let absolute_max = get_env_i64(ENV_TRANSITION_WORKERS_ABSOLUTE_MAX, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX); + if absolute_max > 0 { + absolute_max + } else { + DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX + } +} + fn resolve_transition_queue_capacity() -> usize { get_env_usize(ENV_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_QUEUE_CAPACITY).max(1) } @@ -942,7 +951,7 @@ impl TransitionState { n = effective; } // Allow environment override of maximum workers - let absolute_max = get_env_i64(ENV_TRANSITION_WORKERS_ABSOLUTE_MAX, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX); + let absolute_max = resolve_transition_workers_absolute_max(); n = std::cmp::min(n, absolute_max); let previous_num_workers = GLOBAL_TransitionState.num_workers.load(Ordering::SeqCst); @@ -2294,6 +2303,9 @@ mod tests { assert!(opts.skip_decommissioned); } + // SAFETY: this helper is only used from `#[serial]` tests and those tests run under a + // single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access + // process environment while `env::set_var`/`env::remove_var` is active. #[allow(unsafe_code)] fn with_transition_worker_env(transition: Option<&str>, absolute: Option<&str>, test_fn: F) where @@ -2343,6 +2355,9 @@ mod tests { } } + // SAFETY: this helper is only used from `#[serial]` tests and those tests run under a + // single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access + // process environment while `env::set_var`/`env::remove_var` is active. #[allow(unsafe_code)] async fn with_transition_worker_env_async(transition: Option<&str>, absolute: Option<&str>, test_fn: F) where @@ -2393,6 +2408,9 @@ mod tests { } } + // SAFETY: this helper is only used from `#[serial]` tests and those tests run under a + // single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access + // process environment while `env::set_var`/`env::remove_var` is active. #[allow(unsafe_code)] fn with_transition_queue_env(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F) where @@ -2442,6 +2460,9 @@ mod tests { } } + // SAFETY: this helper is only used from `#[serial]` tests and those tests run under a + // single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access + // process environment while `env::set_var`/`env::remove_var` is active. #[allow(unsafe_code)] async fn with_transition_queue_env_async(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F) where @@ -2554,6 +2575,26 @@ mod tests { }); } + #[test] + #[serial] + fn resolve_transition_worker_count_ignores_non_positive_absolute_max() { + with_transition_worker_env(Some("4"), Some("0"), || { + let (configured, absolute_max, effective) = resolve_transition_worker_count(); + + assert_eq!(configured, 4); + assert_eq!(absolute_max, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX); + assert_eq!(effective, 4); + }); + + with_transition_worker_env(Some("4"), Some("-1"), || { + let (configured, absolute_max, effective) = resolve_transition_worker_count(); + + assert_eq!(configured, 4); + assert_eq!(absolute_max, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX); + assert_eq!(effective, 4); + }); + } + #[test] #[serial] fn resolve_transition_worker_count_falls_back_for_zero_value() { diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 5fba72685..a7baa18b3 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -1179,14 +1179,9 @@ impl LocalDisk { f.write_all(buf).await.map_err(to_file_error)?; } InternalBuf::Owned(buf) => { - // Reduce one copy by using the owned buffer directly. - // It may be more efficient for larger writes. - let mut f = f.into_std().await; - let task = tokio::task::spawn_blocking(move || { - use std::io::Write as _; - f.write_all(buf.as_ref()).map_err(to_file_error) - }); - task.await??; + f.write_all(buf.as_ref()).await.map_err(to_file_error)?; + // Ensure write errors are observed before returning for owned buffers. + f.sync_all().await.map_err(to_file_error)?; } } diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index 40b0d6fd4..9f5bebec5 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/crates/scanner/tests/lifecycle_integration_test.rs @@ -807,6 +807,9 @@ async fn wait_for_transition( } } +// SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio +// runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the +// `env::set_var` / `env::remove_var` window. #[allow(unsafe_code)] async fn with_forced_immediate_enqueue_timeout(test_fn: F) where diff --git a/crates/utils/src/notify/net.rs b/crates/utils/src/notify/net.rs index 391a67811..befe4b193 100644 --- a/crates/utils/src/notify/net.rs +++ b/crates/utils/src/notify/net.rs @@ -98,16 +98,46 @@ pub fn parse_host(s: &str) -> Result { true }; - // Split host and port, similar to net.SplitHostPort. - let (host_str, port_str) = s.rsplit_once(':').map_or((s, ""), |(h, p)| (h, p)); - let port = if !port_str.is_empty() { - Some(port_str.parse().map_err(|_| NetError::ParseError(port_str.to_string()))?) - } else { - None - }; + let (host, port) = if let Some(rest) = s.strip_prefix('[') { + let Some(end) = rest.find(']') else { + return Err(NetError::MissingBracket); + }; + let host = rest[..end].to_string(); + let port_str = &rest[end + 1..]; + let port = if let Some(port_str) = port_str.strip_prefix(':') { + if port_str.is_empty() { + None + } else { + Some(port_str.parse().map_err(|_| NetError::ParseError(port_str.to_string()))?) + } + } else if port_str.is_empty() { + None + } else { + return Err(NetError::InvalidHost); + }; - // Trim IPv6 brackets if present. - let host = trim_ipv6(host_str)?; + (host, port) + } else { + if s.contains(']') { + return Err(NetError::MissingBracket); + } + + // A host with multiple colons is an IPv6 literal, optionally with a + // zone identifier. Unbracketed IPv6 with port is ambiguous, so callers + // must use the standard bracketed form when they need a port. + let (host_str, port_str) = if s.matches(':').count() > 1 { + (s, "") + } else { + s.rsplit_once(':').map_or((s, ""), |(h, p)| (h, p)) + }; + let port = if !port_str.is_empty() { + Some(port_str.parse().map_err(|_| NetError::ParseError(port_str.to_string()))?) + } else { + None + }; + + (trim_ipv6(host_str)?, port) + }; // Handle IPv6 zone identifier. let trimmed_host = host.split('%').next().unwrap_or(&host); @@ -409,6 +439,33 @@ mod tests { assert_eq!(host.port, Some(8080)); } + #[test] + fn parse_host_with_bare_ipv6_without_port() { + let result = parse_host("::1"); + assert!(result.is_ok()); + let host = result.unwrap(); + assert_eq!(host.name, "::1"); + assert_eq!(host.port, None); + } + + #[test] + fn parse_host_with_ipv6_zone_without_port() { + let result = parse_host("fe80::1%eth0"); + assert!(result.is_ok()); + let host = result.unwrap(); + assert_eq!(host.name, "fe80::1%eth0"); + assert_eq!(host.port, None); + } + + #[test] + fn parse_host_with_bracketed_ipv6_without_port() { + let result = parse_host("[::1]"); + assert!(result.is_ok()); + let host = result.unwrap(); + assert_eq!(host.name, "::1"); + assert_eq!(host.port, None); + } + #[test] fn parse_host_with_invalid_ipv6_missing_bracket() { let result = parse_host("::1]:8080"); diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 02ee25543..09f660308 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -327,6 +327,9 @@ async fn wait_for_transition( } } +// SAFETY: this helper is used only by `#[serial]` tests and runs under the single-threaded Tokio +// runtime (`worker_threads = 1`), so no concurrent test can mutate process environment during the +// `env::set_var` / `env::remove_var` window. #[allow(unsafe_code)] async fn with_forced_immediate_enqueue_timeout(test_fn: F) where