Add bounded local telemetry export commands (#7720)

feat(connect): add bounded telemetry producers

Cargo.lock

@@ -9526,12 +9526,14 @@ dependencies = [
  "mime_guess",
  "moka",
  "opentelemetry",
+ "opentelemetry-proto",
  "opentelemetry_sdk",
  "p256 0.14.0",
  "parking_lot",
  "percent-encoding",
  "pin-project-lite",
  "proptest",
+ "prost 0.14.4",
  "quick-xml",
  "rand 0.10.2",
  "rcgen",

rustfs/Cargo.toml

@@ -307,6 +307,7 @@ bytes = { workspace = true, features = ["serde"] }
 chrono = { workspace = true, features = ["serde"] }
 flate2 = { workspace = true }
 flatbuffers.workspace = true
+prost.workspace = true
 rmp-serde.workspace = true
 quick-xml.workspace = true
 rustfs-signer.workspace = true
@@ -364,6 +365,7 @@ chacha20poly1305 = { workspace = true }
 # Observability and Metrics
 metrics = { workspace = true }
 opentelemetry = { workspace = true }
+opentelemetry-proto = { workspace = true, features = ["trace"] }
 tracing-opentelemetry = { workspace = true }
 # Data structures
 hashbrown = { workspace = true, features = ["serde", "rayon"] }

rustfs/src/config/cli.rs

@@ -131,6 +131,113 @@ pub enum ConnectCommands {
     Profile(ConnectProfileOpts),
     /// Capture allow-listed local log events and write a signed export
     Logs(ConnectLogsOpts),
+    /// Record, forward, or replay consent-bound telemetry
+    Telemetry(ConnectTelemetryOpts),
+}
+
+#[derive(Args, Clone)]
+pub struct ConnectTelemetryOpts {
+    #[command(subcommand)]
+    pub command: ConnectTelemetryCommands,
+}
+
+#[derive(Subcommand, Clone)]
+pub enum ConnectTelemetryCommands {
+    /// Capture process-local telemetry when an approved typed source is available
+    Record(ConnectTelemetryRecordOpts),
+    /// Forward an OTLP protobuf batch from stdin to a customer-approved collector
+    Otlp(ConnectTelemetryOtlpOpts),
+    /// Replay reviewed trace JSON read from stdin and write a signed export
+    Replay(ConnectTelemetryReplayOpts),
+}
+
+#[derive(Args, Clone)]
+pub struct ConnectTelemetryArtifactOpts {
+    /// Directory containing an enrolled Connect device identity
+    #[arg(long = "state-dir")]
+    pub state_dir: PathBuf,
+
+    /// New local archive path; an existing file is never replaced
+    #[arg(long)]
+    pub output: PathBuf,
+
+    /// Organization resource name bound to the export
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub organization: String,
+
+    /// Cluster resource name bound to the export
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub cluster: String,
+
+    /// Cluster-device resource name bound to the export
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub device: String,
+
+    /// UUIDv7 diagnostic run identifier issued by Connect
+    #[arg(long = "run-uid", value_parser = NonEmptyStringValueParser::new())]
+    pub run_uid: String,
+
+    /// UUIDv7 artifact identifier issued by Connect
+    #[arg(long = "artifact-uid", value_parser = NonEmptyStringValueParser::new())]
+    pub artifact_uid: String,
+
+    /// UUIDv7 consent identifier issued by Connect
+    #[arg(long = "consent-uid", value_parser = NonEmptyStringValueParser::new())]
+    pub consent_uid: String,
+
+    /// Consent policy revision bound to this operation
+    #[arg(long = "policy-revision")]
+    pub policy_revision: u64,
+
+    /// Consent expiry as UTC Unix seconds
+    #[arg(long = "consent-expires-at")]
+    pub consent_expires_at_unix: i64,
+
+    /// Artifact expiry as UTC Unix seconds
+    #[arg(long = "expires-at")]
+    pub expires_at_unix: i64,
+
+    /// Confirm this explicit local L3 telemetry operation
+    #[arg(long = "acknowledge-l3", required = true, action = clap::ArgAction::SetTrue)]
+    pub acknowledge_l3: bool,
+}
+
+#[derive(Args, Clone)]
+pub struct ConnectTelemetryRecordOpts {
+    #[command(flatten)]
+    pub artifact: ConnectTelemetryArtifactOpts,
+
+    /// Capture duration in milliseconds
+    #[arg(long = "duration-millis")]
+    pub duration_millis: u64,
+
+    /// Maximum exported span count
+    #[arg(long = "max-spans", default_value_t = 1_024)]
+    pub max_spans: usize,
+}
+
+#[derive(Args, Clone)]
+pub struct ConnectTelemetryOtlpOpts {
+    #[command(flatten)]
+    pub artifact: ConnectTelemetryArtifactOpts,
+
+    /// Customer-approved OTLP/HTTP traces endpoint
+    #[arg(long, value_parser = NonEmptyStringValueParser::new())]
+    pub endpoint: String,
+
+    /// Forward timeout in milliseconds
+    #[arg(long = "timeout-millis")]
+    pub timeout_millis: u64,
+
+    /// Environment variable containing the local Authorization header value
+    #[arg(long = "authorization-env", value_parser = NonEmptyStringValueParser::new())]
+    pub authorization_env: Option<String>,
+}
+
+#[derive(Args, Clone)]
+pub struct ConnectTelemetryReplayOpts {
+    #[command(flatten)]
+    pub artifact: ConnectTelemetryArtifactOpts,
 }
 
 /// `connect logs` options.
@@ -690,6 +797,8 @@ pub enum CommandResult {
     ConnectProfile(ConnectProfileOpts),
     /// Consent-bound local Connect log export
     ConnectLogs(ConnectLogsOpts),
+    /// Consent-bound local Connect telemetry operation
+    ConnectTelemetry(ConnectTelemetryCommands),
 }
 
 /// Create default ServerOpts from environment variables
@@ -947,6 +1056,43 @@ mod tests {
         assert!(error.to_string().contains("--acknowledge-l3"));
     }
 
+    #[test]
+    fn connect_telemetry_record_requires_explicit_l3_acknowledgement() {
+        let arguments = [
+            "rustfs",
+            "connect",
+            "telemetry",
+            "record",
+            "--state-dir",
+            "/var/lib/rustfs/connect",
+            "--output",
+            "/tmp/telemetry.zip",
+            "--organization",
+            "organizations/019e3ae0-0000-7000-8000-000000000001",
+            "--cluster",
+            "organizations/019e3ae0-0000-7000-8000-000000000001/clusters/019e3ae0-0000-7000-8000-000000000002",
+            "--device",
+            "organizations/019e3ae0-0000-7000-8000-000000000001/clusters/019e3ae0-0000-7000-8000-000000000002/clusterDevices/019e3ae0-0000-7000-8000-000000000003",
+            "--run-uid",
+            "019e3ae0-0000-7000-8000-000000000004",
+            "--artifact-uid",
+            "019e3ae0-0000-7000-8000-000000000005",
+            "--consent-uid",
+            "019e3ae0-0000-7000-8000-000000000006",
+            "--policy-revision",
+            "1",
+            "--consent-expires-at",
+            "4102444800",
+            "--expires-at",
+            "4102444700",
+            "--duration-millis",
+            "10",
+        ];
+        let error = Cli::try_parse_from(arguments).expect_err("unacknowledged telemetry capture must fail");
+        assert_eq!(error.kind(), ErrorKind::MissingRequiredArgument);
+        assert!(error.to_string().contains("--acknowledge-l3"));
+    }
+
     #[test]
     fn server_help_lists_allocator_reclaim_environment() {
         let result = Cli::try_parse_from(["rustfs", "server", "--help"]);

rustfs/src/config/mod.rs

@@ -54,6 +54,10 @@ pub use cli::{CommandResult, InfoOpts, InfoType};
 pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts};
 pub use cli::{ConnectLogsMode, ConnectLogsOpts};
 pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope};
+pub use cli::{
+    ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectTelemetryOtlpOpts, ConnectTelemetryRecordOpts,
+    ConnectTelemetryReplayOpts,
+};
 pub use cli::{DiagnoseFormat, DiagnoseOpts};
 pub use cli::{InspectBucketMetaOpts, InspectCommands, InspectOpts};
 pub use cli::{TlsCommands, TlsInspectOpts, TlsOpts};

rustfs/src/config/opt.rs

@@ -142,6 +142,7 @@ impl Opt {
                 ConnectCommands::License(opts) => Ok(CommandResult::ConnectLicense(opts.command)),
                 ConnectCommands::Profile(opts) => Ok(CommandResult::ConnectProfile(opts)),
                 ConnectCommands::Logs(opts) => Ok(CommandResult::ConnectLogs(opts)),
+                ConnectCommands::Telemetry(opts) => Ok(CommandResult::ConnectTelemetry(opts.command)),
             },
             Some(Commands::Server(opts)) => Self::server_command_result(Self::from_server_opts(*opts)),
             None => {

rustfs/src/connect/diagnostics/mod.rs

@@ -17,6 +17,10 @@ mod profile_cpu;
 mod profile_memory;
 mod profile_threads;
 mod schedule;
+mod trace_analysis;
+mod trace_otlp;
+mod trace_record;
+mod trace_replay;
 
 pub use logs::{
     CaptureMode, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalLogConsent, LogCaptureError, LogCaptureRequest, LogProvenance,
@@ -34,3 +38,17 @@ pub use schedule::{
     DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError, DiagnosticScheduleRuntime, DiagnosticScheduleStatus,
     ReceiptOutcome, run_local_environment_once, spawn_environment_schedule,
 };
+pub use trace_analysis::{OperationSummary, TraceAnalysis, TraceAnalysisError, analyze_trace};
+pub use trace_otlp::{
+    LocalOtlpHeaders, MAX_OTLP_BODY_BYTES, OtlpBatch, OtlpForwardError, OtlpReceipt, export_trace_otlp, export_trace_otlp_result,
+};
+pub use trace_record::{
+    LocalTelemetryConsent, MAX_SAFE_INTEGER, MAX_TELEMETRY_DURATION, MAX_TELEMETRY_RESULT_BYTES, MAX_TELEMETRY_SPANS,
+    ObservedTelemetrySpan, RecordedTrace, SavedTelemetryExport, SignedTelemetryExport, TELEMETRY_OTLP_CAPABILITY,
+    TELEMETRY_RECORD_CAPABILITY, TELEMETRY_REPLAY_CAPABILITY, TELEMETRY_SCHEMA_VERSION, TelemetryArtifactConsent,
+    TelemetryArtifactError, TelemetryArtifactRequest, TelemetryCoverage, TelemetryDiagnosticResult, TelemetryOperation,
+    TelemetryOutcome, TelemetryProducerError, TelemetryProvenance, TelemetryReasonCode, TelemetrySpan, TelemetrySpanStatus,
+    TelemetryTool, TraceRecordCapture, TraceRecordCompletion, TraceRecordLimits, encode_signed_telemetry_export,
+    record_diagnostic_result, record_trace, record_trace_bus, save_signed_telemetry_export,
+};
+pub use trace_replay::{LocallyReviewedTraceArtifact, ReplayedTrace, TraceReplayError, replay_trace, replay_trace_result};

rustfs/src/connect/diagnostics/trace_analysis.rs

@@ -0,0 +1,149 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Deterministic local summaries over redacted telemetry spans.
+
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+use tokio_util::sync::CancellationToken;
+
+use super::trace_record::{
+    LocalTelemetryConsent, TelemetryOperation, TelemetryProducerError, TelemetrySpanStatus, acquire_telemetry_lease,
+};
+use super::trace_replay::ReplayedTrace;
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields, rename_all = "camelCase")]
+pub struct TraceAnalysis {
+    pub input_artifact_sha256: String,
+    pub span_count: u64,
+    pub error_count: u64,
+    pub total_duration_micros: u64,
+    pub operations: Vec<OperationSummary>,
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields, rename_all = "camelCase")]
+pub struct OperationSummary {
+    pub operation: TelemetryOperation,
+    pub span_count: u64,
+    pub error_count: u64,
+    pub total_duration_micros: u64,
+}
+
+#[derive(Debug, Error, Eq, PartialEq)]
+pub enum TraceAnalysisError {
+    #[error("another telemetry operation is already running")]
+    Busy,
+    #[error("local telemetry consent is expired")]
+    ConsentExpired,
+    #[error("telemetry analysis was cancelled")]
+    Cancelled,
+    #[error("telemetry analysis counters overflowed")]
+    CounterOverflow,
+}
+
+pub fn analyze_trace(
+    replay: &ReplayedTrace,
+    consent: LocalTelemetryConsent,
+    cancel: &CancellationToken,
+) -> Result<TraceAnalysis, TraceAnalysisError> {
+    consent.remaining().map_err(map_consent_error)?;
+    if cancel.is_cancelled() {
+        return Err(TraceAnalysisError::Cancelled);
+    }
+    let _lease = acquire_telemetry_lease().map_err(map_consent_error)?;
+
+    let mut operations = [
+        OperationSummary {
+            operation: TelemetryOperation::GetObject,
+            span_count: 0,
+            error_count: 0,
+            total_duration_micros: 0,
+        },
+        OperationSummary {
+            operation: TelemetryOperation::PutObject,
+            span_count: 0,
+            error_count: 0,
+            total_duration_micros: 0,
+        },
+        OperationSummary {
+            operation: TelemetryOperation::HeadObject,
+            span_count: 0,
+            error_count: 0,
+            total_duration_micros: 0,
+        },
+        OperationSummary {
+            operation: TelemetryOperation::ListObjects,
+            span_count: 0,
+            error_count: 0,
+            total_duration_micros: 0,
+        },
+        OperationSummary {
+            operation: TelemetryOperation::InternalRpc,
+            span_count: 0,
+            error_count: 0,
+            total_duration_micros: 0,
+        },
+    ];
+    let mut error_count = 0u64;
+    let mut total_duration_micros = 0u64;
+    for span in &replay.spans {
+        if cancel.is_cancelled() {
+            return Err(TraceAnalysisError::Cancelled);
+        }
+        let summary = &mut operations[operation_index(span.operation)];
+        summary.span_count = summary.span_count.checked_add(1).ok_or(TraceAnalysisError::CounterOverflow)?;
+        summary.total_duration_micros = summary
+            .total_duration_micros
+            .checked_add(span.duration_micros)
+            .ok_or(TraceAnalysisError::CounterOverflow)?;
+        total_duration_micros = total_duration_micros
+            .checked_add(span.duration_micros)
+            .ok_or(TraceAnalysisError::CounterOverflow)?;
+        if span.status == TelemetrySpanStatus::Error {
+            summary.error_count = summary
+                .error_count
+                .checked_add(1)
+                .ok_or(TraceAnalysisError::CounterOverflow)?;
+            error_count = error_count.checked_add(1).ok_or(TraceAnalysisError::CounterOverflow)?;
+        }
+    }
+    let span_count = u64::try_from(replay.spans.len()).map_err(|_| TraceAnalysisError::CounterOverflow)?;
+    Ok(TraceAnalysis {
+        input_artifact_sha256: replay.input_artifact_sha256.clone(),
+        span_count,
+        error_count,
+        total_duration_micros,
+        operations: operations.into_iter().filter(|summary| summary.span_count != 0).collect(),
+    })
+}
+
+const fn operation_index(operation: TelemetryOperation) -> usize {
+    match operation {
+        TelemetryOperation::GetObject => 0,
+        TelemetryOperation::PutObject => 1,
+        TelemetryOperation::HeadObject => 2,
+        TelemetryOperation::ListObjects => 3,
+        TelemetryOperation::InternalRpc => 4,
+    }
+}
+
+fn map_consent_error(error: TelemetryProducerError) -> TraceAnalysisError {
+    match error {
+        TelemetryProducerError::Busy => TraceAnalysisError::Busy,
+        TelemetryProducerError::ConsentExpired => TraceAnalysisError::ConsentExpired,
+        _ => TraceAnalysisError::CounterOverflow,
+    }
+}

