openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-19 17:28:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bound exec-server process event reordering (#31576)

Browse Source 
## Why

A malicious exec-server can send out-of-order process notifications
faster than missing events arrive. The client retained every future
event in a per-session reorder map, so many tiny events or a few large
output chunks could grow orchestrator memory without bound.

## What changed

- cap each session's pending process-event reorder state at 256 events
and 1 MiB
- reject individual process events larger than 1 MiB
- release byte accounting as events publish or a session fails
- fail and detach only the affected process session when a limit is
exceeded
- let the next expected event drain a full buffer, so the limits apply
to retained future state
- apply the same bounded insertion path during reconnect recovery while
safely handling dense tail output, missing exit events, newer live
notifications, and conflicting `Closed` sequences

Sequence distance is intentionally not capped because it does not affect
allocation; the event-count and byte limits are the resource-exhaustion
boundary.

## Tests

Focused exec-server tests cover oversized output, full count/byte
buffers, gap-closing delivery, dense recovery interleaved with newer
notifications, missing exit reconstruction, and conflicting terminal
sequences.

## Scope

This is limited to per-session process-event reorder state. It does not
introduce a scheduler, quota framework, or new public configuration.
This commit is contained in:
jif
2026-07-09 09:56:14 +01:00
committed by GitHub
parent e8eb60092a
commit e398a99edf
3 changed files with 302 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/exec-server/src/client.rs
+86 -17
View File
@@ -113,6 +113,8 @@ const INITIALIZE_TIMEOUT: Duration = Duration::from_secs(10);
const ENVIRONMENT_INFO_TIMEOUT: Duration = Duration::from_secs(30);
const PROCESS_EVENT_CHANNEL_CAPACITY: usize = 256;
const PROCESS_EVENT_RETAINED_BYTES: usize = 1024 * 1024;
const MAX_PENDING_PROCESS_EVENTS: usize = 256;
const MAX_PENDING_PROCESS_EVENT_BYTES: usize = 1024 * 1024;
impl Default for ExecServerClientConnectOptions {
fn default() -> Self {
@@ -173,6 +175,7 @@ struct OrderedSessionEvents {
// different tasks and can reach the client out of order. Keep future events
// here until all lower sequence numbers have been published.
pending: BTreeMap<u64, ExecProcessEvent>,
pending_bytes: usize,
failure: Option<String>,
}
@@ -881,10 +884,10 @@ impl SessionState {
/// `Closed` event. The caller uses that signal to remove the session route
/// after the terminal event is visible to subscribers, rather than when a
/// possibly-early closed notification first arrives.
fn publish_ordered_event(&self, event: ExecProcessEvent) -> bool {
fn publish_ordered_event(&self, event: ExecProcessEvent) -> Result<bool, String> {
let Some(seq) = event.seq() else {
self.events.publish(event);
return false;
return Ok(false);
};
let mut ordered_events = self
@@ -897,11 +900,11 @@ impl SessionState {
|| ordered_events.closed_published
|| seq <= ordered_events.last_published_seq
{
return false;
return Ok(false);
}
ordered_events.pending.entry(seq).or_insert(event);
self.publish_ready(&mut ordered_events)
ordered_events.insert_pending(event)?;
Ok(self.publish_ready(&mut ordered_events))
}
fn publish_ready(&self, ordered_events: &mut OrderedSessionEvents) -> bool {
@@ -911,6 +914,9 @@ impl SessionState {
let Some(event) = ordered_events.pending.remove(&next_seq) else {
break;
};
ordered_events.pending_bytes = ordered_events
.pending_bytes
.saturating_sub(pending_process_event_bytes(&event));
ordered_events.last_published_seq = next_seq;
ordered_events.exit_published |= matches!(&event, ExecProcessEvent::Exited { .. });
let is_closed = matches!(&event, ExecProcessEvent::Closed { .. });
@@ -931,6 +937,7 @@ impl SessionState {
}
ordered_events.failure = Some(message.clone());
ordered_events.pending.clear();
ordered_events.pending_bytes = 0;
self.events.publish(ExecProcessEvent::Failed(message));
drop(ordered_events);
self.wake_tx
@@ -966,6 +973,68 @@ impl SessionState {
}
}
impl OrderedSessionEvents {
fn insert_pending(&mut self, event: ExecProcessEvent) -> Result<(), String> {
let Some(seq) = event.seq() else {
return Err("cannot reorder an unsequenced process event".to_string());
};
if self.pending.contains_key(&seq) {
return Ok(());
}
let next_seq = self.last_published_seq.saturating_add(1);
// The next expected event is synchronously published by every caller,
// so it can drain a full buffer without becoming retained state.
let closes_gap = seq == next_seq;
if !closes_gap && self.pending.len() >= MAX_PENDING_PROCESS_EVENTS {
return Err(format!(
"process event reorder buffer exceeds {MAX_PENDING_PROCESS_EVENTS} entries"
));
}
let event_bytes = pending_process_event_bytes(&event);
if event_bytes > MAX_PENDING_PROCESS_EVENT_BYTES {
return Err(format!(
"process event exceeds {MAX_PENDING_PROCESS_EVENT_BYTES} bytes"
));
}
let pending_bytes = self.pending_bytes.saturating_add(event_bytes);
if !closes_gap && pending_bytes > MAX_PENDING_PROCESS_EVENT_BYTES {
return Err(format!(
"process event reorder buffer exceeds {MAX_PENDING_PROCESS_EVENT_BYTES} bytes"
));
}
self.pending.insert(seq, event);
self.pending_bytes = pending_bytes;
Ok(())
}
}
fn pending_process_event_bytes(event: &ExecProcessEvent) -> usize {
match event {
ExecProcessEvent::Output(chunk) => chunk.chunk.0.len(),
ExecProcessEvent::Failed(message) => message.len(),
ExecProcessEvent::Exited { .. } | ExecProcessEvent::Closed { .. } => 0,
}
}
fn finish_process_event(
inner: &Inner,
process_id: &ProcessId,
session: &Arc<SessionState>,
result: Result<bool, String>,
) {
match result {
Ok(true) => inner.remove_session_if(process_id, session),
Ok(false) => {}
Err(message) => {
session.set_failure(message);
inner.remove_session_if(process_id, session);
}
}
}
impl Session {
pub(crate) fn process_id(&self) -> &ProcessId {
&self.process_id
@@ -1161,46 +1230,46 @@ async fn handle_server_notification(
let params: ExecOutputDeltaNotification =
serde_json::from_value(notification.params.unwrap_or(Value::Null))?;
if let Some(session) = inner.get_session(&params.process_id) {
session.note_change(params.seq);
let published_closed =
let result =
session.publish_ordered_event(ExecProcessEvent::Output(ProcessOutputChunk {
seq: params.seq,
stream: params.stream,
chunk: params.chunk,
}));
if published_closed {
inner.remove_session_if(&params.process_id, &session);
if result.is_ok() {
session.note_change(params.seq);
}
finish_process_event(inner, &params.process_id, &session, result);
}
}
EXEC_EXITED_METHOD => {
let params: ExecExitedNotification =
serde_json::from_value(notification.params.unwrap_or(Value::Null))?;
if let Some(session) = inner.get_session(&params.process_id) {
session.note_change(params.seq);
let published_closed = session.publish_ordered_event(ExecProcessEvent::Exited {
let result = session.publish_ordered_event(ExecProcessEvent::Exited {
seq: params.seq,
exit_code: params.exit_code,
sandbox_denied: params.sandbox_denied,
});
if published_closed {
inner.remove_session_if(&params.process_id, &session);
if result.is_ok() {
session.note_change(params.seq);
}
finish_process_event(inner, &params.process_id, &session, result);
}
}
EXEC_CLOSED_METHOD => {
let params: ExecClosedNotification =
serde_json::from_value(notification.params.unwrap_or(Value::Null))?;
if let Some(session) = inner.get_session(&params.process_id) {
session.note_change(params.seq);
// Closed is terminal, but it can arrive before tail output or
// exited. Keep routing this process until the ordered publisher
// says Closed has actually been delivered.
let published_closed =
let result =
session.publish_ordered_event(ExecProcessEvent::Closed { seq: params.seq });
if published_closed {
inner.remove_session_if(&params.process_id, &session);
if result.is_ok() {
session.note_change(params.seq);
}
finish_process_event(inner, &params.process_id, &session, result);
}
}
HTTP_REQUEST_BODY_DELTA_METHOD => {
codex-rs/exec-server/src/client_recovery.rs
+71 -36
View File
@@ -78,30 +78,6 @@ impl SessionState {
{
return Ok(false);
}
for chunk in chunks {
if chunk.seq > ordered_events.last_published_seq {
ordered_events
.pending
.entry(chunk.seq)
.or_insert(ExecProcessEvent::Output(chunk));
}
}
if closed {
match ordered_events.pending.get(&target_seq) {
Some(ExecProcessEvent::Closed { .. }) => {}
Some(_) => {
return Err(ExecServerError::Protocol(format!(
"process close sequence {target_seq} conflicts with recovered output"
)));
}
None => {
ordered_events
.pending
.insert(target_seq, ExecProcessEvent::Closed { seq: target_seq });
}
}
}
let pending_exit = ordered_events.pending.range_mut(..=target_seq).find_map(
|(_, event)| match event {
ExecProcessEvent::Exited {
@@ -116,12 +92,71 @@ impl SessionState {
*pending_sandbox_denied =
Some(pending_sandbox_denied.unwrap_or(false) || sandbox_denied);
}
let exit_known = ordered_events.exit_published || exit_pending;
let event_count = target_seq - ordered_events.last_published_seq;
let retained_count = ordered_events
.pending
.range(ordered_events.last_published_seq.saturating_add(1)..=target_seq)
.count() as u64;
let mut exit_known = ordered_events.exit_published || exit_pending;
if closed
&& (matches!(
ordered_events.pending.get(&target_seq),
Some(event) if !matches!(event, ExecProcessEvent::Closed { .. })
) || chunks.iter().any(|chunk| chunk.seq == target_seq))
{
return Err(ExecServerError::Protocol(format!(
"process close sequence {target_seq} conflicts with recovered output"
)));
}
let mut published_closed = false;
for chunk in chunks {
if chunk.seq > target_seq {
return Err(ExecServerError::Protocol(format!(
"recovered process output sequence {} exceeds target sequence {target_seq}",
chunk.seq
)));
}
let next_seq = ordered_events.last_published_seq.saturating_add(1);
if exited && !exit_known && chunk.seq > next_seq {
let exit_code = exit_code.ok_or_else(|| {
ExecServerError::Protocol(
"recovering exited process did not include its exit code".to_string(),
)
})?;
ordered_events
.insert_pending(ExecProcessEvent::Exited {
seq: next_seq,
exit_code,
sandbox_denied: Some(sandbox_denied),
})
.map_err(ExecServerError::Protocol)?;
published_closed |= self.publish_ready(&mut ordered_events);
exit_known = true;
}
if chunk.seq > ordered_events.last_published_seq {
ordered_events
.insert_pending(ExecProcessEvent::Output(chunk))
.map_err(ExecServerError::Protocol)?;
published_closed |= self.publish_ready(&mut ordered_events);
}
}
if closed
&& !ordered_events.closed_published
&& !matches!(
ordered_events.pending.get(&target_seq),
Some(ExecProcessEvent::Closed { .. })
)
{
ordered_events
.insert_pending(ExecProcessEvent::Closed { seq: target_seq })
.map_err(ExecServerError::Protocol)?;
}
let event_count = target_seq.saturating_sub(ordered_events.last_published_seq);
let first_unpublished_seq = ordered_events.last_published_seq.saturating_add(1);
let retained_count = if first_unpublished_seq <= target_seq {
ordered_events
.pending
.range(first_unpublished_seq..=target_seq)
.count() as u64
} else {
0
};
let missing_count = event_count.saturating_sub(retained_count);
if exited && !exit_known {
if missing_count != 1 {
@@ -133,18 +168,18 @@ impl SessionState {
"recovering exited process did not include its exit code".to_string(),
)
})?;
ordered_events.pending.insert(
seq,
ExecProcessEvent::Exited {
ordered_events
.insert_pending(ExecProcessEvent::Exited {
seq,
exit_code,
sandbox_denied: Some(sandbox_denied),
},
);
})
.map_err(ExecServerError::Protocol)?;
} else if missing_count != 0 {
return Err(recovery_gap_error(target_seq));
}
self.publish_ready(&mut ordered_events)
published_closed |= self.publish_ready(&mut ordered_events);
published_closed
};
self.note_change(target_seq);
codex-rs/exec-server/src/client_recovery_tests.rs
+145 -5
View File
@@ -1,5 +1,7 @@
use std::time::Duration;
use pretty_assertions::assert_eq;
use super::*;
use crate::protocol::ExecOutputStream;
use crate::protocol::ProcessOutputChunk;
@@ -57,14 +59,152 @@ fn recovery_does_not_retry_other_registry_conflicts() {
assert!(!is_retryable_recovery_error(&error));
}
#[test]
fn process_event_reorder_rejects_oversized_output() {
let state = SessionState::new(/*recoverable*/ true);
let error = state
.publish_ordered_event(ExecProcessEvent::Output(ProcessOutputChunk {
seq: 1,
stream: ExecOutputStream::Stdout,
chunk: vec![0; super::super::MAX_PENDING_PROCESS_EVENT_BYTES + 1].into(),
}))
.expect_err("oversized pending process output should be rejected");
assert!(error.contains("bytes"));
}
#[test]
fn process_event_reorder_accepts_gap_closing_event_at_limits() {
let state = SessionState::new(/*recoverable*/ true);
let chunk_size =
super::super::MAX_PENDING_PROCESS_EVENT_BYTES / super::super::MAX_PENDING_PROCESS_EVENTS;
let last_seq = super::super::MAX_PENDING_PROCESS_EVENTS as u64 + 1;
for seq in 2..=last_seq {
assert!(
!state
.publish_ordered_event(ExecProcessEvent::Output(ProcessOutputChunk {
seq,
stream: ExecOutputStream::Stdout,
chunk: vec![0; chunk_size].into(),
}))
.expect("future output should fit within reorder limits")
);
}
assert!(
!state
.publish_ordered_event(ExecProcessEvent::Output(ProcessOutputChunk {
seq: 1,
stream: ExecOutputStream::Stdout,
chunk: b"x".to_vec().into(),
}))
.expect("gap-closing output should drain the reorder buffer")
);
let ordered_events = state
.ordered_events
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
assert_eq!(
(
ordered_events.last_published_seq,
ordered_events.pending.len(),
ordered_events.pending_bytes,
),
(last_seq, 0, 0)
);
}
#[test]
fn recovery_handles_dense_tail_output_and_newer_notification() {
let state = SessionState::new(/*recoverable*/ true);
let last_seq = super::super::MAX_PENDING_PROCESS_EVENTS as u64 + 2;
let live_seq = last_seq + 1;
assert!(
!state
.publish_ordered_event(ExecProcessEvent::Output(ProcessOutputChunk {
seq: live_seq,
stream: ExecOutputStream::Stdout,
chunk: b"live".to_vec().into(),
}))
.expect("live output should remain bounded while recovery fills the gap")
);
let chunks = (2..=last_seq)
.map(|seq| ProcessOutputChunk {
seq,
stream: ExecOutputStream::Stdout,
chunk: b"x".to_vec().into(),
})
.collect();
assert!(
!state
.recover_events(ReadResponse {
chunks,
next_seq: last_seq + 1,
exited: true,
exit_code: Some(17),
closed: false,
failure: None,
sandbox_denied: false,
})
.expect("dense retained output should recover")
);
let ordered_events = state
.ordered_events
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner);
assert_eq!(
(
ordered_events.last_published_seq,
ordered_events.pending.len(),
ordered_events.pending_bytes,
),
(live_seq, 0, 0)
);
}
#[test]
fn recovery_rejects_output_at_closed_sequence() {
let state = SessionState::new(/*recoverable*/ true);
let error = state
.recover_events(ReadResponse {
chunks: vec![ProcessOutputChunk {
seq: 1,
stream: ExecOutputStream::Stdout,
chunk: b"output".to_vec().into(),
}],
next_seq: 2,
exited: false,
exit_code: None,
closed: true,
failure: None,
sandbox_denied: false,
})
.expect_err("output should not occupy the closed sequence");
assert!(
error
.to_string()
.contains("conflicts with recovered output")
);
}
#[tokio::test]
async fn recovery_adds_sandbox_denial_to_pending_exit_event() {
let state = SessionState::new(/*recoverable*/ true);
assert!(!state.publish_ordered_event(ExecProcessEvent::Exited {
seq: 2,
exit_code: 1,
sandbox_denied: None,
}));
assert!(
!state
.publish_ordered_event(ExecProcessEvent::Exited {
seq: 2,
exit_code: 1,
sandbox_denied: None,
})
.expect("pending exit should fit within reorder limits")
);
state
.recover_events(ReadResponse {