From aaa7ed042eaa1f8c6c03e41673bdf0efc7851c14 Mon Sep 17 00:00:00 2001 From: chess Date: Sat, 29 Aug 2026 20:10:18 +0000 Subject: [PATCH] Harden diagnostic report uploads (#41569) ## What changed - Send the core report event before its attachments, then upload each attachment in a separate gzip-compressed envelope linked to that event. - Bound encoded and decoded payload sizes, truncate oversized attachment copies with format-aware handling for JSONL, and preserve attachment priority when applying size limits. - Retry transient attachment failures within a shared upload-time budget while honoring Sentry rate-limit and retry headers without replaying the core event. - Add APIs to prepare and send persisted report events and attachments using a stable report ID, explicit delivery outcomes, and caller-controlled retries. ## Testing - Cover attachment retries, truncation, rate limits, rejected responses, transport failures, redirect handling, content types, diagnostic consent, and report tags. GitOrigin-RevId: 5eacd384843795d05d1c024b76e1c6d0c1c4b8d7 --- codex-rs/Cargo.lock | 4 + .../request_processors/feedback_processor.rs | 3 + codex-rs/feedback/Cargo.toml | 5 + .../feedback/src/attachment_truncation.rs | 41 ++ codex-rs/feedback/src/lib.rs | 696 +++++++++++++----- codex-rs/feedback/src/report_upload.rs | 155 ++++ codex-rs/feedback/src/upload.rs | 236 ++++++ 7 files changed, 951 insertions(+), 189 deletions(-) create mode 100644 codex-rs/feedback/src/attachment_truncation.rs create mode 100644 codex-rs/feedback/src/report_upload.rs create mode 100644 codex-rs/feedback/src/upload.rs diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index 38a1b295a9..ba42adc6c2 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3206,9 +3206,13 @@ name = "codex-feedback" version = "0.0.0" dependencies = [ "anyhow", + "bytes", "codex-http-client", "codex-login", "codex-protocol", + "flate2", + "http 1.4.0", + "httpdate", "log", "mime_guess", "pretty_assertions", diff --git a/codex-rs/app-server/src/request_processors/feedback_processor.rs b/codex-rs/app-server/src/request_processors/feedback_processor.rs index e6b8eef3a2..d2948c4ade 100644 --- a/codex-rs/app-server/src/request_processors/feedback_processor.rs +++ b/codex-rs/app-server/src/request_processors/feedback_processor.rs @@ -185,6 +185,9 @@ impl FeedbackRequestProcessor { let mut attachment_paths = Vec::new(); let mut seen_attachment_paths = HashSet::new(); + // File priority after logs and generated diagnostics: reported thread, subagent + // descendants (oldest to newest within the retained recent set), guardian rollout, + // sandbox log, tool caches, then caller files. Keep this order for size budgeting. if include_logs { for feedback_thread_id in &feedback_thread_ids { let Some(rollout_path) = self diff --git a/codex-rs/feedback/Cargo.toml b/codex-rs/feedback/Cargo.toml index 8a01696952..303cde11fe 100644 --- a/codex-rs/feedback/Cargo.toml +++ b/codex-rs/feedback/Cargo.toml @@ -9,11 +9,16 @@ workspace = true [dependencies] anyhow = { workspace = true } +bytes = { workspace = true } codex-http-client = { workspace = true } codex-login = { workspace = true } codex-protocol = { workspace = true } +flate2 = { workspace = true } +http = { workspace = true } +httpdate = { workspace = true } mime_guess = { workspace = true } sentry = { version = "0.46" } +tokio = { workspace = true, features = ["time"] } tracing = { workspace = true } tracing-subscriber = { workspace = true } diff --git a/codex-rs/feedback/src/attachment_truncation.rs b/codex-rs/feedback/src/attachment_truncation.rs new file mode 100644 index 0000000000..ef8fc3dd7c --- /dev/null +++ b/codex-rs/feedback/src/attachment_truncation.rs @@ -0,0 +1,41 @@ +use std::path::Path; + +use anyhow::Context; +use anyhow::Result; + +/// Shorten and label an attachment copy, leaving files within the target untouched. +/// JSONL keeps at least one complete record when a smaller record prefix is possible, +/// even if it exceeds the target. Callers must check the resulting upload size. +/// Other formats use a byte prefix. +pub(super) fn truncate_attachment( + filename: &mut String, + buffer: &mut Vec, + target_bytes: usize, +) -> Result<()> { + if buffer.len() <= target_bytes { + return Ok(()); + } + + let end = match Path::new(filename) + .extension() + .and_then(|extension| extension.to_str()) + { + Some(extension) if extension.eq_ignore_ascii_case("jsonl") => buffer[..target_bytes] + .iter() + .rposition(|byte| *byte == b'\n') + .or_else(|| { + buffer + .iter() + .position(|byte| *byte == b'\n') + .filter(|end| *end + 1 < buffer.len()) + }) + .map(|end| end + 1) + .context("feedback attachment cannot be shortened to a complete JSONL record")?, + _ => target_bytes, + }; + buffer.truncate(end); + if !filename.starts_with("truncated-") { + filename.insert_str(0, "truncated-"); + } + Ok(()) +} diff --git a/codex-rs/feedback/src/lib.rs b/codex-rs/feedback/src/lib.rs index dad90f56a6..34f7b01b1a 100644 --- a/codex-rs/feedback/src/lib.rs +++ b/codex-rs/feedback/src/lib.rs @@ -2,6 +2,7 @@ use std::collections::BTreeMap; use std::collections::VecDeque; use std::collections::btree_map::Entry; use std::fs; +use std::io::Read; use std::io::Write; use std::io::{self}; use std::path::Path; @@ -29,10 +30,16 @@ use tracing_subscriber::filter::Targets; use tracing_subscriber::fmt::writer::MakeWriter; use tracing_subscriber::registry::LookupSpan; +mod attachment_truncation; pub(crate) mod feedback_diagnostics; +mod report_upload; +mod upload; pub use feedback_diagnostics::FEEDBACK_DIAGNOSTICS_ATTACHMENT_FILENAME; pub use feedback_diagnostics::FeedbackDiagnostic; pub use feedback_diagnostics::FeedbackDiagnostics; +pub use report_upload::FeedbackDelivery; +pub use report_upload::FeedbackTransport; +pub use report_upload::prepare_report_attachment; /// Filename used for the redacted `codex doctor --json` feedback attachment. pub const DOCTOR_REPORT_ATTACHMENT_FILENAME: &str = "codex-doctor-report.json"; @@ -46,6 +53,15 @@ const DEFAULT_MAX_BYTES: usize = 4 * 1024 * 1024; // 4 MiB const SENTRY_DSN: &str = "https://ae32ed50620d7a7792c1ce5df38b3e3e@o33249.ingest.us.sentry.io/4510195390611458"; const UPLOAD_TIMEOUT_SECS: u64 = 10; +// Raw collection budgets used by the report API, not the interactive upload. +pub const MAX_ATTACHMENT_BYTES: usize = 64 * 1024 * 1024; +pub const MAX_ATTACHMENTS_BYTES: usize = 126 * 1024 * 1024; +// Check complete envelopes against Sentry's published limits, including framing: +// https://develop.sentry.dev/sdk/foundations/envelopes/#size-limits +// https://docs.sentry.io/platforms/javascript/enriching-events/attachments/ +const MAX_DECODED_UPLOAD_BYTES: usize = 200 * 1024 * 1024; +const MAX_EVENT_BYTES: usize = 1024 * 1024; +const MAX_UPLOAD_BYTES: usize = 40_000_000; const FEEDBACK_TAGS_TARGET: &str = "feedback_tags"; const MAX_FEEDBACK_TAGS: usize = 64; @@ -370,6 +386,64 @@ pub struct FeedbackAttachmentPath { pub attachment_filename_override: Option, } +enum AttachmentReadMode { + Whole, + Prefix, +} + +impl FeedbackAttachmentPath { + /// Read a whole regular file within the caller's size limit. + pub fn read_attachment(&self, max_bytes: usize) -> io::Result> { + self.read_attachment_with_mode(max_bytes, AttachmentReadMode::Whole) + } + + fn read_attachment_with_mode( + &self, + max_bytes: usize, + mode: AttachmentReadMode, + ) -> io::Result> { + let metadata = fs::metadata(&self.path)?; + if !metadata.is_file() + || (metadata.len() > max_bytes as u64 && matches!(mode, AttachmentReadMode::Whole)) + { + return Ok(None); + } + let mut buffer = Vec::new(); + // Keep one extra byte so the encoder can detect and label a truncated prefix, + // including when the file grows after the metadata check. + fs::File::open(&self.path)? + .take(max_bytes as u64 + 1) + .read_to_end(&mut buffer)?; + if buffer.len() > max_bytes && matches!(mode, AttachmentReadMode::Whole) { + return Ok(None); + } + let filename = self + .attachment_filename_override + .clone() + .unwrap_or_else(|| { + self.path + .file_name() + .map(|name| name.to_string_lossy().to_string()) + .unwrap_or_else(|| "extra-log.log".to_string()) + }); + let content_type = match Path::new(&filename) + .extension() + .and_then(|extension| extension.to_str()) + { + Some(extension) if extension.eq_ignore_ascii_case("jsonl") => "text/plain".to_string(), + _ => mime_guess::from_path(&filename) + .first_or_octet_stream() + .essence_str() + .to_string(), + }; + Ok(Some(FeedbackAttachment { + filename, + content_type: Some(content_type), + buffer, + })) + } +} + /// In-memory attachment to include in a feedback upload. /// /// Use this for generated diagnostics that should not be materialized on disk, @@ -407,6 +481,51 @@ pub struct FeedbackUploadOptions<'a> { } impl FeedbackSnapshot { + fn feedback_event( + &self, + classification: &str, + reason: Option<&str>, + tags: Option<&BTreeMap>, + session_source: Option<&SessionSource>, + ) -> sentry::protocol::Event<'static> { + use sentry::protocol::Event; + use sentry::protocol::Exception; + use sentry::protocol::Level; + use sentry::protocol::Values; + + let level = match classification { + "bug" | "bad_result" | "safety_check" => Level::Error, + _ => Level::Info, + }; + let title = format!( + "[{}]: Codex session {}", + display_classification(classification), + self.thread_id + ); + let mut event = Event { + level, + message: Some(title.clone()), + tags: self.upload_tags(classification, reason, tags, session_source), + ..Default::default() + }; + if let Some(reason) = reason { + event.exception = Values::from(vec![Exception { + ty: title, + value: Some(reason.to_string()), + ..Default::default() + }]); + } + event + } + + pub fn log_attachment(&self, logs_override: Option>) -> FeedbackAttachment { + FeedbackAttachment { + filename: "codex-logs.log".to_string(), + content_type: Some("text/plain".to_string()), + buffer: logs_override.unwrap_or_else(|| self.bytes.clone()), + } + } + pub fn feedback_diagnostics(&self) -> &FeedbackDiagnostics { &self.feedback_diagnostics } @@ -442,55 +561,48 @@ impl FeedbackSnapshot { ) -> Result<()> { use std::str::FromStr; - use sentry::ClientOptions; use sentry::protocol::Envelope; use sentry::protocol::EnvelopeItem; - use sentry::protocol::Event; - use sentry::protocol::Level; use sentry::types::Dsn; let started_at = Instant::now(); let dsn = Dsn::from_str(dsn).map_err(|error| anyhow!("invalid DSN: {error}"))?; - let upload_url = dsn.envelope_api_url(); - let sentry_options = ClientOptions::default(); - let sentry_auth = dsn.to_auth(Some(sentry_options.user_agent.as_ref())); - let tags = self.upload_tags( + let event = self.feedback_event( options.classification, options.reason, options.tags, options.session_source.as_ref(), ); - - let level = match options.classification { - "bug" | "bad_result" | "safety_check" => Level::Error, - _ => Level::Info, - }; - let mut envelope = Envelope::new(); - let title = format!( - "[{}]: Codex session {}", - display_classification(options.classification), - self.thread_id - ); - - let mut event = Event { - level, - message: Some(title.clone()), - tags, - ..Default::default() - }; - if let Some(r) = options.reason { - use sentry::protocol::Exception; - use sentry::protocol::Values; - - event.exception = Values::from(vec![Exception { - ty: title, - value: Some(r.to_string()), - ..Default::default() - }]); - } envelope.add_item(EnvelopeItem::Event(event)); + let (event_body, event_bytes) = + upload::gzip_envelope(&envelope).context("failed to serialize feedback event")?; + anyhow::ensure!( + event_bytes <= MAX_EVENT_BYTES, + "feedback event exceeds the size limit" + ); + let headers = envelope.headers().clone(); + drop(envelope); + let client_pool = RouteAwareClientPool::new_without_redirects_or_request_logging( + http_client_factory.clone(), + ClientRouteClass::Other, + ); + // Acknowledge the report before reading diagnostic files. Keep one network wait + // budget for the whole submission; file reads and compression do not consume it. + let mut remaining_upload_time = Duration::from_secs(UPLOAD_TIMEOUT_SECS); + let status = upload::send_gzip_envelope( + &client_pool, + &dsn, + event_body, + upload::EnvelopeKind::Event, + &mut remaining_upload_time, + ) + .await?; + anyhow::ensure!( + status.is_success(), + "Sentry rejected feedback upload with HTTP status {status}" + ); let attachments = self.feedback_attachments( options.include_logs, @@ -498,74 +610,50 @@ impl FeedbackSnapshot { options.extra_attachment_paths, options.logs_override, ); - let attachment_count = attachments.len(); + let mut uploaded_attachments = 0; + let mut attachments_failed = false; + // Keep attachments linked to the accepted event, without replaying its contents. for attachment in attachments { - envelope.add_item(EnvelopeItem::Attachment(attachment)); - } - - let mut body = Vec::new(); - envelope - .to_writer(&mut body) - .context("failed to serialize feedback upload")?; - - tracing::info!( - thread_id = %self.thread_id, - classification = options.classification, - include_logs = options.include_logs, - attachment_count, - payload_bytes = body.len(), - "uploading feedback to Sentry" - ); - - let mut status = None; - let result: Result<()> = async { - let client_pool = RouteAwareClientPool::new_without_redirects( - http_client_factory.clone(), - ClientRouteClass::Other, - ); - let response = client_pool - .post(upload_url.as_str()) - .header("X-Sentry-Auth", sentry_auth.to_string()) - .body(body) - .timeout(Duration::from_secs(UPLOAD_TIMEOUT_SECS)) - .send() - .await - .context("failed to upload feedback to Sentry")?; - let response_status = response.status(); - status = Some(response_status.as_u16()); - response - .error_for_status() - .context("failed to upload feedback to Sentry")?; - anyhow::ensure!( - response_status.is_success(), - "Sentry rejected feedback upload with HTTP status {response_status}" - ); - Ok(()) - } - .await; - - match &result { - Ok(()) => tracing::info!( - thread_id = %self.thread_id, - classification = options.classification, - include_logs = options.include_logs, - status, - elapsed_ms = started_at.elapsed().as_millis(), - "feedback uploaded to Sentry" - ), - Err(error) => { - tracing::warn!( - thread_id = %self.thread_id, - classification = options.classification, - include_logs = options.include_logs, - status, - elapsed_ms = started_at.elapsed().as_millis(), - error = %format!("{error:#}"), - "feedback upload failed" + if remaining_upload_time.is_zero() { + attachments_failed = true; + break; + } + let mut status = None; + let result: Result<()> = async { + let body = upload::gzip_attachment_envelope(&headers, attachment)?; + let response_status = upload::send_gzip_envelope( + &client_pool, + &dsn, + body, + upload::EnvelopeKind::Attachment, + &mut remaining_upload_time, ) + .await?; + status = Some(response_status.as_u16()); + anyhow::ensure!(response_status.is_success(), "Sentry rejected attachment"); + Ok(()) + } + .await; + if result.is_ok() { + uploaded_attachments += 1; + } else { + attachments_failed = true; + // Keep trying other diagnostics before reporting the partial failure. + tracing::warn!(status, "feedback attachment upload failed; continuing"); } } - result + tracing::info!( + thread_id = %self.thread_id, + uploaded_attachments, + attachments_failed, + elapsed_ms = started_at.elapsed().as_millis(), + "feedback event uploaded to Sentry" + ); + anyhow::ensure!( + !attachments_failed, + "feedback report was accepted, but some attachments failed to upload" + ); + Ok(()) } fn upload_tags( @@ -617,85 +705,63 @@ impl FeedbackSnapshot { tags } - fn feedback_attachments( - &self, + fn feedback_attachments<'a>( + &'a self, include_logs: bool, - extra_attachments: &[FeedbackAttachment], - extra_attachment_paths: &[FeedbackAttachmentPath], + extra_attachments: &'a [FeedbackAttachment], + extra_attachment_paths: &'a [FeedbackAttachmentPath], logs_override: Option>, - ) -> Vec { + ) -> impl Iterator + 'a { use sentry::protocol::Attachment; - let mut attachments = Vec::new(); - - if include_logs { - attachments.push(Attachment { - buffer: logs_override.unwrap_or_else(|| self.bytes.clone()), - filename: String::from("codex-logs.log"), - content_type: Some("text/plain".to_string()), - ty: None, - }); - } - - attachments.extend(extra_attachments.iter().map(|attachment| Attachment { - buffer: attachment.buffer.clone(), - filename: attachment.filename.clone(), - content_type: attachment.content_type.clone(), - ty: None, - })); - - if let Some(text) = self.feedback_diagnostics_attachment_text(include_logs) { - attachments.push(Attachment { + // Priority: logs, generated attachments (doctor report), connectivity diagnostics, + // then files in caller order. Read and compress each file independently; + // raw sizes across separate requests do not determine whether their gzip bodies fit. + let logs = include_logs.then(|| self.log_attachment(logs_override)); + let diagnostics = self + .feedback_diagnostics_attachment_text(include_logs) + .map(|text| FeedbackAttachment { buffer: text.into_bytes(), filename: FEEDBACK_DIAGNOSTICS_ATTACHMENT_FILENAME.to_string(), content_type: Some("text/plain".to_string()), - ty: None, }); - } - for attachment_path in extra_attachment_paths { - let data = match fs::read(&attachment_path.path) { - Ok(data) => data, - Err(err) => { - tracing::warn!( - path = %attachment_path.path.display(), - error = %err, - "failed to read log attachment; skipping" - ); - continue; + logs.into_iter() + .chain( + extra_attachments + .iter() + .map(|attachment| FeedbackAttachment { + buffer: attachment.buffer.clone(), + filename: attachment.filename.clone(), + content_type: attachment.content_type.clone(), + }), + ) + .chain(diagnostics) + .chain(extra_attachment_paths.iter().filter_map(|attachment_path| { + match attachment_path + .read_attachment_with_mode(MAX_DECODED_UPLOAD_BYTES, AttachmentReadMode::Prefix) + { + Ok(Some(attachment)) => Some(attachment), + Ok(None) => { + tracing::warn!("feedback attachment is not a regular file; skipping"); + None + } + Err(err) => { + tracing::warn!( + path = %attachment_path.path.display(), + error = %err, + "failed to read log attachment; skipping" + ); + None + } } - }; - let filename = attachment_path - .attachment_filename_override - .clone() - .unwrap_or_else(|| { - attachment_path - .path - .file_name() - .map(|s| s.to_string_lossy().to_string()) - .unwrap_or_else(|| "extra-log.log".to_string()) - }); - let content_type = match Path::new(&filename) - .extension() - .and_then(|extension| extension.to_str()) - { - Some(extension) if extension.eq_ignore_ascii_case("jsonl") => { - "text/plain".to_string() - } - _ => mime_guess::from_path(&filename) - .first_or_octet_stream() - .essence_str() - .to_string(), - }; - attachments.push(Attachment { - buffer: data, - filename, - content_type: Some(content_type), + })) + .map(|attachment| Attachment { + buffer: attachment.buffer, + filename: attachment.filename, + content_type: attachment.content_type, ty: None, - }); - } - - attachments + }) } } @@ -783,16 +849,21 @@ impl Visit for FeedbackTagsVisitor { mod tests { use std::ffi::OsStr; use std::fs; + use std::sync::atomic::AtomicUsize; + use std::sync::atomic::Ordering; use super::*; use crate::FeedbackDiagnostic; use codex_http_client::OutboundProxyPolicy; + use flate2::read::GzDecoder; + use http::StatusCode; use pretty_assertions::assert_eq; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; use wiremock::Mock; use wiremock::MockServer; use wiremock::ResponseTemplate; + use wiremock::matchers::header; use wiremock::matchers::header_exists; use wiremock::matchers::method; use wiremock::matchers::path; @@ -871,16 +942,21 @@ mod tests { pretty_assertions::assert_eq!(snap.tags.get("cached").map(String::as_str), Some("true")); } - async fn upload_test_feedback(feedback: &CodexFeedback, dsn: &str) -> Result<()> { + async fn upload_test_feedback( + feedback: &CodexFeedback, + dsn: &str, + extra_attachments: &[FeedbackAttachment], + ) -> Result<()> { feedback .snapshot(/*session_id*/ None) + .with_feedback_diagnostics(FeedbackDiagnostics::default()) .upload_feedback_with_dsn( FeedbackUploadOptions { classification: "bug", reason: Some("private feedback"), tags: None, include_logs: true, - extra_attachments: &[], + extra_attachments, extra_attachment_paths: &[], session_source: Some(SessionSource::Cli), logs_override: Some(b"private log contents".to_vec()), @@ -892,20 +968,258 @@ mod tests { } #[tokio::test] - async fn feedback_upload_waits_for_successful_sentry_response() { + async fn feedback_upload_retries_diagnostics_without_replaying_core() { let server = MockServer::start().await; + let attempt = AtomicUsize::default(); Mock::given(method("POST")) .and(path("/api/42/envelope/")) .and(header_exists("X-Sentry-Auth")) - .respond_with(ResponseTemplate::new(200)) - .expect(1) + .and(header("Content-Encoding", "gzip")) + .respond_with(move |_: &wiremock::Request| { + match attempt.fetch_add(/*val*/ 1, Ordering::SeqCst) { + // Accept the core once; accept logs only on their second attempt. + 0 => ResponseTemplate::new(StatusCode::OK).insert_header( + "X-Sentry-Rate-Limits", + "0:attachment:project,60:transaction:project", + ), + 2 => ResponseTemplate::new(StatusCode::OK).insert_header("Retry-After", "0"), + 6 => ResponseTemplate::new(StatusCode::OK) + .insert_header("X-Sentry-Rate-Limits", "60:transaction:project"), + 1 => ResponseTemplate::new(StatusCode::SERVICE_UNAVAILABLE) + .insert_header("Retry-After", "2"), + 5 => ResponseTemplate::new(StatusCode::SERVICE_UNAVAILABLE) + .insert_header("Retry-After", "8"), + _ => ResponseTemplate::new(StatusCode::SERVICE_UNAVAILABLE) + .insert_header("Retry-After", "0"), + } + }) + .expect(/*r*/ 7) .mount(&server) .await; let dsn = format!("http://public@{}/42", server.address()); - upload_test_feedback(&CodexFeedback::new(), &dsn) + let attachments = ["retry-exhausted.txt", "later.txt"].map(|filename| FeedbackAttachment { + filename: filename.to_string(), + content_type: Some("text/plain".to_string()), + buffer: filename.as_bytes().to_vec(), + }); + let started = Instant::now(); + upload_test_feedback(&CodexFeedback::new(), &dsn, &attachments) .await - .expect("successful Sentry response should complete feedback upload"); + .expect_err("legacy uploads report incomplete diagnostics"); + assert!(started.elapsed() >= Duration::from_secs(2)); + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests[1].body, requests[2].body, "retry exact gzip bytes"); + // Exhausting one diagnostic must not delay later.txt or replay earlier files. + assert!( + requests[3..6] + .iter() + .all(|request| request.body == requests[3].body) + ); + let mut decoded = String::new(); + GzDecoder::new(requests[1].body.as_slice()) + .read_to_string(&mut decoded) + .unwrap(); + assert!(decoded.ends_with("private log contents\n")); + } + + #[tokio::test] + async fn feedback_upload_delivers_whole_files_or_marked_prefixes() { + let suffix = ThreadId::new(); + let first_path = std::env::temp_dir().join(format!("feedback-first-{suffix}.jsonl")); + let second_path = std::env::temp_dir().join(format!("feedback-second-{suffix}.jsonl")); + let binary_path = std::env::temp_dir().join(format!("feedback-binary-{suffix}.bin")); + let pending_path = second_path.with_extension("pending"); + let block = "{\"type\":\"event_msg\",\"payload\":{\"type\":\"agent_message\",\"message\":\"diagnostic fixture 🍵\"}}\n" + .repeat(1024); + let block_counts = [65, 205].map(|mib| mib * 1024 * 1024 / block.len() + 1); + for (path, blocks) in [&first_path, &pending_path].into_iter().zip(block_counts) { + let mut file = fs::File::create(path).unwrap(); + for _ in 0..blocks { + file.write_all(block.as_bytes()).unwrap(); + } + } + // Repeats beyond gzip's 32 KiB window keep this fixture above the wire limit. + let record_block = (0..4096) + .flat_map(|_| sentry::types::Uuid::new_v4().into_bytes()) + .map(|byte| { + b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/" + [usize::from(byte % 64)] + }) + .collect::>(); + let mut file = fs::File::create(&binary_path).unwrap(); + // The first record fits compressed, but exceeds the encoder's initial + // half-file target. It must survive when this same file is sent as JSONL. + for mib in [32, 30] { + file.write_all(b"{\"message\":\"").unwrap(); + for _ in 0..mib * 1024 * 1024 / record_block.len() { + file.write_all(&record_block).unwrap(); + } + file.write_all(b"\"}\n").unwrap(); + } + drop(file); + + let server = MockServer::start().await; + let attempt = AtomicUsize::default(); + let ready_path = second_path.clone(); + Mock::given(method("POST")) + .and(path("/api/42/envelope/")) + .and(header("Content-Encoding", "gzip")) + .respond_with(move |_: &wiremock::Request| { + if attempt.fetch_add(/*val*/ 1, Ordering::SeqCst) == 1 { + // The next file becomes readable only after the first file arrives. + fs::rename(&pending_path, &ready_path).unwrap(); + } + ResponseTemplate::new(StatusCode::OK) + }) + .expect(/*r*/ 5) + .mount(&server) + .await; + + CodexFeedback::new() + .snapshot(/*session_id*/ None) + .upload_feedback_with_dsn( + FeedbackUploadOptions { + classification: "bug", + reason: Some("large diagnostic upload"), + tags: None, + include_logs: false, + extra_attachments: &[], + extra_attachment_paths: &[ + FeedbackAttachmentPath { + path: first_path.clone(), + attachment_filename_override: None, + }, + FeedbackAttachmentPath { + path: second_path.clone(), + attachment_filename_override: None, + }, + FeedbackAttachmentPath { + path: binary_path.clone(), + attachment_filename_override: None, + }, + FeedbackAttachmentPath { + path: binary_path.clone(), + attachment_filename_override: Some(format!( + "feedback-records-{suffix}.jsonl" + )), + }, + ], + session_source: Some(SessionSource::Cli), + logs_override: None, + }, + &HttpClientFactory::new(OutboundProxyPolicy::ReqwestDefault), + &format!("http://public@{}/42", server.address()), + ) + .await + .unwrap(); + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests.len(), 5); + let mut event = String::new(); + GzDecoder::new(requests[0].body.as_slice()) + .read_to_string(&mut event) + .unwrap(); + for (request, (filename, source_path)) in requests[1..].iter().zip([ + (format!("feedback-first-{suffix}.jsonl"), &first_path), + ( + format!("truncated-feedback-second-{suffix}.jsonl"), + &second_path, + ), + ( + format!("truncated-feedback-binary-{suffix}.bin"), + &binary_path, + ), + ( + format!("truncated-feedback-records-{suffix}.jsonl"), + &binary_path, + ), + ]) { + assert!(request.body.len() <= MAX_UPLOAD_BYTES); + let mut decoded = Vec::new(); + GzDecoder::new(request.body.as_slice()) + .read_to_end(&mut decoded) + .unwrap(); + assert!(decoded.len() <= MAX_DECODED_UPLOAD_BYTES); + let mut parts = decoded.splitn(3, |byte| *byte == b'\n'); + assert_eq!( + parts.next().unwrap(), + event.lines().next().unwrap().as_bytes() + ); + let header = std::str::from_utf8(parts.next().unwrap()).unwrap(); + assert!(header.contains(&format!("\"filename\":\"{filename}\""))); + let payload = parts.next().unwrap().strip_suffix(b"\n").unwrap(); + let mut source = fs::File::open(source_path).unwrap(); + let original_bytes = source.metadata().unwrap().len() as usize; + if filename.starts_with("truncated-") { + assert!(!payload.is_empty() && payload.len() < original_bytes); + } else { + assert_eq!(payload.len(), original_bytes); + } + if filename.ends_with(".jsonl") { + let text = std::str::from_utf8(payload).unwrap(); + assert!(text.ends_with('\n')); + } + let mut expected = [0; 64 * 1024]; + for chunk in payload.chunks(expected.len()) { + source.read_exact(&mut expected[..chunk.len()]).unwrap(); + assert_eq!(chunk, &expected[..chunk.len()]); + } + } + fs::remove_file(first_path).unwrap(); + fs::remove_file(second_path).unwrap(); + fs::remove_file(binary_path).unwrap(); + } + + #[tokio::test] + async fn feedback_upload_stops_diagnostics_for_sentry_rate_limits() { + for (accepted_before_limit, rate_limit) in [ + (1, ResponseTemplate::new(StatusCode::TOO_MANY_REQUESTS)), + ( + 1, + ResponseTemplate::new(StatusCode::TOO_MANY_REQUESTS) + .insert_header("X-Sentry-Rate-Limits", "60:attachment:project") + .insert_header("Retry-After", "0"), + ), + ( + 0, + ResponseTemplate::new(StatusCode::OK) + .insert_header("X-Sentry-Rate-Limits", "60:attachment:project"), + ), + ( + 1, + ResponseTemplate::new(StatusCode::OK) + .insert_header("X-Sentry-Rate-Limits", "60::project"), + ), + ( + 0, + ResponseTemplate::new(StatusCode::OK).insert_header("Retry-After", "60"), + ), + ] { + let server = MockServer::start().await; + let attempt = AtomicUsize::default(); + Mock::given(method("POST")) + .and(path("/api/42/envelope/")) + .respond_with(move |_: &wiremock::Request| { + if attempt.fetch_add(/*val*/ 1, Ordering::SeqCst) < accepted_before_limit { + ResponseTemplate::new(StatusCode::OK) + } else { + rate_limit.clone() + } + }) + .expect((accepted_before_limit + 1) as u64) + .mount(&server) + .await; + + let dsn = format!("http://public@{}/42", server.address()); + let later = FeedbackAttachment { + filename: "later.txt".to_string(), + content_type: None, + buffer: b"later diagnostic".to_vec(), + }; + upload_test_feedback(&CodexFeedback::new(), &dsn, &[later]) + .await + .expect_err("legacy uploads report rate-limited diagnostics"); + } } #[tokio::test] @@ -921,7 +1235,7 @@ mod tests { let feedback = CodexFeedback::new(); let dsn = format!("http://public@{}/42", server.address()); - let error = upload_test_feedback(&feedback, &dsn) + let error = upload_test_feedback(&feedback, &dsn, &[]) .await .expect_err("rejected Sentry responses must fail feedback uploads"); @@ -944,7 +1258,7 @@ mod tests { .mount(&sentry_server) .await; let dsn = format!("http://public@{}/42", sentry_server.address()); - let error = upload_test_feedback(&CodexFeedback::new(), &dsn) + let error = upload_test_feedback(&CodexFeedback::new(), &dsn, &[]) .await .expect_err("redirected feedback uploads must be rejected"); @@ -968,7 +1282,7 @@ mod tests { drop(listener); let dsn = format!("http://public@{address}/42"); - let error = upload_test_feedback(&CodexFeedback::new(), &dsn) + let error = upload_test_feedback(&CodexFeedback::new(), &dsn, &[]) .await .expect_err("transport failures must fail feedback uploads"); @@ -997,16 +1311,18 @@ mod tests { details: vec!["HTTPS_PROXY = https://example.com:443".to_string()], }])); - let attachments_with_diagnostics = snapshot_with_diagnostics.feedback_attachments( - /*include_logs*/ true, - &[FeedbackAttachment { - filename: DOCTOR_REPORT_ATTACHMENT_FILENAME.to_string(), - content_type: Some("application/json".to_string()), - buffer: b"{\"overallStatus\":\"ok\"}".to_vec(), - }], - std::slice::from_ref(&extra_attachment_path), - Some(vec![1]), - ); + let attachments_with_diagnostics = snapshot_with_diagnostics + .feedback_attachments( + /*include_logs*/ true, + &[FeedbackAttachment { + filename: DOCTOR_REPORT_ATTACHMENT_FILENAME.to_string(), + content_type: Some("application/json".to_string()), + buffer: b"{\"overallStatus\":\"ok\"}".to_vec(), + }], + &[extra_attachment_path], + Some(vec![1]), + ) + .collect::>(); assert_eq!( attachments_with_diagnostics @@ -1041,7 +1357,8 @@ mod tests { let attachments_without_diagnostics = CodexFeedback::new() .snapshot(/*session_id*/ None) .with_feedback_diagnostics(FeedbackDiagnostics::default()) - .feedback_attachments(/*include_logs*/ true, &[], &[], Some(vec![1])); + .feedback_attachments(/*include_logs*/ true, &[], &[], Some(vec![1])) + .collect::>(); assert_eq!( attachments_without_diagnostics @@ -1082,7 +1399,8 @@ mod tests { }, ], /*logs_override*/ None, - ); + ) + .collect::>(); fs::remove_file(gzip_path).expect("gzip attachment should be removed"); fs::remove_file(unknown_path).expect("unknown attachment should be removed"); diff --git a/codex-rs/feedback/src/report_upload.rs b/codex-rs/feedback/src/report_upload.rs new file mode 100644 index 0000000000..1410832c7e --- /dev/null +++ b/codex-rs/feedback/src/report_upload.rs @@ -0,0 +1,155 @@ +use std::collections::BTreeMap; +use std::time::Duration; + +use anyhow::Context; +use anyhow::Result; +use anyhow::anyhow; +use codex_http_client::ClientRouteClass; +use codex_http_client::HttpClientFactory; +use codex_http_client::RouteAwareClientPool; +use codex_protocol::protocol::SessionSource; +use http::StatusCode; +use sentry::protocol::Attachment; +use sentry::protocol::Envelope; +use sentry::protocol::EnvelopeHeaders; +use sentry::protocol::EnvelopeItem; +use sentry::types::Dsn; +use sentry::types::Uuid; + +use crate::FeedbackAttachment; +use crate::FeedbackSnapshot; +use crate::MAX_DECODED_UPLOAD_BYTES; +use crate::MAX_EVENT_BYTES; +use crate::MAX_UPLOAD_BYTES; +use crate::SENTRY_DSN; +use crate::upload; + +// Background parts must tolerate slow links without inheriting the interactive +// upload's ten-second budget. The report API can cancel the request at any time. +const REPORT_PART_UPLOAD_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 300); + +/// The upstream HTTP result, not proof of durable storage in Sentry. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum FeedbackDelivery { + Accepted { + retry_after: Option, + }, + Rejected { + status: u16, + retry_after: Option, + }, + /// No response was received. The upstream may still have accepted the envelope. + Unconfirmed, +} + +/// Sends persisted report parts without retries or request diagnostics. +pub struct FeedbackTransport { + client_pool: RouteAwareClientPool, + dsn: Dsn, +} + +impl FeedbackTransport { + pub fn new(http_client_factory: HttpClientFactory) -> Result { + Ok(Self { + client_pool: RouteAwareClientPool::new_without_redirects_or_request_logging( + http_client_factory, + ClientRouteClass::Other, + ), + dsn: SENTRY_DSN + .parse() + .map_err(|_| anyhow!("invalid feedback DSN"))?, + }) + } + + /// Make one attempt. Callers own retry decisions and must retain the same bytes. + pub async fn send(&self, envelope: Vec) -> FeedbackDelivery { + let response = upload::gzip_envelope_request( + &self.client_pool, + &self.dsn, + envelope.into(), + REPORT_PART_UPLOAD_TIMEOUT, + ) + .send() + .await; + let Ok(response) = response else { + return FeedbackDelivery::Unconfirmed; + }; + let status = response.status(); + let headers = response.headers(); + let mut retry_after = headers + .get_all("Retry-After") + .iter() + .map(|value| { + upload::parse_retry_after(value.to_str().unwrap_or_default()) + .unwrap_or(upload::DEFAULT_RATE_LIMIT) + }) + .max(); + // Report events and attachments share one conservative cooldown. Honor quotas on + // successful responses too; ignore quotas for unrelated Sentry categories. + // https://develop.sentry.dev/sdk/foundations/transport/rate-limiting/ + retry_after = retry_after.max(upload::sentry_rate_limit_delay(headers)); + if status == StatusCode::TOO_MANY_REQUESTS && retry_after.is_none() { + retry_after = Some(upload::DEFAULT_RATE_LIMIT); + } + if status.is_success() { + FeedbackDelivery::Accepted { retry_after } + } else { + FeedbackDelivery::Rejected { + status: status.as_u16(), + retry_after, + } + } + } +} + +impl FeedbackSnapshot { + /// Prepare only the core report, using the caller's stable UUID as its Sentry event ID. + pub fn prepare_report_event( + &self, + report_id: &str, + classification: &str, + reason: Option<&str>, + tags: Option<&BTreeMap>, + session_source: Option<&SessionSource>, + ) -> Result> { + let mut event = self.feedback_event(classification, reason, tags, session_source); + // Auth tags must come from this report's caller, not metadata retained + // from an earlier account in the process-wide tracing layer. + event.tags.retain(|key, _| { + !matches!(key.as_str(), "account_id" | "chatgpt_user_id") + || tags.is_some_and(|tags| tags.contains_key(key)) + }); + event.event_id = + Uuid::parse_str(report_id).map_err(|_| anyhow!("invalid feedback report ID"))?; + let mut envelope = Envelope::new(); + envelope.add_item(EnvelopeItem::Event(event)); + prepare_envelope(envelope, MAX_EVENT_BYTES) + } +} + +/// Prepare one whole attachment linked to an existing report, without replaying its event. +/// The caller must apply consent and the report's cumulative attachment budget first. +pub fn prepare_report_attachment( + report_id: &str, + attachment: FeedbackAttachment, +) -> Result> { + let event_id = Uuid::parse_str(report_id).map_err(|_| anyhow!("invalid feedback report ID"))?; + let mut envelope = Envelope::new().with_headers(EnvelopeHeaders::new().with_event_id(event_id)); + envelope.add_item(EnvelopeItem::Attachment(Attachment { + buffer: attachment.buffer, + filename: attachment.filename, + content_type: attachment.content_type, + ty: None, + })); + prepare_envelope(envelope, MAX_DECODED_UPLOAD_BYTES) +} + +fn prepare_envelope(envelope: Envelope, max_decoded_bytes: usize) -> Result> { + let (bytes, decoded_bytes) = + upload::gzip_envelope(&envelope).context("failed to serialize feedback envelope")?; + anyhow::ensure!( + bytes.len() <= MAX_UPLOAD_BYTES && decoded_bytes <= max_decoded_bytes, + "feedback envelope exceeds the size limit" + ); + Ok(bytes) +} diff --git a/codex-rs/feedback/src/upload.rs b/codex-rs/feedback/src/upload.rs new file mode 100644 index 0000000000..c617e84a5b --- /dev/null +++ b/codex-rs/feedback/src/upload.rs @@ -0,0 +1,236 @@ +use std::io; +use std::io::Write; +use std::time::Duration; +use std::time::Instant; +use std::time::SystemTime; + +use anyhow::Context; +use anyhow::Result; +use bytes::Bytes; +use codex_http_client::RouteAwareClientPool; +use codex_http_client::RouteAwareRequestBuilder; +use flate2::Compression; +use flate2::write::GzEncoder; +use http::HeaderMap; +use http::StatusCode; +use sentry::ClientOptions; +use sentry::protocol::Attachment; +use sentry::protocol::Envelope; +use sentry::protocol::EnvelopeHeaders; +use sentry::types::Dsn; + +use crate::MAX_DECODED_UPLOAD_BYTES; +use crate::MAX_UPLOAD_BYTES; +use crate::attachment_truncation::truncate_attachment; + +pub(super) const DEFAULT_RATE_LIMIT: Duration = Duration::from_secs(/*secs*/ 60); + +pub(super) enum EnvelopeKind { + Event, + Attachment, +} + +pub(super) fn gzip_envelope_request( + client_pool: &RouteAwareClientPool, + dsn: &Dsn, + body: Bytes, + timeout: Duration, +) -> RouteAwareRequestBuilder { + let sentry_options = ClientOptions::default(); + let sentry_auth = dsn.to_auth(Some(sentry_options.user_agent.as_ref())); + client_pool + .post(dsn.envelope_api_url().as_str()) + .header("X-Sentry-Auth", sentry_auth.to_string()) + .header("Content-Encoding", "gzip") + .body(body) + .timeout(timeout) +} + +/// Send an already-serialized envelope, charging requests and backoff to the shared budget. +pub(super) async fn send_gzip_envelope( + client_pool: &RouteAwareClientPool, + dsn: &Dsn, + body: Vec, + kind: EnvelopeKind, + remaining_upload_time: &mut Duration, +) -> Result { + let request_started_at = Instant::now(); + let body = Bytes::from(body); + // Retain the exact gzip bytes for at most two diagnostic retries. Never replay the core. + let mut retry_delays = [250, 500].map(Duration::from_millis).into_iter(); + let response = loop { + let timeout = remaining_upload_time.saturating_sub(request_started_at.elapsed()); + if timeout.is_zero() { + *remaining_upload_time = Duration::ZERO; + anyhow::bail!("feedback upload network wait budget exhausted"); + } + let response = gzip_envelope_request(client_pool, dsn, body.clone(), timeout) + .send() + .await; + if response.as_ref().is_ok_and(|response| { + sentry_rate_limit_delay(response.headers()).is_some_and(|delay| !delay.is_zero()) + || (response.status().is_success() + && response.headers().get("Retry-After").is_some_and(|value| { + parse_retry_after(value.to_str().unwrap_or_default()) + .is_none_or(|delay| !delay.is_zero()) + })) + }) { + // Even accepted requests can impose a quota on events or attachments. + // https://develop.sentry.dev/sdk/foundations/transport/rate-limiting/ + *remaining_upload_time = Duration::ZERO; + break response; + } + let retryable = match &response { + Ok(response) => matches!(response.status().as_u16(), 408 | 429 | 500..=599), + Err(error) => { + error.is_timeout() + || (error.failure_class().is_none() + && (error.is_connect() || error.is_request() || error.is_body())) + } + }; + if matches!(kind, EnvelopeKind::Event) || !retryable { + break response; + } + let retry_delay = retry_delays.next(); + if response.as_ref().is_ok_and(|response| { + response.status() == StatusCode::TOO_MANY_REQUESTS + && (retry_delay.is_none() || !response.headers().contains_key("Retry-After")) + }) { + // Bare 429s default to a cooldown longer than this upload budget. + // Never let later files bypass an exhausted rate limit. + *remaining_upload_time = Duration::ZERO; + break response; + } + let Some(retry_delay) = retry_delay else { + break response; + }; + let retry_after = response + .as_ref() + .ok() + .and_then(|response| response.headers().get("Retry-After")) + .map(|value| { + // An invalid cooldown is not permission to send immediately. + parse_retry_after(value.to_str().unwrap_or_default()) + .unwrap_or(*remaining_upload_time) + }); + let delay = retry_delay.max(retry_after.unwrap_or_default()); + if delay >= remaining_upload_time.saturating_sub(request_started_at.elapsed()) { + // The cooldown applies to later files too, not just this retry. + *remaining_upload_time = Duration::ZERO; + break response; + } + tokio::time::sleep(delay).await; + if request_started_at.elapsed() >= *remaining_upload_time { + break response; + } + }; + *remaining_upload_time = remaining_upload_time.saturating_sub(request_started_at.elapsed()); + response + .map(|response| response.status()) + .context("failed to upload feedback to Sentry") +} + +pub(super) fn sentry_rate_limit_delay(headers: &HeaderMap) -> Option { + let mut retry_after = None; + for value in headers.get_all("X-Sentry-Rate-Limits") { + for quota in value.to_str().unwrap_or_default().split(',') { + let mut fields = quota.trim().split(':'); + let seconds = fields.next().unwrap_or_default(); + let categories = fields.next().unwrap_or_default(); + if !categories.is_empty() + && !categories + .split(';') + .any(|category| matches!(category, "error" | "attachment")) + { + continue; + } + let delay = seconds + .parse::() + .ok() + .and_then(|seconds| Duration::try_from_secs_f64(seconds.ceil()).ok()) + .unwrap_or(DEFAULT_RATE_LIMIT); + retry_after = retry_after.max(Some(delay)); + } + } + retry_after +} + +pub(super) fn parse_retry_after(value: &str) -> Option { + value + .parse::() + .map(Duration::from_secs) + .or_else(|_| { + httpdate::parse_http_date(value) + .map(|date| date.duration_since(SystemTime::now()).unwrap_or_default()) + }) + .ok() +} + +pub(super) fn gzip_envelope(envelope: &Envelope) -> io::Result<(Vec, usize)> { + let mut writer = CountingWriter { + inner: GzEncoder::new(Vec::new(), Compression::fast()), + bytes: 0, + }; + envelope.to_writer(&mut writer)?; + Ok((writer.inner.finish()?, writer.bytes)) +} + +pub(super) fn gzip_attachment_envelope( + headers: &EnvelopeHeaders, + mut attachment: Attachment, +) -> Result> { + let envelope = Envelope::new().with_headers(headers.clone()); + // Apply the file reader's bound to in-memory attachments too. The extra byte + // lets the format-aware truncation below detect and label a shortened copy. + attachment.buffer.truncate(MAX_DECODED_UPLOAD_BYTES + 1); + let mut target_bytes = MAX_DECODED_UPLOAD_BYTES; + loop { + truncate_attachment( + &mut attachment.filename, + &mut attachment.buffer, + target_bytes, + )?; + let mut writer = CountingWriter { + inner: GzEncoder::new(Vec::new(), Compression::fast()), + bytes: 0, + }; + // Use the SDK's framing while retaining the buffer for in-place truncation. + // https://github.com/getsentry/sentry-rust/blob/0.46.1/sentry-types/src/protocol/envelope.rs#L453-L487 + envelope.to_writer(&mut writer)?; + attachment.to_writer(&mut writer)?; + writeln!(writer)?; + let decoded_bytes = writer.bytes; + let body = writer.inner.finish()?; + if body.len() <= MAX_UPLOAD_BYTES && decoded_bytes <= MAX_DECODED_UPLOAD_BYTES { + return Ok(body); + } + anyhow::ensure!( + !attachment.buffer.is_empty(), + "feedback attachment headers exceed the size limit" + ); + target_bytes = attachment + .buffer + .len() + .saturating_sub(decoded_bytes.saturating_sub(MAX_DECODED_UPLOAD_BYTES)); + if body.len() > MAX_UPLOAD_BYTES { + target_bytes /= 2; + } + } +} + +struct CountingWriter { + inner: W, + bytes: usize, +} + +impl Write for CountingWriter { + fn write(&mut self, buf: &[u8]) -> io::Result { + let written = self.inner.write(buf)?; + self.bytes += written; + Ok(written) + } + + fn flush(&mut self) -> io::Result<()> { + self.inner.flush() + } +}