rustfs/src/connect/diagnostics/trace_otlp.rs

@@ -0,0 +1,271 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Customer-side OTLP/HTTP forwarding with bounded payloads and local secrets.
+
+use std::time::{Duration, Instant};
+
+use opentelemetry_proto::tonic::collector::trace::v1::ExportTraceServiceRequest;
+use prost::Message as _;
+use reqwest::{
+    Client, StatusCode, Url,
+    header::{self, HeaderMap},
+};
+use serde::{Deserialize, Serialize};
+use thiserror::Error;
+use tokio_util::sync::CancellationToken;
+
+use super::trace_record::{
+    LocalTelemetryConsent, MAX_TELEMETRY_DURATION, TelemetryArtifactRequest, TelemetryDiagnosticResult, TelemetryLease,
+    TelemetryProducerError, TelemetryTool, acquire_telemetry_lease, ensure_result_size,
+};
+
+pub const MAX_OTLP_BODY_BYTES: usize = 1_048_576;
+
+/// An encoded OTLP/HTTP protobuf batch from the local OpenTelemetry adapter.
+/// The bytes are never included in the diagnostic result or error values.
+pub struct OtlpBatch {
+    body: Vec<u8>,
+    span_count: u64,
+    _lease: TelemetryLease,
+}
+
+impl OtlpBatch {
+    pub fn new(body: Vec<u8>) -> Result<Self, OtlpForwardError> {
+        if body.is_empty() || body.len() > MAX_OTLP_BODY_BYTES {
+            return Err(OtlpForwardError::InvalidBatch);
+        }
+        let lease = acquire_telemetry_lease().map_err(map_producer_error)?;
+        let request = ExportTraceServiceRequest::decode(body.as_slice()).map_err(|_| OtlpForwardError::InvalidBatch)?;
+        if request.resource_spans.len() > 1024 {
+            return Err(OtlpForwardError::InvalidBatch);
+        }
+        let scope_count = request.resource_spans.iter().try_fold(0_usize, |count, resource| {
+            count
+                .checked_add(resource.scope_spans.len())
+                .filter(|count| *count <= 1024)
+                .ok_or(OtlpForwardError::InvalidBatch)
+        })?;
+        if scope_count == 0 {
+            return Err(OtlpForwardError::InvalidBatch);
+        }
+        let span_count = request
+            .resource_spans
+            .iter()
+            .flat_map(|resource| &resource.scope_spans)
+            .try_fold(0_u64, |count, scope| {
+                let spans = u64::try_from(scope.spans.len()).map_err(|_| OtlpForwardError::InvalidBatch)?;
+                count.checked_add(spans).ok_or(OtlpForwardError::InvalidBatch)
+            })?;
+        if span_count == 0 || span_count > 1024 {
+            return Err(OtlpForwardError::InvalidBatch);
+        }
+        Ok(Self {
+            body,
+            span_count,
+            _lease: lease,
+        })
+    }
+}
+
+pub async fn export_trace_otlp_result(
+    request: &TelemetryArtifactRequest,
+    endpoint: Url,
+    headers: LocalOtlpHeaders,
+    batch: OtlpBatch,
+    consent: LocalTelemetryConsent,
+    timeout: Duration,
+    cancel: &CancellationToken,
+) -> Result<TelemetryDiagnosticResult<OtlpReceipt>, OtlpForwardError> {
+    let started = Instant::now();
+    let receipt = export_trace_otlp(endpoint, headers, batch, consent, timeout, cancel).await?;
+    Ok(TelemetryDiagnosticResult::succeeded(
+        request,
+        TelemetryTool::Otlp,
+        started.elapsed(),
+        receipt,
+    ))
+}
+
+/// Locally supplied collector authentication. This type has no `Debug` or
+/// serialization implementation, keeping credentials out of results and logs.
+pub struct LocalOtlpHeaders(HeaderMap);
+
+impl LocalOtlpHeaders {
+    pub fn new(headers: HeaderMap) -> Self {
+        Self(headers)
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields, rename_all = "camelCase")]
+pub struct OtlpReceipt {
+    pub accepted_span_count: u64,
+    pub rejected_span_count: u64,
+    pub exported_bytes: u64,
+}
+
+#[derive(Debug, Error, Eq, PartialEq)]
+pub enum OtlpForwardError {
+    #[error("another telemetry operation is already running")]
+    Busy,
+    #[error("OTLP endpoint must be HTTPS or an HTTP loopback address without credentials, query or fragment")]
+    Endpoint,
+    #[error("OTLP batch must contain 1..1024 spans and 1..1048576 bytes")]
+    InvalidBatch,
+    #[error("OTLP forward timeout must be between 1ms and 30s")]
+    InvalidTimeout,
+    #[error("OTLP forwarding was cancelled")]
+    Cancelled,
+    #[error("local OTLP collector rejected the batch")]
+    Rejected,
+    #[error("local OTLP collector is unavailable")]
+    Unavailable,
+    #[error("OTLP receipt exceeds its result limit")]
+    ResultTooLarge,
+    #[error("local telemetry consent is expired")]
+    ConsentExpired,
+}
+
+pub async fn export_trace_otlp(
+    endpoint: Url,
+    headers: LocalOtlpHeaders,
+    batch: OtlpBatch,
+    consent: LocalTelemetryConsent,
+    timeout: Duration,
+    cancel: &CancellationToken,
+) -> Result<OtlpReceipt, OtlpForwardError> {
+    validate_endpoint(&endpoint)?;
+    if timeout.is_zero() || timeout > MAX_TELEMETRY_DURATION {
+        return Err(OtlpForwardError::InvalidTimeout);
+    }
+    let remaining = consent.remaining().map_err(map_consent_error)?;
+    let consent_binds = remaining <= timeout;
+    let timeout = timeout.min(remaining);
+    let timeout_millis = usize::try_from(timeout.as_millis()).map_err(|_| OtlpForwardError::InvalidTimeout)?;
+    if timeout_millis == 0 {
+        return Err(OtlpForwardError::InvalidTimeout);
+    }
+    let deadline = tokio::time::Instant::now() + timeout;
+    let OtlpBatch {
+        body,
+        span_count,
+        _lease,
+    } = batch;
+    let byte_budget = MAX_OTLP_BODY_BYTES
+        .checked_mul(timeout_millis.min(1000))
+        .and_then(|bytes| bytes.checked_div(1000))
+        .ok_or(OtlpForwardError::InvalidBatch)?;
+    if body.len() > byte_budget {
+        return Err(OtlpForwardError::InvalidBatch);
+    }
+    let exported_bytes = u64::try_from(body.len()).map_err(|_| OtlpForwardError::InvalidBatch)?;
+    let mut headers = headers.0;
+    for transport_header in [
+        header::HOST,
+        header::CONTENT_LENGTH,
+        header::CONTENT_TYPE,
+        header::CONNECTION,
+        header::TRANSFER_ENCODING,
+    ] {
+        headers.remove(transport_header);
+    }
+    let client = Client::builder()
+        .redirect(reqwest::redirect::Policy::none())
+        .timeout(timeout)
+        .build()
+        .map_err(|_| OtlpForwardError::Unavailable)?;
+
+    let request = client
+        .post(endpoint)
+        .headers(headers)
+        .header(reqwest::header::CONTENT_TYPE, "application/x-protobuf")
+        .body(body);
+    let mut response = tokio::select! {
+        biased;
+        _ = cancel.cancelled() => return Err(OtlpForwardError::Cancelled),
+        _ = tokio::time::sleep_until(deadline) => return Err(deadline_error(consent_binds)),
+        response = request.send() => response.map_err(|_| OtlpForwardError::Unavailable)?,
+    };
+    if response.status() != StatusCode::OK {
+        return Err(if response.status().is_client_error() {
+            OtlpForwardError::Rejected
+        } else {
+            OtlpForwardError::Unavailable
+        });
+    }
+    loop {
+        let chunk = tokio::select! {
+            biased;
+            _ = cancel.cancelled() => return Err(OtlpForwardError::Cancelled),
+            _ = tokio::time::sleep_until(deadline) => return Err(deadline_error(consent_binds)),
+            chunk = response.chunk() => chunk.map_err(|_| OtlpForwardError::Unavailable)?,
+        };
+        let Some(chunk) = chunk else {
+            break;
+        };
+        if !chunk.is_empty() {
+            // A successful OTLP response may carry partial-success details.
+            // Until the approved protobuf adapter is wired, refusing such a
+            // response avoids claiming every span was accepted.
+            return Err(OtlpForwardError::Rejected);
+        }
+    }
+
+    let receipt = OtlpReceipt {
+        accepted_span_count: span_count,
+        rejected_span_count: 0,
+        exported_bytes,
+    };
+    ensure_result_size(&receipt).map_err(|_| OtlpForwardError::ResultTooLarge)?;
+    Ok(receipt)
+}
+
+fn deadline_error(consent_binds: bool) -> OtlpForwardError {
+    if consent_binds {
+        OtlpForwardError::ConsentExpired
+    } else {
+        OtlpForwardError::Unavailable
+    }
+}
+
+fn validate_endpoint(endpoint: &Url) -> Result<(), OtlpForwardError> {
+    let loopback_http = endpoint.scheme() == "http"
+        && endpoint
+            .host_str()
+            .is_some_and(|host| matches!(host, "localhost" | "127.0.0.1" | "[::1]" | "::1"));
+    if (endpoint.scheme() != "https" && !loopback_http)
+        || !endpoint.username().is_empty()
+        || endpoint.password().is_some()
+        || endpoint.query().is_some()
+        || endpoint.fragment().is_some()
+    {
+        return Err(OtlpForwardError::Endpoint);
+    }
+    Ok(())
+}
+
+fn map_consent_error(error: TelemetryProducerError) -> OtlpForwardError {
+    match error {
+        TelemetryProducerError::ConsentExpired => OtlpForwardError::ConsentExpired,
+        _ => OtlpForwardError::Unavailable,
+    }
+}
+
+fn map_producer_error(error: TelemetryProducerError) -> OtlpForwardError {
+    match error {
+        TelemetryProducerError::Busy => OtlpForwardError::Busy,
+        _ => OtlpForwardError::InvalidBatch,
+    }
+}

