Merge branch 'main' into feat/internode-transport-observability

.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

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<F>(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<F, Fut>(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<F>(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<F, Fut>(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() {

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)?;
             }
         }
 

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<F, Fut>(test_fn: F)
 where

crates/utils/src/notify/net.rs

@@ -98,16 +98,46 @@ pub fn parse_host(s: &str) -> Result<Host, NetError> {
         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");

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<F, Fut>(test_fn: F)
 where