openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-03 19:53:25 +00:00
Code Issues Packages Projects Releases Wiki Activity

Harden diagnostic report uploads (#41569)

Browse Source 
## 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
This commit is contained in:
chess
2026-08-29 20:26:42 +00:00
committed by copyberry
parent f5636bb733
commit aaa7ed042e
7 changed files with 951 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+4
View File
@@ -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",
codex-rs/app-server/src/request_processors/feedback_processor.rs
+3
View File
@@ -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
codex-rs/feedback/Cargo.toml
+5
View File
@@ -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 }
codex-rs/feedback/src/attachment_truncation.rs
+41
View File
@@ -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<u8>,
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(())
}
codex-rs/feedback/src/lib.rs
+507 -189
View File
@@ -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<String>,
}
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<Option<FeedbackAttachment>> {
self.read_attachment_with_mode(max_bytes, AttachmentReadMode::Whole)
}
fn read_attachment_with_mode(
&self,
max_bytes: usize,
mode: AttachmentReadMode,
) -> io::Result<Option<FeedbackAttachment>> {
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<String, String>>,
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<Vec<u8>>) -> 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<u8>>,
) -> Vec<sentry::protocol::Attachment> {
) -> impl Iterator<Item = sentry::protocol::Attachment> + '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::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>();
assert_eq!(
attachments_without_diagnostics
@@ -1082,7 +1399,8 @@ mod tests {
},
],
/*logs_override*/ None,
);
)
.collect::<Vec<_>>();
fs::remove_file(gzip_path).expect("gzip attachment should be removed");
fs::remove_file(unknown_path).expect("unknown attachment should be removed");
codex-rs/feedback/src/report_upload.rs
+155
View File
@@ -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<Duration>,
},
Rejected {
status: u16,
retry_after: Option<Duration>,
},
/// 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<Self> {
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<u8>) -> 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<String, String>>,
session_source: Option<&SessionSource>,
) -> Result<Vec<u8>> {
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<Vec<u8>> {
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<Vec<u8>> {
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)
}
codex-rs/feedback/src/upload.rs
+236
View File
@@ -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<u8>,
kind: EnvelopeKind,
remaining_upload_time: &mut Duration,
) -> Result<StatusCode> {
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<Duration> {
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::<f64>()
.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<Duration> {
value
.parse::<u64>()
.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<u8>, 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<Vec<u8>> {
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<W> {
inner: W,
bytes: usize,
}
impl<W: Write> Write for CountingWriter<W> {
fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
let written = self.inner.write(buf)?;
self.bytes += written;
Ok(written)
}
fn flush(&mut self) -> io::Result<()> {
self.inner.flush()
}
}