rustfs/src/connect/diagnostics/trace_replay.rs

@@ -0,0 +1,127 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! In-memory replay of a locally reviewed telemetry record.
+//!
+//! This module never opens a path and never replays S3 operations. The caller
+//! supplies reviewed bytes; replay validates the exact closed record shape and
+//! returns only its redacted operation/timing/status observations.
+
+use std::time::Instant;
+
+use serde::{Deserialize, Serialize};
+use sha2::{Digest, Sha256};
+use thiserror::Error;
+use tokio_util::sync::CancellationToken;
+
+use super::trace_record::{
+    LocalTelemetryConsent, MAX_SAFE_INTEGER, MAX_TELEMETRY_RESULT_BYTES, MAX_TELEMETRY_SPANS, RecordedTrace,
+    TelemetryArtifactRequest, TelemetryDiagnosticResult, TelemetryProducerError, TelemetrySpan, TelemetryTool,
+    acquire_telemetry_lease, ensure_result_size,
+};
+
+pub struct LocallyReviewedTraceArtifact<'a> {
+    bytes: &'a [u8],
+}
+
+impl<'a> LocallyReviewedTraceArtifact<'a> {
+    pub fn new(bytes: &'a [u8]) -> Result<Self, TraceReplayError> {
+        if bytes.is_empty() || bytes.len() > MAX_TELEMETRY_RESULT_BYTES {
+            return Err(TraceReplayError::InvalidArtifact);
+        }
+        Ok(Self { bytes })
+    }
+}
+
+#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
+#[serde(deny_unknown_fields, rename_all = "camelCase")]
+pub struct ReplayedTrace {
+    pub input_artifact_sha256: String,
+    pub spans: Vec<TelemetrySpan>,
+}
+
+#[derive(Debug, Error, Eq, PartialEq)]
+pub enum TraceReplayError {
+    #[error("another telemetry operation is already running")]
+    Busy,
+    #[error("reviewed telemetry artifact is invalid")]
+    InvalidArtifact,
+    #[error("telemetry replay was cancelled")]
+    Cancelled,
+    #[error("local telemetry consent is expired")]
+    ConsentExpired,
+    #[error("telemetry replay result exceeds 262144 bytes")]
+    ResultTooLarge,
+}
+
+pub fn replay_trace(
+    artifact: LocallyReviewedTraceArtifact<'_>,
+    consent: LocalTelemetryConsent,
+    cancel: &CancellationToken,
+) -> Result<ReplayedTrace, TraceReplayError> {
+    consent.remaining().map_err(map_consent_error)?;
+    if cancel.is_cancelled() {
+        return Err(TraceReplayError::Cancelled);
+    }
+    let _lease = acquire_telemetry_lease().map_err(map_consent_error)?;
+    let record: RecordedTrace = serde_json::from_slice(artifact.bytes).map_err(|_| TraceReplayError::InvalidArtifact)?;
+    if record.spans.len() > MAX_TELEMETRY_SPANS
+        || record.dropped_span_count > MAX_SAFE_INTEGER
+        || record.spans.iter().any(|span| span.duration_micros > MAX_SAFE_INTEGER)
+    {
+        return Err(TraceReplayError::InvalidArtifact);
+    }
+    if cancel.is_cancelled() {
+        return Err(TraceReplayError::Cancelled);
+    }
+    let result = ReplayedTrace {
+        input_artifact_sha256: hex_lower(&Sha256::digest(artifact.bytes)),
+        spans: record.spans,
+    };
+    ensure_result_size(&result).map_err(|_| TraceReplayError::ResultTooLarge)?;
+    Ok(result)
+}
+
+pub fn replay_trace_result(
+    request: &TelemetryArtifactRequest,
+    artifact: LocallyReviewedTraceArtifact<'_>,
+    consent: LocalTelemetryConsent,
+    cancel: &CancellationToken,
+) -> Result<TelemetryDiagnosticResult<ReplayedTrace>, TraceReplayError> {
+    let started = Instant::now();
+    let replay = replay_trace(artifact, consent, cancel)?;
+    Ok(TelemetryDiagnosticResult::succeeded(
+        request,
+        TelemetryTool::Replay,
+        started.elapsed(),
+        replay,
+    ))
+}
+
+fn map_consent_error(error: TelemetryProducerError) -> TraceReplayError {
+    match error {
+        TelemetryProducerError::Busy => TraceReplayError::Busy,
+        TelemetryProducerError::ConsentExpired => TraceReplayError::ConsentExpired,
+        _ => TraceReplayError::InvalidArtifact,
+    }
+}
+
+fn hex_lower(bytes: &[u8]) -> String {
+    use std::fmt::Write as _;
+
+    bytes.iter().fold(String::with_capacity(bytes.len() * 2), |mut output, byte| {
+        let _ = write!(output, "{byte:02x}");
+        output
+    })
+}

