clone/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-04-24 14:45:27 +00:00
Code Issues Packages Projects Releases Wiki Activity

Plan outbound router ordering fixes

Browse Source
This commit is contained in:
jif-oai
2026-02-10 14:24:07 +00:00
parent f742019bec
commit bd89f60348
2 changed files with 23 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
codex-rs/app-server/src/lib.rs
View File

@@ -366,6 +366,7 @@ pub async fn run_main_with_transport(
}
}
let transport_event_tx_for_outbound = transport_event_tx.clone();
let outbound_handle = tokio::spawn(async move {
let mut outbound_connections = HashMap::<ConnectionId, OutboundConnectionState>::new();
loop {
@@ -374,7 +375,22 @@ pub async fn run_main_with_transport(
let Some(envelope) = envelope else {
break;
};
route_outgoing_envelope(&mut outbound_connections, envelope).await;
let disconnected_connections =
route_outgoing_envelope(&mut outbound_connections, envelope).await;
let mut should_exit = false;
for connection_id in disconnected_connections {
if transport_event_tx_for_outbound
.send(TransportEvent::ConnectionClosed { connection_id })
.await
.is_err()
{
should_exit = true;
break;
}
}
if should_exit {
break;
}
}
event = outbound_control_rx.recv() => {
let Some(event) = event else {

8
codex-rs/app-server/src/transport.rs
View File

@@ -422,7 +422,8 @@ fn serialize_outgoing_message(outgoing_message: OutgoingMessage) -> Option<Strin
pub(crate) async fn route_outgoing_envelope(
connections: &mut HashMap<ConnectionId, OutboundConnectionState>,
envelope: OutgoingEnvelope,
) {
) -> Vec<ConnectionId> {
let mut disconnected = Vec::new();
match envelope {
OutgoingEnvelope::ToConnection {
connection_id,
@@ -433,10 +434,11 @@ pub(crate) async fn route_outgoing_envelope(
"dropping message for disconnected connection: {:?}",
connection_id
);
return;
return disconnected;
};
if connection_state.writer.send(message).await.is_err() {
connections.remove(&connection_id);
disconnected.push(connection_id);
}
}
OutgoingEnvelope::Broadcast { message } => {
@@ -457,10 +459,12 @@ pub(crate) async fn route_outgoing_envelope(
};
if connection_state.writer.send(message.clone()).await.is_err() {
connections.remove(&connection_id);
disconnected.push(connection_id);
}
}
}
}
disconnected
}
pub(crate) fn has_initialized_connections(