feat(connect): collect native Linux thread states (#7806)

Collect bounded aggregate thread states from Linux procfs while preserving explicit unsupported outcomes for unavailable scopes.

rustfs/src/connect/diagnostics/mod.rs

@@ -95,8 +95,9 @@ pub use perf_site_replication::{
 pub use profile_cpu::{
     CPU_PROFILE_CAPABILITY, LocalProfileConsent, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY, PROFILE_SCHEMA_VERSION,
     ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance, ProfileReasonCode, ProfileResult,
-    ProfileTool, SavedProfileExport, SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileScope, capture_cpu_profile,
-    encode_signed_profile_export, export_cpu_profile, save_signed_profile_export,
+    ProfileTool, SavedProfileExport, SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileData, ThreadProfileScope,
+    ThreadState, ThreadStateCount, capture_cpu_profile, encode_signed_profile_export, export_cpu_profile,
+    save_signed_profile_export,
 };
 pub use profile_memory::export_memory_profile;
 pub use profile_threads::{capture_thread_profile, export_thread_profile};

rustfs/src/connect/diagnostics/profile_cpu.rs

@@ -300,6 +300,44 @@ impl MemoryProfileData {
     }
 }
 
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ThreadProfileData {
+    scope: ThreadProfileScope,
+    states: Vec<ThreadStateCount>,
+}
+
+impl ThreadProfileData {
+    pub(super) fn native(states: Vec<ThreadStateCount>) -> Self {
+        Self {
+            scope: ThreadProfileScope::NativeThreads,
+            states,
+        }
+    }
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
+pub enum ThreadState {
+    Runnable,
+    Waiting,
+    Blocked,
+    Unknown,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ThreadStateCount {
+    state: ThreadState,
+    thread_count: u64,
+}
+
+impl ThreadStateCount {
+    pub(super) const fn new(state: ThreadState, thread_count: u64) -> Self {
+        Self { state, thread_count }
+    }
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 pub enum ThreadProfileScope {
@@ -312,6 +350,7 @@ pub enum ThreadProfileScope {
 pub enum ProfileData {
     Cpu(CpuProfileData),
     Memory(MemoryProfileData),
+    Threads(ThreadProfileData),
 }
 
 #[derive(Clone, Debug, PartialEq, Eq, Serialize)]
@@ -416,6 +455,8 @@ pub enum ProfileError {
     SourceUnavailable,
     #[error("profile_counter_reset")]
     CounterReset,
+    #[error("profile_collection_failed")]
+    CollectionFailed,
     #[error("profile_export_signing_failed")]
     Signing,
     #[error("profile_export_exists")]

rustfs/src/connect/diagnostics/profile_threads.rs

@@ -12,43 +12,189 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Explicit thread/runtime profile capability result.
+//! Bounded thread-state profile collection.
 //!
-//! Dial9 currently exposes session and disk-buffer state, not bounded counts
-//! for RUNNABLE, WAITING, BLOCKED, and UNKNOWN threads. Native thread state
-//! would require a separate reviewed platform adapter. Neither source is
-//! relabelled as the contract's thread data.
+//! Linux native collection reads only the state byte from this process's
+//! `/proc/self/task/*/stat` records. Thread names, identifiers, stacks, paths,
+//! addresses, and raw procfs bytes cannot enter the exported result. Tokio
+//! runtime state remains explicitly unsupported because Dial9 does not expose
+//! the contract's RUNNABLE, WAITING, BLOCKED, and UNKNOWN counts.
 
+#[cfg(target_os = "linux")]
+use std::fs::{self, File};
+#[cfg(target_os = "linux")]
+use std::io::{ErrorKind, Read as _};
+#[cfg(target_os = "linux")]
+use std::time::Instant;
 use tokio_util::sync::CancellationToken;
 
+#[cfg(target_os = "linux")]
+use super::profile_cpu::{CollectorLease, ProfileData, ThreadProfileData, ThreadState, ThreadStateCount};
 use super::profile_cpu::{
-    ProfileCaptureRequest, ProfileError, ProfileReasonCode, ProfileResult, ProfileTool, ThreadProfileScope, check_cancel,
-    encode_signed_profile_export, unix_now,
+    ProfileCaptureRequest, ProfileError, ProfileReasonCode, ProfileResult, ProfileTool, SignedProfileExport, ThreadProfileScope,
+    check_cancel, encode_signed_profile_export, unix_now,
 };
 use crate::connect::DeviceIdentity;
 
-use super::profile_cpu::SignedProfileExport;
+#[cfg(target_os = "linux")]
+const PROC_TASK_DIRECTORY: &str = "/proc/self/task";
+#[cfg(target_os = "linux")]
+const MAX_NATIVE_THREADS: usize = 4_096;
+#[cfg(target_os = "linux")]
+const MAX_PROC_STAT_BYTES: u64 = 4_096;
 
 pub fn capture_thread_profile(
     request: &ProfileCaptureRequest,
-    _scope: ThreadProfileScope,
+    scope: ThreadProfileScope,
     cancel: &CancellationToken,
 ) -> Result<ProfileResult, ProfileError> {
     request.validate(ProfileTool::Threads, unix_now()?)?;
     check_cancel(cancel)?;
-    Ok(ProfileResult::unsupported(
+
+    if scope == ThreadProfileScope::TokioRuntime {
+        return Ok(ProfileResult::unsupported(
+            request,
+            ProfileTool::Threads,
+            ProfileReasonCode::UnsupportedTool,
+        ));
+    }
+
+    #[cfg(not(target_os = "linux"))]
+    return Ok(ProfileResult::unsupported(
         request,
         ProfileTool::Threads,
-        ProfileReasonCode::UnsupportedTool,
-    ))
+        ProfileReasonCode::UnsupportedPlatform,
+    ));
+
+    #[cfg(target_os = "linux")]
+    {
+        let _lease = CollectorLease::acquire()?;
+        let started = Instant::now();
+        let deadline = started.checked_add(request.duration).ok_or(ProfileError::LimitExceeded)?;
+        let data = collect_native_thread_states(cancel, deadline)?;
+        check_cancel(cancel)?;
+        return Ok(ProfileResult::succeeded(
+            request,
+            ProfileTool::Threads,
+            started.elapsed(),
+            ProfileData::Threads(data),
+        ));
+    }
 }
 
-pub fn export_thread_profile(
+pub async fn export_thread_profile(
     request: &ProfileCaptureRequest,
     scope: ThreadProfileScope,
     key: &DeviceIdentity,
     cancel: &CancellationToken,
 ) -> Result<SignedProfileExport, ProfileError> {
-    let result = capture_thread_profile(request, scope, cancel)?;
+    let owned_request = request.clone();
+    let owned_cancel = cancel.clone();
+    let result = tokio::task::spawn_blocking(move || capture_thread_profile(&owned_request, scope, &owned_cancel))
+        .await
+        .map_err(|_| ProfileError::CollectionFailed)??;
     encode_signed_profile_export(request, &result, key, cancel)
 }
+
+#[cfg(target_os = "linux")]
+fn collect_native_thread_states(cancel: &CancellationToken, deadline: Instant) -> Result<ThreadProfileData, ProfileError> {
+    let entries = fs::read_dir(PROC_TASK_DIRECTORY).map_err(|_| ProfileError::SourceUnavailable)?;
+    let mut counts = [0_u64; 4];
+    let mut visited = 0_usize;
+
+    for entry in entries {
+        check_cancel(cancel)?;
+        if Instant::now() >= deadline {
+            return Err(ProfileError::TimedOut);
+        }
+        let entry = entry.map_err(|_| ProfileError::SourceUnavailable)?;
+        let name = entry.file_name();
+        let Some(name) = name.to_str() else {
+            return Err(ProfileError::SourceUnavailable);
+        };
+        if name.is_empty() || !name.bytes().all(|byte| byte.is_ascii_digit()) {
+            return Err(ProfileError::SourceUnavailable);
+        }
+        visited = visited.checked_add(1).ok_or(ProfileError::LimitExceeded)?;
+        if visited > MAX_NATIVE_THREADS {
+            return Err(ProfileError::LimitExceeded);
+        }
+
+        let state = match read_proc_stat_state(&entry.path().join("stat"))? {
+            Some(state) => state,
+            None => continue,
+        };
+        let index = match state {
+            ThreadState::Runnable => 0,
+            ThreadState::Waiting => 1,
+            ThreadState::Blocked => 2,
+            ThreadState::Unknown => 3,
+        };
+        counts[index] = counts[index].checked_add(1).ok_or(ProfileError::LimitExceeded)?;
+    }
+
+    if counts.iter().all(|count| *count == 0) {
+        return Err(ProfileError::SourceUnavailable);
+    }
+
+    Ok(ThreadProfileData::native(vec![
+        ThreadStateCount::new(ThreadState::Runnable, counts[0]),
+        ThreadStateCount::new(ThreadState::Waiting, counts[1]),
+        ThreadStateCount::new(ThreadState::Blocked, counts[2]),
+        ThreadStateCount::new(ThreadState::Unknown, counts[3]),
+    ]))
+}
+
+#[cfg(target_os = "linux")]
+fn read_proc_stat_state(path: &std::path::Path) -> Result<Option<ThreadState>, ProfileError> {
+    let file = match File::open(path) {
+        Ok(file) => file,
+        Err(error) if error.kind() == ErrorKind::NotFound => return Ok(None),
+        Err(_) => return Err(ProfileError::SourceUnavailable),
+    };
+    let mut bytes = Vec::with_capacity(256);
+    file.take(MAX_PROC_STAT_BYTES + 1)
+        .read_to_end(&mut bytes)
+        .map_err(|_| ProfileError::SourceUnavailable)?;
+    if bytes.is_empty() || bytes.len() as u64 > MAX_PROC_STAT_BYTES {
+        return Err(ProfileError::SourceUnavailable);
+    }
+    parse_proc_stat_state(&bytes).map(Some)
+}
+
+#[cfg(target_os = "linux")]
+fn parse_proc_stat_state(stat: &[u8]) -> Result<ThreadState, ProfileError> {
+    let closing = stat
+        .iter()
+        .rposition(|byte| *byte == b')')
+        .ok_or(ProfileError::SourceUnavailable)?;
+    let suffix = stat.get(closing + 1..).ok_or(ProfileError::SourceUnavailable)?;
+    let state = match suffix {
+        [b' ', state, b' ', ..] => *state,
+        _ => return Err(ProfileError::SourceUnavailable),
+    };
+    Ok(match state {
+        b'R' => ThreadState::Runnable,
+        b'D' => ThreadState::Blocked,
+        b'S' | b'I' | b'T' | b't' | b'W' => ThreadState::Waiting,
+        _ => ThreadState::Unknown,
+    })
+}
+
+#[cfg(all(test, target_os = "linux"))]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn proc_stat_parser_uses_only_the_kernel_state_byte() {
+        for (raw, expected) in [
+            (b"123 (worker secret path) R 1 2".as_slice(), ThreadState::Runnable),
+            (b"123 (worker) S 1 2", ThreadState::Waiting),
+            (b"123 (worker) D 1 2", ThreadState::Blocked),
+            (b"123 (worker) Z 1 2", ThreadState::Unknown),
+        ] {
+            assert_eq!(parse_proc_stat_state(raw).expect("valid proc stat"), expected);
+        }
+        assert!(matches!(parse_proc_stat_state(b"123 malformed"), Err(ProfileError::SourceUnavailable)));
+    }
+}

rustfs/src/connect/mod.rs

@@ -70,14 +70,15 @@ pub use diagnostics::{
     TELEMETRY_OTLP_CAPABILITY, TELEMETRY_RECORD_CAPABILITY, TELEMETRY_REPLAY_CAPABILITY, TELEMETRY_SCHEMA_VERSION,
     THREAD_PROFILE_CAPABILITY, TelemetryArtifactConsent, TelemetryArtifactError, TelemetryArtifactRequest, TelemetryCoverage,
     TelemetryDiagnosticResult, TelemetryOperation, TelemetryOutcome, TelemetryProducerError, TelemetryProvenance,
-    TelemetryReasonCode, TelemetrySpan, TelemetrySpanStatus, TelemetryTool, ThreadProfileScope, TraceAnalysis,
-    TraceAnalysisError, TraceRecordCapture, TraceRecordCompletion, TraceRecordLimits, TraceReplayError, analyze_trace,
-    capture_cpu_profile, capture_thread_profile, encode_signed_profile_export, encode_signed_telemetry_export,
-    export_cpu_profile, export_logs, export_memory_profile, export_thread_profile, export_trace_otlp, export_trace_otlp_result,
-    measure_client, measure_drive, read_protected_client_credential, record_diagnostic_result, record_trace, record_trace_bus,
-    replay_trace, replay_trace_result, run_local_environment_once, save_signed_client_export, save_signed_drive_export,
-    save_signed_log_export, save_signed_profile_export, save_signed_telemetry_export, sign_client_export, sign_drive_export,
-    spawn_environment_schedule, validate_client_limits, validate_drive_limits,
+    TelemetryReasonCode, TelemetrySpan, TelemetrySpanStatus, TelemetryTool, ThreadProfileData, ThreadProfileScope, ThreadState,
+    ThreadStateCount, TraceAnalysis, TraceAnalysisError, TraceRecordCapture, TraceRecordCompletion, TraceRecordLimits,
+    TraceReplayError, analyze_trace, capture_cpu_profile, capture_thread_profile, encode_signed_profile_export,
+    encode_signed_telemetry_export, export_cpu_profile, export_logs, export_memory_profile, export_thread_profile,
+    export_trace_otlp, export_trace_otlp_result, measure_client, measure_drive, read_protected_client_credential,
+    record_diagnostic_result, record_trace, record_trace_bus, replay_trace, replay_trace_result, run_local_environment_once,
+    save_signed_client_export, save_signed_drive_export, save_signed_log_export, save_signed_profile_export,
+    save_signed_telemetry_export, sign_client_export, sign_drive_export, spawn_environment_schedule, validate_client_limits,
+    validate_drive_limits,
 };
 pub use diagnostics::{
     LocalNetworkConsent, MAX_NETWORK_ARCHIVE_BYTES, MAX_NETWORK_BANDWIDTH_BYTES_PER_SECOND, MAX_NETWORK_DECOMPRESSED_BYTES,

rustfs/src/startup_entrypoint.rs

@@ -1250,7 +1250,9 @@ async fn execute_connect_profile(options: ConnectProfileOpts) -> Result<()> {
                         Some(ConnectThreadProfileScope::NativeThreads) => ThreadProfileScope::NativeThreads,
                         None => return Err(Error::other("--thread-scope is required for the threads profile")),
                     };
-                    export_thread_profile(&request, scope, &key, &cancel).map_err(Error::other)
+                    export_thread_profile(&request, scope, &key, &cancel)
+                        .await
+                        .map_err(Error::other)
                 }
             }
         };

rustfs/tests/connect_profile_threads.rs

@@ -29,7 +29,11 @@ use profile_cpu::{
     THREAD_PROFILE_CAPABILITY, ThreadProfileScope,
 };
 use profile_threads::{capture_thread_profile, export_thread_profile};
+#[cfg(target_os = "linux")]
+use std::io::{Cursor, Read as _};
 use tokio_util::sync::CancellationToken;
+#[cfg(target_os = "linux")]
+use zip::ZipArchive;
 
 fn request() -> ProfileCaptureRequest {
     let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64;
@@ -58,24 +62,84 @@ fn request() -> ProfileCaptureRequest {
     }
 }
 
-#[test]
-fn tokio_and_native_thread_scopes_remain_explicitly_unsupported() {
+#[tokio::test]
+async fn tokio_thread_scope_remains_explicitly_unsupported() {
     let key = connect::DeviceIdentity::generate();
-    for scope in [ThreadProfileScope::TokioRuntime, ThreadProfileScope::NativeThreads] {
-        let result = capture_thread_profile(&request(), scope, &CancellationToken::new()).expect("unsupported result");
-        assert_eq!(result.outcome(), ProfileOutcome::Unsupported);
-        assert_eq!(result.reason_code(), ProfileReasonCode::UnsupportedTool);
-        assert!(result.data().is_none(), "unsupported scope must not publish zero state counts");
-        let json = serde_json::to_value(result).expect("result JSON");
-        assert_eq!(json["toolId"], "profile.threads");
-        assert_eq!(json["capability"], "profile.threads@1");
-        assert!(json["data"].is_null());
+    let result = capture_thread_profile(&request(), ThreadProfileScope::TokioRuntime, &CancellationToken::new())
+        .expect("unsupported result");
+    assert_eq!(result.outcome(), ProfileOutcome::Unsupported);
+    assert_eq!(result.reason_code(), ProfileReasonCode::UnsupportedTool);
+    assert!(result.data().is_none(), "unsupported scope must not publish zero state counts");
+    let json = serde_json::to_value(result).expect("result JSON");
+    assert_eq!(json["toolId"], "profile.threads");
+    assert_eq!(json["capability"], "profile.threads@1");
+    assert!(json["data"].is_null());
 
-        let export = export_thread_profile(&request(), scope, &key, &CancellationToken::new()).expect("unsupported export");
-        assert_eq!(export.tool.id(), "profile.threads");
-        assert_eq!(export.outcome, ProfileOutcome::Unsupported);
-        assert_eq!(export.reason_code, ProfileReasonCode::UnsupportedTool);
+    let export = export_thread_profile(&request(), ThreadProfileScope::TokioRuntime, &key, &CancellationToken::new())
+        .await
+        .expect("unsupported export");
+    assert_eq!(export.tool.id(), "profile.threads");
+    assert_eq!(export.outcome, ProfileOutcome::Unsupported);
+    assert_eq!(export.reason_code, ProfileReasonCode::UnsupportedTool);
+}
+
+#[cfg(target_os = "linux")]
+#[tokio::test]
+async fn native_thread_scope_exports_bounded_redacted_state_counts() {
+    let key = connect::DeviceIdentity::generate();
+    let result = capture_thread_profile(&request(), ThreadProfileScope::NativeThreads, &CancellationToken::new())
+        .expect("native thread result");
+    assert_eq!(result.outcome(), ProfileOutcome::Succeeded);
+    assert_eq!(result.reason_code(), ProfileReasonCode::Complete);
+    let json = serde_json::to_value(result).expect("result JSON");
+    assert_eq!(json["data"]["scope"], "NATIVE_THREADS");
+    let states = json["data"]["states"].as_array().expect("thread states");
+    assert_eq!(states.len(), 4);
+    assert!(states.iter().all(|state| state["threadCount"].as_u64().is_some()));
+    assert!(states.iter().map(|state| state["threadCount"].as_u64().unwrap()).sum::<u64>() > 0);
+    let encoded = serde_json::to_string(&json).expect("encoded result");
+    for forbidden in ["/proc/", "task/", "worker", "secret", "stack", "address", "threadId"] {
+        assert!(!encoded.contains(forbidden), "result leaked forbidden material: {forbidden}");
     }
+
+    let export = export_thread_profile(&request(), ThreadProfileScope::NativeThreads, &key, &CancellationToken::new())
+        .await
+        .expect("native thread export");
+    assert_eq!(export.outcome, ProfileOutcome::Succeeded);
+    assert!(export.archive_bytes.len() <= profile_cpu::MAX_ARCHIVE_BYTES);
+    let mut archive = ZipArchive::new(Cursor::new(export.archive_bytes)).expect("profile archive");
+    let mut result = String::new();
+    archive
+        .by_name("result.json")
+        .expect("profile result")
+        .read_to_string(&mut result)
+        .expect("read profile result");
+    assert!(result.contains("\"scope\":\"NATIVE_THREADS\""));
+    for forbidden in ["/proc/", "task/", "worker", "secret", "stack", "address", "threadId"] {
+        assert!(!result.contains(forbidden), "archive leaked forbidden material: {forbidden}");
+    }
+}
+
+#[cfg(target_os = "linux")]
+#[test]
+fn native_thread_scope_honors_the_monotonic_deadline() {
+    let mut expired = request();
+    expired.duration = Duration::from_nanos(1);
+    expired.sample_period = Duration::from_nanos(1);
+    assert!(matches!(
+        capture_thread_profile(&expired, ThreadProfileScope::NativeThreads, &CancellationToken::new()),
+        Err(ProfileError::TimedOut)
+    ));
+}
+
+#[cfg(not(target_os = "linux"))]
+#[test]
+fn native_thread_scope_is_explicitly_unsupported_off_linux() {
+    let result = capture_thread_profile(&request(), ThreadProfileScope::NativeThreads, &CancellationToken::new())
+        .expect("unsupported result");
+    assert_eq!(result.outcome(), ProfileOutcome::Unsupported);
+    assert_eq!(result.reason_code(), ProfileReasonCode::UnsupportedPlatform);
+    assert!(result.data().is_none());
 }
 
 #[test]