rustfs/src/connect/mod.rs

@@ -46,13 +46,22 @@ pub use config::{HeartbeatConfig, HeartbeatConfigError, HeartbeatSchedule};
 pub use credential_store::{CredentialStore, DeviceCredential};
 pub use diagnostics::{
     CPU_PROFILE_CAPABILITY, CaptureMode, DiagnosticCollectionPolicy, DiagnosticReceipt, DiagnosticScheduleError,
-    DiagnosticScheduleRuntime, DiagnosticScheduleStatus, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalLogConsent,
-    LocalProfileConsent, LogCaptureError, LogCaptureRequest, LogProvenance, MAX_PROFILE_DURATION, MEMORY_PROFILE_CAPABILITY,
-    PROFILE_SCHEMA_VERSION, ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance,
-    ProfileReasonCode, ProfileResult, ProfileTool, ReceiptOutcome, SavedLogExport, SavedProfileExport, SignedLogExport,
-    SignedProfileExport, THREAD_PROFILE_CAPABILITY, ThreadProfileScope, capture_cpu_profile, capture_thread_profile,
-    encode_signed_profile_export, export_cpu_profile, export_logs, export_memory_profile, export_thread_profile,
-    run_local_environment_once, save_signed_log_export, save_signed_profile_export, spawn_environment_schedule,
+    DiagnosticScheduleRuntime, DiagnosticScheduleStatus, LOGS_CAPABILITY, LOGS_SCHEMA_VERSION, LocalLogConsent, LocalOtlpHeaders,
+    LocalProfileConsent, LocalTelemetryConsent, LocallyReviewedTraceArtifact, LogCaptureError, LogCaptureRequest, LogProvenance,
+    MAX_OTLP_BODY_BYTES, MAX_PROFILE_DURATION, MAX_SAFE_INTEGER, MAX_TELEMETRY_DURATION, MAX_TELEMETRY_RESULT_BYTES,
+    MAX_TELEMETRY_SPANS, MEMORY_PROFILE_CAPABILITY, ObservedTelemetrySpan, OperationSummary, OtlpBatch, OtlpForwardError,
+    OtlpReceipt, PROFILE_SCHEMA_VERSION, ProfileCaptureRequest, ProfileData, ProfileError, ProfileOutcome, ProfileProvenance,
+    ProfileReasonCode, ProfileResult, ProfileTool, ReceiptOutcome, RecordedTrace, ReplayedTrace, SavedLogExport,
+    SavedProfileExport, SavedTelemetryExport, SignedLogExport, SignedProfileExport, SignedTelemetryExport,
+    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,
+    record_diagnostic_result, record_trace, record_trace_bus, replay_trace, replay_trace_result, run_local_environment_once,
+    save_signed_log_export, save_signed_profile_export, save_signed_telemetry_export, spawn_environment_schedule,
 };
 pub use environment::{
     ENVIRONMENT_CAPABILITY, ENVIRONMENT_SCHEMA_VERSION, EnvironmentCollectionRequest, EnvironmentError,

rustfs/src/startup_entrypoint.rs

@@ -15,7 +15,8 @@
 use crate::{
     config::{
         CommandResult, Config, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts,
-        ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope, Opt,
+        ConnectProfileOpts, ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands,
+        ConnectThreadProfileScope, Opt,
     },
     startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle},
     startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight},
@@ -26,7 +27,8 @@ use crate::{
     storage_api::startup::storage::bootstrap_instance_ctx,
 };
 use std::io::{Error, Read as _, Result};
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+use tokio_util::sync::CancellationToken;
 use tracing::{error, instrument};
 
 const LOG_COMPONENT_MAIN: &str = "main";
@@ -136,6 +138,7 @@ async fn async_main() -> Result<()> {
         CommandResult::ConnectLicense(command) => return execute_connect_license(command),
         CommandResult::ConnectProfile(options) => return execute_connect_profile(options).await,
         CommandResult::ConnectLogs(options) => return execute_connect_logs(options).await,
+        CommandResult::ConnectTelemetry(command) => return execute_connect_telemetry(command).await,
         CommandResult::Server(config) => config,
     };
 
@@ -247,6 +250,231 @@ async fn execute_connect_logs(options: ConnectLogsOpts) -> Result<()> {
     Ok(())
 }
 
+async fn execute_connect_telemetry(command: ConnectTelemetryCommands) -> Result<()> {
+    use crate::connect::{
+        LocalOtlpHeaders, LocallyReviewedTraceArtifact, MAX_OTLP_BODY_BYTES, MAX_TELEMETRY_RESULT_BYTES, OtlpBatch,
+        RecordedTrace, TelemetryDiagnosticResult, TelemetryProducerError, TelemetryTool, TraceRecordLimits, analyze_trace,
+        export_trace_otlp_result, record_diagnostic_result, record_trace_bus, replay_trace_result,
+    };
+    use reqwest::header::{AUTHORIZATION, HeaderMap, HeaderValue};
+
+    match command {
+        ConnectTelemetryCommands::Record(options) => {
+            let (key, request, consent) = telemetry_context(&options.artifact)?;
+            request.validate().map_err(Error::other)?;
+            if options.duration_millis == 0
+                || options.duration_millis > 30_000
+                || options.max_spans == 0
+                || options.max_spans > 1_024
+            {
+                return Err(Error::other("telemetry record limits are invalid"));
+            }
+            let cancel = CancellationToken::new();
+            let started = Instant::now();
+            let capture = record_trace_bus(
+                consent,
+                TraceRecordLimits {
+                    duration: Duration::from_millis(options.duration_millis),
+                    max_spans: options.max_spans,
+                },
+                &cancel,
+            );
+            tokio::pin!(capture);
+            let capture = tokio::select! {
+                biased;
+                signal = tokio::signal::ctrl_c() => {
+                    signal.map_err(Error::other)?;
+                    cancel.cancel();
+                    return Err(Error::other("telemetry record cancelled"));
+                }
+                result = capture.as_mut() => result,
+            };
+            match capture {
+                Ok(capture) => {
+                    let result = record_diagnostic_result(&request, capture, started.elapsed());
+                    save_telemetry_result(&options.artifact, &request, &result, &key, &cancel, None)
+                }
+                Err(TelemetryProducerError::SourceUnavailable) => {
+                    let result = TelemetryDiagnosticResult::<RecordedTrace>::unsupported(
+                        &request,
+                        TelemetryTool::Record,
+                        started.elapsed(),
+                    );
+                    println!("{}", serde_json::to_string(&result).map_err(Error::other)?);
+                    Ok(())
+                }
+                Err(error) => Err(Error::other(error)),
+            }
+        }
+        ConnectTelemetryCommands::Otlp(options) => {
+            let (key, request, consent) = telemetry_context(&options.artifact)?;
+            request.validate().map_err(Error::other)?;
+            let body = read_bounded_stdin(MAX_OTLP_BODY_BYTES)?;
+            let batch = OtlpBatch::new(body).map_err(Error::other)?;
+            let endpoint = reqwest::Url::parse(&options.endpoint).map_err(|_| Error::other("invalid OTLP endpoint"))?;
+            let mut headers = HeaderMap::new();
+            if let Some(name) = options.authorization_env.as_deref() {
+                if !valid_environment_name(name) {
+                    return Err(Error::other("invalid OTLP authorization environment variable name"));
+                }
+                let value = std::env::var(name).map_err(|_| Error::other("OTLP authorization is unavailable"))?;
+                let value = HeaderValue::from_str(&value).map_err(|_| Error::other("OTLP authorization is invalid"))?;
+                headers.insert(AUTHORIZATION, value);
+            }
+            let cancel = CancellationToken::new();
+            let result = export_trace_otlp_result(
+                &request,
+                endpoint,
+                LocalOtlpHeaders::new(headers),
+                batch,
+                consent,
+                Duration::from_millis(options.timeout_millis),
+                &cancel,
+            )
+            .await
+            .map_err(Error::other)?;
+            save_telemetry_result(&options.artifact, &request, &result, &key, &cancel, None)
+        }
+        ConnectTelemetryCommands::Replay(options) => {
+            let (key, request, consent) = telemetry_context(&options.artifact)?;
+            request.validate().map_err(Error::other)?;
+            let bytes = read_bounded_stdin(MAX_TELEMETRY_RESULT_BYTES)?;
+            let cancel = CancellationToken::new();
+            let result = replay_trace_result(
+                &request,
+                LocallyReviewedTraceArtifact::new(&bytes).map_err(Error::other)?,
+                consent,
+                &cancel,
+            )
+            .map_err(Error::other)?;
+            let analysis = analyze_trace(
+                result
+                    .data()
+                    .ok_or_else(|| Error::other("telemetry replay returned no data"))?,
+                consent,
+                &cancel,
+            )
+            .map_err(Error::other)?;
+            save_telemetry_result(
+                &options.artifact,
+                &request,
+                &result,
+                &key,
+                &cancel,
+                Some(serde_json::to_value(analysis).map_err(Error::other)?),
+            )
+        }
+    }
+}
+
+fn telemetry_context(
+    options: &ConnectTelemetryArtifactOpts,
+) -> Result<(
+    crate::connect::DeviceIdentity,
+    crate::connect::TelemetryArtifactRequest,
+    crate::connect::LocalTelemetryConsent,
+)> {
+    use crate::connect::{
+        IdentityStore, LocalTelemetryConsent, TelemetryArtifactConsent, TelemetryArtifactRequest, TelemetryProvenance,
+    };
+    use rand::{TryRng as _, rngs::SysRng};
+
+    let key = IdentityStore::new(options.state_dir.join("identity"))
+        .load()
+        .map_err(Error::other)?
+        .ok_or_else(|| Error::other("connect telemetry requires an enrolled device identity"))?;
+    let executable_sha256 = hash_current_executable()?;
+    let produced_at_unix = unix_now()?;
+    let remaining = options
+        .consent_expires_at_unix
+        .checked_sub(produced_at_unix)
+        .and_then(|seconds| u64::try_from(seconds).ok())
+        .filter(|seconds| *seconds > 0)
+        .ok_or_else(|| Error::other("local telemetry consent is expired"))?;
+    let consent = LocalTelemetryConsent::new(
+        Instant::now()
+            .checked_add(Duration::from_secs(remaining))
+            .ok_or_else(|| Error::other("local telemetry consent is expired"))?,
+    )
+    .map_err(Error::other)?;
+    let mut nonce = [0_u8; 32];
+    SysRng.try_fill_bytes(&mut nonce).map_err(Error::other)?;
+    let request = TelemetryArtifactRequest {
+        organization_name: options.organization.clone(),
+        cluster_name: options.cluster.clone(),
+        device_name: options.device.clone(),
+        run_uid: options.run_uid.clone(),
+        artifact_uid: options.artifact_uid.clone(),
+        schema_version: crate::connect::TELEMETRY_SCHEMA_VERSION,
+        consent: TelemetryArtifactConsent {
+            consent_uid: options.consent_uid.clone(),
+            policy_revision: options.policy_revision,
+            expires_at_unix: options.consent_expires_at_unix,
+            confirmed: options.acknowledge_l3,
+        },
+        produced_at_unix,
+        expires_at_unix: options.expires_at_unix,
+        nonce,
+        provenance: TelemetryProvenance::new(
+            crate::version::build::COMMIT_HASH,
+            executable_sha256,
+            env!("CARGO_PKG_VERSION"),
+            enabled_build_features(),
+        ),
+    };
+    Ok((key, request, consent))
+}
+
+fn save_telemetry_result<T: serde::Serialize>(
+    options: &ConnectTelemetryArtifactOpts,
+    request: &crate::connect::TelemetryArtifactRequest,
+    result: &crate::connect::TelemetryDiagnosticResult<T>,
+    key: &crate::connect::DeviceIdentity,
+    cancel: &CancellationToken,
+    analysis: Option<serde_json::Value>,
+) -> Result<()> {
+    let export = crate::connect::encode_signed_telemetry_export(request, result, key, cancel).map_err(Error::other)?;
+    let receipt = crate::connect::save_signed_telemetry_export(&options.output, &export, cancel).map_err(Error::other)?;
+    println!(
+        "{}",
+        serde_json::json!({
+            "toolId": export.tool.id(),
+            "outcome": export.outcome.as_str(),
+            "reasonCode": export.reason_code.as_str(),
+            "artifactUid": receipt.artifact_uid,
+            "archiveSizeBytes": receipt.archive_size_bytes,
+            "archiveSha256": receipt.archive_sha256,
+            "analysis": analysis,
+            "upload": "NOT_PERFORMED",
+        })
+    );
+    Ok(())
+}
+
+fn read_bounded_stdin(limit: usize) -> Result<Vec<u8>> {
+    let mut bytes = Vec::with_capacity(limit.min(64 * 1024));
+    std::io::stdin()
+        .take(u64::try_from(limit).unwrap_or(u64::MAX).saturating_add(1))
+        .read_to_end(&mut bytes)?;
+    if bytes.is_empty() || bytes.len() > limit {
+        return Err(Error::other("telemetry stdin is empty or exceeds its limit"));
+    }
+    Ok(bytes)
+}
+
+fn valid_environment_name(value: &str) -> bool {
+    !value.is_empty()
+        && value.len() <= 64
+        && value
+            .bytes()
+            .all(|byte| byte.is_ascii_uppercase() || byte.is_ascii_digit() || byte == b'_')
+}
+
+fn unix_now() -> Result<i64> {
+    let duration = SystemTime::now().duration_since(UNIX_EPOCH).map_err(Error::other)?;
+    i64::try_from(duration.as_secs()).map_err(Error::other)
+}
+
 async fn execute_connect_profile(options: ConnectProfileOpts) -> Result<()> {
     use crate::connect::{
         IdentityStore, LocalProfileConsent, ProfileCaptureRequest, ProfileProvenance, ThreadProfileScope, export_cpu_profile,

rustfs/tests/connect_trace_analysis.rs

@@ -0,0 +1,46 @@
+use serial_test::serial;
+use std::time::{Duration, Instant};
+
+use rustfs::connect::{LocalTelemetryConsent, LocallyReviewedTraceArtifact, TraceAnalysisError, analyze_trace, replay_trace};
+use tokio_util::sync::CancellationToken;
+
+fn consent() -> LocalTelemetryConsent {
+    LocalTelemetryConsent::new(Instant::now() + Duration::from_secs(1)).expect("future consent")
+}
+
+#[test]
+#[serial]
+fn connect_trace_analysis_is_deterministic_and_uses_only_observed_values() {
+    let bytes = br#"{"spans":[{"operation":"GET_OBJECT","durationMicros":500,"status":"OK"},{"operation":"GET_OBJECT","durationMicros":250,"status":"ERROR"},{"operation":"INTERNAL_RPC","durationMicros":25,"status":"OK"}],"droppedSpanCount":0}"#;
+    let replay = replay_trace(
+        LocallyReviewedTraceArtifact::new(bytes).expect("bounded artifact"),
+        consent(),
+        &CancellationToken::new(),
+    )
+    .expect("valid replay");
+    let first = analyze_trace(&replay, consent(), &CancellationToken::new()).expect("analysis succeeds");
+    let second = analyze_trace(&replay, consent(), &CancellationToken::new()).expect("analysis is repeatable");
+
+    assert_eq!(first, second);
+    assert_eq!(first.span_count, 3);
+    assert_eq!(first.error_count, 1);
+    assert_eq!(first.total_duration_micros, 775);
+    assert_eq!(first.operations.len(), 2);
+    assert_eq!(first.operations[0].span_count, 2);
+    assert_eq!(first.operations[0].error_count, 1);
+}
+
+#[test]
+#[serial]
+fn connect_trace_analysis_honors_stop() {
+    let replay = replay_trace(
+        LocallyReviewedTraceArtifact::new(br#"{"spans":[],"droppedSpanCount":0}"#).expect("bounded artifact"),
+        consent(),
+        &CancellationToken::new(),
+    )
+    .expect("valid replay");
+    let cancel = CancellationToken::new();
+    cancel.cancel();
+    let error = analyze_trace(&replay, consent(), &cancel).expect_err("cancelled analysis must fail");
+    assert_eq!(error, TraceAnalysisError::Cancelled);
+}

rustfs/tests/connect_trace_otlp.rs

@@ -0,0 +1,220 @@
+use serial_test::serial;
+use std::time::{Duration, Instant};
+
+use opentelemetry_proto::tonic::{
+    collector::trace::v1::ExportTraceServiceRequest,
+    trace::v1::{ResourceSpans, ScopeSpans, Span},
+};
+use prost::Message as _;
+use reqwest::{Url, header};
+use rustfs::connect::{
+    LocalOtlpHeaders, LocalTelemetryConsent, LocallyReviewedTraceArtifact, OtlpBatch, OtlpForwardError, TraceReplayError,
+    export_trace_otlp, replay_trace,
+};
+use tokio::{
+    io::{AsyncReadExt, AsyncWriteExt},
+    net::TcpListener,
+};
+use tokio_util::sync::CancellationToken;
+
+fn consent() -> LocalTelemetryConsent {
+    LocalTelemetryConsent::new(Instant::now() + Duration::from_secs(2)).expect("future consent")
+}
+
+fn encoded_batch(span_count: usize) -> Vec<u8> {
+    ExportTraceServiceRequest {
+        resource_spans: vec![ResourceSpans {
+            scope_spans: vec![ScopeSpans {
+                spans: vec![Span::default(); span_count],
+                ..Default::default()
+            }],
+            ..Default::default()
+        }],
+    }
+    .encode_to_vec()
+}
+
+fn encoded_named_batch(name_length: usize) -> Vec<u8> {
+    let mut span = Span::default();
+    span.name = "x".repeat(name_length);
+    ExportTraceServiceRequest {
+        resource_spans: vec![ResourceSpans {
+            scope_spans: vec![ScopeSpans {
+                spans: vec![span],
+                ..Default::default()
+            }],
+            ..Default::default()
+        }],
+    }
+    .encode_to_vec()
+}
+
+async fn collector(status: &str) -> (Url, tokio::task::JoinHandle<Vec<u8>>) {
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("bind collector");
+    let address = listener.local_addr().expect("collector address");
+    let status = status.to_owned();
+    let task = tokio::spawn(async move {
+        let (mut stream, _) = listener.accept().await.expect("accept OTLP request");
+        let mut received = vec![0u8; 4096];
+        let size = stream.read(&mut received).await.expect("read OTLP request");
+        received.truncate(size);
+        let response = format!("HTTP/1.1 {status}\r\nContent-Length: 0\r\nConnection: close\r\n\r\n");
+        stream.write_all(response.as_bytes()).await.expect("write response");
+        received
+    });
+    (Url::parse(&format!("http://{address}/v1/traces")).expect("collector URL"), task)
+}
+
+#[tokio::test]
+#[serial]
+async fn connect_trace_otlp_forwards_bounded_protobuf_and_keeps_auth_out_of_receipt() {
+    let (endpoint, server) = collector("200 OK").await;
+    let mut headers = header::HeaderMap::new();
+    headers.insert(header::AUTHORIZATION, header::HeaderValue::from_static("Bearer local-secret"));
+    let receipt = export_trace_otlp(
+        endpoint,
+        LocalOtlpHeaders::new(headers),
+        OtlpBatch::new(encoded_batch(2)).expect("valid batch"),
+        consent(),
+        Duration::from_secs(1),
+        &CancellationToken::new(),
+    )
+    .await
+    .expect("collector accepts batch");
+    let request = String::from_utf8(server.await.expect("collector task")).expect("HTTP is UTF-8");
+    assert!(request.to_ascii_lowercase().contains("authorization: bearer local-secret"));
+    assert!(request.to_ascii_lowercase().contains("content-type: application/x-protobuf"));
+    assert_eq!(receipt.accepted_span_count, 2);
+    assert_eq!(receipt.exported_bytes, encoded_batch(2).len() as u64);
+    assert!(
+        !serde_json::to_string(&receipt)
+            .expect("receipt JSON")
+            .contains("local-secret")
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn connect_trace_otlp_reports_fixed_failure_without_response_or_secret_body() {
+    let (endpoint, server) = collector("401 Unauthorized").await;
+    let error = export_trace_otlp(
+        endpoint,
+        LocalOtlpHeaders::new(header::HeaderMap::new()),
+        OtlpBatch::new(encoded_batch(1)).expect("valid batch"),
+        consent(),
+        Duration::from_secs(1),
+        &CancellationToken::new(),
+    )
+    .await
+    .expect_err("rejection must fail");
+    server.await.expect("collector task");
+    assert_eq!(error, OtlpForwardError::Rejected);
+}
+
+#[test]
+#[serial]
+fn connect_trace_otlp_rejects_remote_plaintext_and_oversized_batches() {
+    assert!(OtlpBatch::new(encoded_batch(1)).is_ok());
+    assert!(OtlpBatch::new(encoded_batch(1024)).is_ok());
+    assert!(OtlpBatch::new(encoded_batch(0)).is_err());
+    assert!(OtlpBatch::new(encoded_batch(1025)).is_err());
+    assert!(OtlpBatch::new(vec![0; 1_048_577]).is_err());
+    assert!(OtlpBatch::new(vec![1]).is_err());
+    let endpoint = Url::parse("http://collector.example/v1/traces").expect("URL");
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("runtime");
+    let error = runtime
+        .block_on(export_trace_otlp(
+            endpoint,
+            LocalOtlpHeaders::new(header::HeaderMap::new()),
+            OtlpBatch::new(encoded_batch(1)).expect("valid batch"),
+            consent(),
+            Duration::from_secs(1),
+            &CancellationToken::new(),
+        ))
+        .expect_err("plaintext remote collector must fail");
+    assert_eq!(error, OtlpForwardError::Endpoint);
+}
+
+#[test]
+#[serial]
+fn connect_trace_otlp_lease_blocks_replay_until_the_batch_is_dropped() {
+    let held = OtlpBatch::new(encoded_batch(1)).expect("held parsed batch");
+    let reviewed = br#"{"spans":[],"droppedSpanCount":0}"#;
+    let error = replay_trace(
+        LocallyReviewedTraceArtifact::new(reviewed).expect("reviewed artifact"),
+        consent(),
+        &CancellationToken::new(),
+    )
+    .expect_err("cross-tool concurrency must fail");
+    assert_eq!(error, TraceReplayError::Busy);
+
+    drop(held);
+    replay_trace(
+        LocallyReviewedTraceArtifact::new(reviewed).expect("reviewed artifact"),
+        consent(),
+        &CancellationToken::new(),
+    )
+    .expect("replay succeeds after lease release");
+}
+
+#[tokio::test]
+#[serial]
+async fn connect_trace_otlp_honors_stop_without_contacting_the_collector() {
+    let cancel = CancellationToken::new();
+    cancel.cancel();
+    let error = export_trace_otlp(
+        Url::parse("http://127.0.0.1:9/v1/traces").expect("URL"),
+        LocalOtlpHeaders::new(header::HeaderMap::new()),
+        OtlpBatch::new(encoded_batch(1)).expect("valid batch"),
+        consent(),
+        Duration::from_secs(1),
+        &cancel,
+    )
+    .await
+    .expect_err("cancelled forward must fail");
+    assert_eq!(error, OtlpForwardError::Cancelled);
+}
+
+#[tokio::test]
+#[serial]
+async fn connect_trace_otlp_enforces_the_timeout_byte_budget_before_network_io() {
+    let endpoint = Url::parse("http://127.0.0.1:9/v1/traces").expect("URL");
+    let error = export_trace_otlp(
+        endpoint.clone(),
+        LocalOtlpHeaders::new(header::HeaderMap::new()),
+        OtlpBatch::new(encoded_batch(1024)).expect("bounded actual batch"),
+        consent(),
+        Duration::from_millis(1),
+        &CancellationToken::new(),
+    )
+    .await
+    .expect_err("one millisecond cannot send the full batch budget");
+    assert_eq!(error, OtlpForwardError::InvalidBatch);
+
+    let error = export_trace_otlp(
+        endpoint,
+        LocalOtlpHeaders::new(header::HeaderMap::new()),
+        OtlpBatch::new(encoded_batch(1)).expect("bounded actual batch"),
+        consent(),
+        Duration::from_micros(999),
+        &CancellationToken::new(),
+    )
+    .await
+    .expect_err("sub-millisecond timeout has no byte budget");
+    assert_eq!(error, OtlpForwardError::InvalidTimeout);
+
+    let error = export_trace_otlp(
+        Url::parse("http://127.0.0.1:9/v1/traces").expect("URL"),
+        LocalOtlpHeaders::new(header::HeaderMap::new()),
+        OtlpBatch::new(encoded_named_batch(200_000)).expect("bounded actual batch"),
+        LocalTelemetryConsent::new(Instant::now() + Duration::from_millis(100)).expect("short consent"),
+        Duration::from_secs(30),
+        &CancellationToken::new(),
+    )
+    .await
+    .expect_err("effective consent timeout limits the byte budget");
+    assert_eq!(error, OtlpForwardError::InvalidBatch);
+}

rustfs/tests/connect_trace_record.rs

@@ -0,0 +1,440 @@
+use serial_test::serial;
+use std::io::{Cursor, Read as _};
+#[cfg(unix)]
+use std::os::unix::fs::PermissionsExt as _;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use base64_simd::URL_SAFE_NO_PAD;
+use p256::ecdsa::signature::Verifier as _;
+use p256::ecdsa::{Signature, VerifyingKey};
+use p256::pkcs8::DecodePublicKey as _;
+use rustfs::connect::{
+    DeviceIdentity, LocalTelemetryConsent, ObservedTelemetrySpan, TelemetryArtifactConsent, TelemetryArtifactError,
+    TelemetryArtifactRequest, TelemetryDiagnosticResult, TelemetryOperation, TelemetryProducerError, TelemetryProvenance,
+    TelemetrySpanStatus, TelemetryTool, TraceRecordCompletion, TraceRecordLimits, encode_signed_telemetry_export,
+    record_diagnostic_result, record_trace, record_trace_bus, save_signed_telemetry_export,
+};
+use rustfs_common::trace_bus::{TraceEvent, TraceFunc, TraceKind, trace_emit};
+use sha2::{Digest as _, Sha256};
+use tokio::sync::mpsc;
+use tokio_util::sync::CancellationToken;
+use zip::ZipArchive;
+
+fn consent() -> LocalTelemetryConsent {
+    LocalTelemetryConsent::new(Instant::now() + Duration::from_secs(5)).expect("future consent")
+}
+
+fn artifact_request() -> TelemetryArtifactRequest {
+    let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64;
+    let organization = "organizations/019e3ae0-0000-7000-8000-000000000001";
+    let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000002");
+    TelemetryArtifactRequest {
+        organization_name: organization.to_owned(),
+        cluster_name: cluster.clone(),
+        device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000003"),
+        run_uid: "019e3ae0-0000-7000-8000-000000000004".to_owned(),
+        artifact_uid: "019e3ae0-0000-7000-8000-000000000005".to_owned(),
+        schema_version: 1,
+        consent: TelemetryArtifactConsent {
+            consent_uid: "019e3ae0-0000-7000-8000-000000000006".to_owned(),
+            policy_revision: 1,
+            expires_at_unix: now + 120,
+            confirmed: true,
+        },
+        produced_at_unix: now,
+        expires_at_unix: now + 60,
+        nonce: [0x5a; 32],
+        provenance: TelemetryProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-rc.6", vec![]),
+    }
+}
+
+fn archive_entry(archive: &mut ZipArchive<Cursor<Vec<u8>>>, name: &str) -> Vec<u8> {
+    let mut entry = archive.by_name(name).expect("archive entry");
+    let mut bytes = Vec::new();
+    entry.read_to_end(&mut bytes).expect("read archive entry");
+    bytes
+}
+
+#[tokio::test]
+#[serial]
+async fn connect_trace_record_emits_only_the_frozen_redacted_shape() {
+    let (sender, receiver) = mpsc::channel(4);
+    sender
+        .send(ObservedTelemetrySpan::new(
+            TelemetryOperation::GetObject,
+            Duration::from_micros(500),
+            TelemetrySpanStatus::Ok,
+        ))
+        .await
+        .expect("open source");
+    let task = tokio::spawn(async move {
+        record_trace(
+            receiver,
+            consent(),
+            TraceRecordLimits {
+                duration: Duration::from_millis(15),
+                max_spans: 4,
+            },
+            &CancellationToken::new(),
+        )
+        .await
+    });
+    tokio::time::sleep(Duration::from_millis(20)).await;
+    let record = task.await.expect("capture task").expect("capture succeeds");
+    assert_eq!(record.completion, TraceRecordCompletion::Complete);
+    let json = serde_json::to_value(&record.data).expect("serialize record");
+
+    assert_eq!(json["spans"][0]["operation"], "GET_OBJECT");
+    assert_eq!(json["spans"][0]["durationMicros"], 500);
+    assert_eq!(json["spans"][0]["status"], "OK");
+    assert_eq!(json["droppedSpanCount"], 0);
+    let mut root_keys = json
+        .as_object()
+        .expect("record object")
+        .keys()
+        .map(String::as_str)
+        .collect::<Vec<_>>();
+    let mut span_keys = json["spans"][0]
+        .as_object()
+        .expect("span object")
+        .keys()
+        .map(String::as_str)
+        .collect::<Vec<_>>();
+    root_keys.sort_unstable();
+    span_keys.sort_unstable();
+    assert_eq!(root_keys, ["droppedSpanCount", "spans"]);
+    assert_eq!(span_keys, ["durationMicros", "operation", "status"]);
+}
+
+#[tokio::test]
+#[serial]
+async fn connect_trace_record_stops_at_span_limit_and_counts_queued_drops() {
+    let (sender, receiver) = mpsc::channel(8);
+    for _ in 0..3 {
+        sender
+            .send(ObservedTelemetrySpan::new(
+                TelemetryOperation::InternalRpc,
+                Duration::from_micros(1),
+                TelemetrySpanStatus::Error,
+            ))
+            .await
+            .expect("open source");
+    }
+    let record = record_trace(
+        receiver,
+        consent(),
+        TraceRecordLimits {
+            duration: Duration::from_secs(1),
+            max_spans: 1,
+        },
+        &CancellationToken::new(),
+    )
+    .await
+    .expect("bounded capture succeeds");
+
+    assert_eq!(record.data.spans.len(), 1);
+    assert_eq!(record.data.dropped_span_count, 2);
+    assert_eq!(record.completion, TraceRecordCompletion::LimitExceeded);
+}
+
+#[tokio::test]
+#[serial]
+async fn connect_trace_record_marks_a_closed_source_incomplete_after_real_observations() {
+    let (sender, receiver) = mpsc::channel(1);
+    sender
+        .send(ObservedTelemetrySpan::new(
+            TelemetryOperation::HeadObject,
+            Duration::from_micros(2),
+            TelemetrySpanStatus::Ok,
+        ))
+        .await
+        .expect("open source");
+    drop(sender);
+    let record = record_trace(
+        receiver,
+        consent(),
+        TraceRecordLimits {
+            duration: Duration::from_secs(1),
+            max_spans: 2,
+        },
+        &CancellationToken::new(),
+    )
+    .await
+    .expect("observed data remains usable");
+
+    assert_eq!(record.data.spans.len(), 1);
+    assert_eq!(record.completion, TraceRecordCompletion::SourceUnavailable);
+}
+
+#[tokio::test]
+#[serial]
+async fn connect_trace_record_rejects_limits_and_honors_stop() {
+    let (_sender, receiver) = mpsc::channel(1);
+    let error = record_trace(
+        receiver,
+        consent(),
+        TraceRecordLimits {
+            duration: Duration::from_secs(31),
+            max_spans: 1,
+        },
+        &CancellationToken::new(),
+    )
+    .await
+    .expect_err("overlong capture must fail");
+    assert_eq!(error, TelemetryProducerError::InvalidDuration);
+
+    let (_sender, receiver) = mpsc::channel(1);
+    let cancel = CancellationToken::new();
+    cancel.cancel();
+    let error = record_trace(
+        receiver,
+        consent(),
+        TraceRecordLimits {
+            duration: Duration::from_secs(1),
+            max_spans: 1,
+        },
+        &cancel,
+    )
+    .await
+    .expect_err("cancelled capture must fail");
+    assert_eq!(error, TelemetryProducerError::Cancelled);
+}
+
+#[test]
+#[serial]
+fn connect_trace_record_requires_live_local_consent() {
+    assert_eq!(
+        LocalTelemetryConsent::new(Instant::now()).expect_err("expired consent must fail"),
+        TelemetryProducerError::ConsentExpired
+    );
+}
+
+#[test]
+#[serial]
+fn connect_trace_record_enforces_exact_duration_and_sample_boundaries() {
+    assert!(
+        TraceRecordLimits {
+            duration: Duration::from_secs(30),
+            max_spans: 1024,
+        }
+        .validate()
+        .is_ok()
+    );
+    assert_eq!(
+        TraceRecordLimits {
+            duration: Duration::from_secs(30) + Duration::from_nanos(1),
+            max_spans: 1024,
+        }
+        .validate()
+        .expect_err("duration N+1 must fail"),
+        TelemetryProducerError::InvalidDuration
+    );
+    assert_eq!(
+        TraceRecordLimits {
+            duration: Duration::from_secs(30),
+            max_spans: 1025,
+        }
+        .validate()
+        .expect_err("sample N+1 must fail"),
+        TelemetryProducerError::InvalidSpanLimit
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn connect_trace_record_refuses_to_relabel_the_real_heal_bus_as_frozen_telemetry() {
+    let task = tokio::spawn(async {
+        record_trace_bus(
+            consent(),
+            TraceRecordLimits {
+                duration: Duration::from_millis(80),
+                max_spans: 8,
+            },
+            &CancellationToken::new(),
+        )
+        .await
+    });
+    tokio::time::sleep(Duration::from_millis(10)).await;
+    assert!(trace_emit(|| {
+        TraceEvent::new(TraceKind::Scanner, TraceFunc::ScannerHealCandidate)
+            .with_bucket("SYNTHETIC_SECRET_BUCKET")
+            .with_object("private/object")
+            .with_duration(Duration::from_micros(41))
+            .with_attr("error", "SYNTHETIC_SECRET_ERROR")
+    }));
+    let error = task
+        .await
+        .expect("capture task")
+        .expect_err("heal/scanner events have no frozen telemetry semantics");
+    assert_eq!(error, TelemetryProducerError::SourceUnavailable);
+}
+
+#[test]
+#[serial]
+fn connect_trace_record_writes_a_verified_no_clobber_signed_archive() {
+    let request = artifact_request();
+    let key = DeviceIdentity::generate();
+    let result = record_diagnostic_result(
+        &request,
+        rustfs::connect::TraceRecordCapture {
+            data: rustfs::connect::RecordedTrace {
+                spans: vec![rustfs::connect::TelemetrySpan {
+                    operation: TelemetryOperation::GetObject,
+                    duration_micros: 500,
+                    status: TelemetrySpanStatus::Ok,
+                }],
+                dropped_span_count: 0,
+            },
+            completion: TraceRecordCompletion::Complete,
+        },
+        Duration::from_millis(1),
+    );
+    let export =
+        encode_signed_telemetry_export(&request, &result, &key, &CancellationToken::new()).expect("signed telemetry export");
+    let mut archive = ZipArchive::new(Cursor::new(export.archive_bytes.clone())).expect("telemetry archive");
+    assert_eq!(archive.len(), 3);
+    let envelope_bytes = archive_entry(&mut archive, "envelope.json");
+    let signature_bytes = archive_entry(&mut archive, "envelope.sig");
+    let result_bytes = archive_entry(&mut archive, "result.json");
+    let envelope: serde_json::Value = serde_json::from_slice(&envelope_bytes).expect("envelope JSON");
+    let signature: serde_json::Value = serde_json::from_slice(&signature_bytes).expect("signature JSON");
+    let result_json: serde_json::Value = serde_json::from_slice(&result_bytes).expect("result JSON");
+    assert_eq!(envelope["formatVersion"], "rustfs.connect.diagnosticEnvelope/1");
+    assert_eq!(envelope["toolId"], "telemetry.record");
+    assert_eq!(envelope["classification"], "L3");
+    assert_eq!(
+        envelope["payload"]["sha256"],
+        hex_simd::encode_to_string(Sha256::digest(&result_bytes), hex_simd::AsciiCase::Lower)
+    );
+    assert_eq!(result_json["toolId"], "telemetry.record");
+    assert_eq!(result_json["capability"], "telemetry.record@1");
+    assert_eq!(result_json["outcome"], "SUCCEEDED");
+    assert_eq!(result_json["provenance"]["sourceCommit"], "a".repeat(40));
+
+    let raw_signature = URL_SAFE_NO_PAD
+        .decode_to_vec(signature["value"].as_str().expect("signature value"))
+        .expect("base64url signature");
+    let signature = Signature::from_slice(&raw_signature).expect("P-256 signature");
+    let public = VerifyingKey::from_public_key_der(&key.public_key_der()).expect("public key");
+    let mut input = b"rustfs-diagnostic-envelope-v1\0".to_vec();
+    input.extend_from_slice(&envelope_bytes);
+    public.verify(&input, &signature).expect("valid envelope signature");
+
+    let directory = tempfile::tempdir().expect("temporary output");
+    let output = directory.path().join("telemetry.zip");
+    save_signed_telemetry_export(&output, &export, &CancellationToken::new()).expect("saved telemetry export");
+    #[cfg(unix)]
+    assert_eq!(std::fs::metadata(&output).expect("output metadata").permissions().mode() & 0o777, 0o600);
+    assert!(matches!(
+        save_signed_telemetry_export(&output, &export, &CancellationToken::new()),
+        Err(TelemetryArtifactError::AlreadyExists)
+    ));
+}
+
+#[test]
+#[serial]
+fn connect_trace_record_rejects_tampered_or_cancelled_archives_without_a_file() {
+    let request = artifact_request();
+    let key = DeviceIdentity::generate();
+    let result = TelemetryDiagnosticResult::succeeded(
+        &request,
+        TelemetryTool::Record,
+        Duration::ZERO,
+        rustfs::connect::RecordedTrace {
+            spans: vec![],
+            dropped_span_count: 0,
+        },
+    );
+    let export =
+        encode_signed_telemetry_export(&request, &result, &key, &CancellationToken::new()).expect("signed telemetry export");
+    let directory = tempfile::tempdir().expect("temporary output");
+
+    let tampered_output = directory.path().join("tampered.zip");
+    let mut tampered = export.clone();
+    tampered.archive_bytes.push(0);
+    assert!(matches!(
+        save_signed_telemetry_export(&tampered_output, &tampered, &CancellationToken::new()),
+        Err(TelemetryArtifactError::InvalidRequest)
+    ));
+    assert!(!tampered_output.exists());
+
+    let cancelled_output = directory.path().join("cancelled.zip");
+    let cancel = CancellationToken::new();
+    cancel.cancel();
+    assert!(matches!(
+        save_signed_telemetry_export(&cancelled_output, &export, &cancel),
+        Err(TelemetryArtifactError::Cancelled)
+    ));
+    assert!(!cancelled_output.exists());
+}
+
+#[test]
+#[serial]
+fn connect_trace_record_never_publishes_an_unsupported_artifact() {
+    let request = artifact_request();
+    let key = DeviceIdentity::generate();
+    let result =
+        TelemetryDiagnosticResult::<rustfs::connect::RecordedTrace>::unsupported(&request, TelemetryTool::Record, Duration::ZERO);
+    assert!(matches!(
+        encode_signed_telemetry_export(&request, &result, &key, &CancellationToken::new()),
+        Err(TelemetryArtifactError::InvalidRequest)
+    ));
+}
+
+#[test]
+#[serial]
+fn connect_trace_record_never_publishes_an_incomplete_single_window() {
+    let request = artifact_request();
+    let key = DeviceIdentity::generate();
+    for (completion, expected_reason) in [
+        (TraceRecordCompletion::LimitExceeded, "COLLECTION_FAILED"),
+        (TraceRecordCompletion::SourceUnavailable, "SOURCE_UNAVAILABLE"),
+    ] {
+        let result = record_diagnostic_result(
+            &request,
+            rustfs::connect::TraceRecordCapture {
+                data: rustfs::connect::RecordedTrace {
+                    spans: vec![rustfs::connect::TelemetrySpan {
+                        operation: TelemetryOperation::GetObject,
+                        duration_micros: 500,
+                        status: TelemetrySpanStatus::Ok,
+                    }],
+                    dropped_span_count: 1,
+                },
+                completion,
+            },
+            Duration::from_millis(1),
+        );
+        let serialized = serde_json::to_value(&result).expect("failed diagnostic result");
+        assert_eq!(serialized["outcome"], "FAILED");
+        assert_eq!(serialized["coverage"]["requestedUnits"], 1);
+        assert_eq!(serialized["coverage"]["completedUnits"], 0);
+        assert_eq!(serialized["reasonCode"], expected_reason);
+        assert!(serialized["data"].is_null());
+        assert!(matches!(
+            encode_signed_telemetry_export(&request, &result, &key, &CancellationToken::new()),
+            Err(TelemetryArtifactError::InvalidRequest)
+        ));
+    }
+}
+
+#[test]
+#[serial]
+fn connect_trace_record_rejects_a_forged_partial_complete_result() {
+    let request = artifact_request();
+    let key = DeviceIdentity::generate();
+    let result = TelemetryDiagnosticResult::partial(
+        &request,
+        TelemetryTool::Record,
+        Duration::ZERO,
+        rustfs::connect::TelemetryReasonCode::Complete,
+        rustfs::connect::RecordedTrace {
+            spans: vec![],
+            dropped_span_count: 0,
+        },
+    );
+    assert!(matches!(
+        encode_signed_telemetry_export(&request, &result, &key, &CancellationToken::new()),
+        Err(TelemetryArtifactError::InvalidRequest)
+    ));
+}

rustfs/tests/connect_trace_replay.rs

@@ -0,0 +1,135 @@
+use serial_test::serial;
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
+
+use rustfs::connect::{
+    LocalTelemetryConsent, LocallyReviewedTraceArtifact, TelemetryArtifactConsent, TelemetryArtifactRequest, TelemetryProvenance,
+    TraceReplayError, analyze_trace, replay_trace, replay_trace_result,
+};
+use tokio_util::sync::CancellationToken;
+
+fn consent() -> LocalTelemetryConsent {
+    LocalTelemetryConsent::new(Instant::now() + Duration::from_secs(1)).expect("future consent")
+}
+
+fn artifact_request() -> TelemetryArtifactRequest {
+    let now = SystemTime::now().duration_since(UNIX_EPOCH).expect("current time").as_secs() as i64;
+    let organization = "organizations/019e3ae0-0000-7000-8000-000000000001";
+    let cluster = format!("{organization}/clusters/019e3ae0-0000-7000-8000-000000000002");
+    TelemetryArtifactRequest {
+        organization_name: organization.to_owned(),
+        cluster_name: cluster.clone(),
+        device_name: format!("{cluster}/clusterDevices/019e3ae0-0000-7000-8000-000000000003"),
+        run_uid: "019e3ae0-0000-7000-8000-000000000004".to_owned(),
+        artifact_uid: "019e3ae0-0000-7000-8000-000000000005".to_owned(),
+        schema_version: 1,
+        consent: TelemetryArtifactConsent {
+            consent_uid: "019e3ae0-0000-7000-8000-000000000006".to_owned(),
+            policy_revision: 1,
+            expires_at_unix: now + 120,
+            confirmed: true,
+        },
+        produced_at_unix: now,
+        expires_at_unix: now + 60,
+        nonce: [0x5a; 32],
+        provenance: TelemetryProvenance::new("a".repeat(40), "b".repeat(64), "1.0.0-rc.6", vec![]),
+    }
+}
+
+#[test]
+#[serial]
+fn connect_trace_replay_validates_and_digests_reviewed_bytes_without_file_io() {
+    let bytes = br#"{"spans":[{"operation":"GET_OBJECT","durationMicros":500,"status":"OK"}],"droppedSpanCount":0}"#;
+    let replay = replay_trace(
+        LocallyReviewedTraceArtifact::new(bytes).expect("bounded artifact"),
+        consent(),
+        &CancellationToken::new(),
+    )
+    .expect("valid record replays");
+
+    assert_eq!(replay.spans.len(), 1);
+    assert_eq!(
+        replay.input_artifact_sha256,
+        "1daadd3e226e7dd67a35dd56092ec985f464d832756e75277bc11d4a25c0291b"
+    );
+}
+
+#[test]
+#[serial]
+fn connect_trace_replay_wraps_reviewed_bytes_in_the_frozen_result_contract() {
+    let bytes = br#"{"spans":[{"operation":"GET_OBJECT","durationMicros":500,"status":"OK"}],"droppedSpanCount":0}"#;
+    let result = replay_trace_result(
+        &artifact_request(),
+        LocallyReviewedTraceArtifact::new(bytes).expect("bounded artifact"),
+        consent(),
+        &CancellationToken::new(),
+    )
+    .expect("typed replay result");
+    let analysis = analyze_trace(result.data().expect("replayed data"), consent(), &CancellationToken::new())
+        .expect("replayed trace is locally analyzable");
+    assert_eq!(analysis.span_count, 1);
+    assert_eq!(analysis.total_duration_micros, 500);
+    let json = serde_json::to_value(result).expect("result JSON");
+    assert_eq!(json["toolId"], "telemetry.replay");
+    assert_eq!(json["capability"], "telemetry.replay@1");
+    assert_eq!(json["outcome"], "SUCCEEDED");
+    assert_eq!(json["reasonCode"], "COMPLETE");
+    assert_eq!(json["coverage"]["requestedUnits"], 1);
+    assert_eq!(json["coverage"]["completedUnits"], 1);
+    assert_eq!(json["data"]["spans"][0]["operation"], "GET_OBJECT");
+}
+
+#[test]
+#[serial]
+fn connect_trace_replay_rejects_unknown_or_secret_bearing_fields() {
+    let bytes = br#"{"spans":[],"droppedSpanCount":0,"authorization":"Bearer secret"}"#;
+    let error = replay_trace(
+        LocallyReviewedTraceArtifact::new(bytes).expect("bounded artifact"),
+        consent(),
+        &CancellationToken::new(),
+    )
+    .expect_err("unknown field must fail");
+    assert_eq!(error, TraceReplayError::InvalidArtifact);
+}
+
+#[test]
+#[serial]
+fn connect_trace_replay_honors_stop_before_parsing() {
+    let cancel = CancellationToken::new();
+    cancel.cancel();
+    let error = replay_trace(
+        LocallyReviewedTraceArtifact::new(br#"{"spans":[],"droppedSpanCount":0}"#).expect("bounded artifact"),
+        consent(),
+        &cancel,
+    )
+    .expect_err("cancelled replay must fail");
+    assert_eq!(error, TraceReplayError::Cancelled);
+}
+
+#[test]
+#[serial]
+fn connect_trace_replay_rejects_duplicate_fields_and_safe_integer_overflow() {
+    for bytes in [
+        br#"{"spans":[],"spans":[],"droppedSpanCount":0}"#.as_slice(),
+        br#"{"spans":[{"operation":"GET_OBJECT","durationMicros":9007199254740992,"status":"OK"}],"droppedSpanCount":0}"#
+            .as_slice(),
+        br#"{"spans":[],"droppedSpanCount":9007199254740992}"#.as_slice(),
+    ] {
+        let error = replay_trace(
+            LocallyReviewedTraceArtifact::new(bytes).expect("bounded artifact"),
+            consent(),
+            &CancellationToken::new(),
+        )
+        .expect_err("invalid record must fail");
+        assert_eq!(error, TraceReplayError::InvalidArtifact);
+    }
+}
+
+#[test]
+#[serial]
+fn connect_trace_replay_enforces_exact_artifact_size_boundary() {
+    assert!(LocallyReviewedTraceArtifact::new(&vec![b' '; 262_144]).is_ok());
+    assert_eq!(
+        LocallyReviewedTraceArtifact::new(&vec![b' '; 262_145]).err(),
+        Some(TraceReplayError::InvalidArtifact)
+    );
+}