mirror of
https://github.com/openai/codex.git
synced 2026-11-13 14:46:14 +00:00
Reduce app-server JSON serialization overhead (#34761)
## What changed - Decode `JSONRPCRequest` into `ClientRequest` without serializing the entire request back into an intermediate JSON value. - Define every client method's wire name explicitly and use it directly for request, response, and tracing method lookups. - Serialize outgoing transport messages directly to JSON strings instead of first building a `serde_json::Value`. ## Testing - Add regression coverage that compares direct request conversion with Serde decoding, including omitted and null parameters, legacy methods, and unknown methods. - Verify that direct outgoing-message serialization preserves the notification wire shape. GitOrigin-RevId: 304e14bcaea24e32743bf852a60a208e2f273db3
This commit is contained in:
1 parent
65ae4c26e0
commit
f69dc49b15
4 files changed
+148
-47
No files matched your search
@@ -200,7 +200,7 @@ macro_rules! client_request_definitions {
|
|||||||
$(
|
$(
|
||||||
$(#[experimental($reason:expr)])?
|
$(#[experimental($reason:expr)])?
|
||||||
$(#[doc = $variant_doc:literal])*
|
$(#[doc = $variant_doc:literal])*
|
||||||
$variant:ident $(=> $wire:literal)? {
|
$variant:ident => $wire:literal {
|
||||||
params: $(#[$params_meta:meta])* $params:ty,
|
params: $(#[$params_meta:meta])* $params:ty,
|
||||||
$(inspect_params: $inspect_params:tt,)?
|
$(inspect_params: $inspect_params:tt,)?
|
||||||
serialization: $serialization:ident $( ( $($serialization_args:tt)* ) )?,
|
serialization: $serialization:ident $( ( $($serialization_args:tt)* ) )?,
|
||||||
@@ -215,7 +215,8 @@ macro_rules! client_request_definitions {
|
|||||||
pub enum ClientRequest {
|
pub enum ClientRequest {
|
||||||
$(
|
$(
|
||||||
$(#[doc = $variant_doc])*
|
$(#[doc = $variant_doc])*
|
||||||
$(#[serde(rename = $wire)] #[ts(rename = $wire)])?
|
#[serde(rename = $wire)]
|
||||||
|
#[ts(rename = $wire)]
|
||||||
$variant {
|
$variant {
|
||||||
#[serde(rename = "id")]
|
#[serde(rename = "id")]
|
||||||
request_id: RequestId,
|
request_id: RequestId,
|
||||||
@@ -232,16 +233,14 @@ macro_rules! client_request_definitions {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub const fn method_name(&self) -> &'static str {
|
||||||
|
match self {
|
||||||
|
$(Self::$variant { .. } => $wire,)*
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pub fn method(&self) -> String {
|
pub fn method(&self) -> String {
|
||||||
serde_json::to_value(self)
|
self.method_name().to_string()
|
||||||
.ok()
|
|
||||||
.and_then(|value| {
|
|
||||||
value
|
|
||||||
.get("method")
|
|
||||||
.and_then(serde_json::Value::as_str)
|
|
||||||
.map(str::to_owned)
|
|
||||||
})
|
|
||||||
.unwrap_or_else(|| "<unknown>".to_string())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn serialization_scope(&self) -> Option<ClientRequestSerializationScope> {
|
pub fn serialization_scope(&self) -> Option<ClientRequestSerializationScope> {
|
||||||
@@ -258,6 +257,26 @@ macro_rules! client_request_definitions {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
impl TryFrom<JSONRPCRequest> for ClientRequest {
|
||||||
|
type Error = serde_json::Error;
|
||||||
|
|
||||||
|
fn try_from(request: JSONRPCRequest) -> Result<Self, Self::Error> {
|
||||||
|
let JSONRPCRequest {
|
||||||
|
id: request_id,
|
||||||
|
method,
|
||||||
|
params,
|
||||||
|
trace: _,
|
||||||
|
} = request;
|
||||||
|
let mut request = serde_json::Map::new();
|
||||||
|
request.insert("id".to_string(), serde_json::to_value(request_id)?);
|
||||||
|
request.insert("method".to_string(), serde_json::Value::String(method));
|
||||||
|
if let Some(params) = params {
|
||||||
|
request.insert("params".to_string(), params);
|
||||||
|
}
|
||||||
|
serde_json::from_value(serde_json::Value::Object(request))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Typed response from the server to the client.
|
/// Typed response from the server to the client.
|
||||||
#[derive(Serialize, Deserialize, Debug, Clone)]
|
#[derive(Serialize, Deserialize, Debug, Clone)]
|
||||||
#[allow(clippy::large_enum_variant)]
|
#[allow(clippy::large_enum_variant)]
|
||||||
@@ -265,7 +284,7 @@ macro_rules! client_request_definitions {
|
|||||||
pub enum ClientResponse {
|
pub enum ClientResponse {
|
||||||
$(
|
$(
|
||||||
$(#[doc = $variant_doc])*
|
$(#[doc = $variant_doc])*
|
||||||
$(#[serde(rename = $wire)])?
|
#[serde(rename = $wire)]
|
||||||
$variant {
|
$variant {
|
||||||
#[serde(rename = "id")]
|
#[serde(rename = "id")]
|
||||||
request_id: RequestId,
|
request_id: RequestId,
|
||||||
@@ -282,15 +301,9 @@ macro_rules! client_request_definitions {
|
|||||||
}
|
}
|
||||||
|
|
||||||
pub fn method(&self) -> String {
|
pub fn method(&self) -> String {
|
||||||
serde_json::to_value(self)
|
match self {
|
||||||
.ok()
|
$(Self::$variant { .. } => $wire.to_string(),)*
|
||||||
.and_then(|value| {
|
}
|
||||||
value
|
|
||||||
.get("method")
|
|
||||||
.and_then(serde_json::Value::as_str)
|
|
||||||
.map(str::to_owned)
|
|
||||||
})
|
|
||||||
.unwrap_or_else(|| "<unknown>".to_string())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn into_jsonrpc_parts(
|
pub fn into_jsonrpc_parts(
|
||||||
@@ -405,7 +418,7 @@ macro_rules! client_request_definitions {
|
|||||||
|
|
||||||
pub(crate) const EXPERIMENTAL_CLIENT_METHODS: &[&str] = &[
|
pub(crate) const EXPERIMENTAL_CLIENT_METHODS: &[&str] = &[
|
||||||
$(
|
$(
|
||||||
experimental_method_entry!($(#[experimental($reason)])? $(=> $wire)?),
|
experimental_method_entry!($(#[experimental($reason)])? => $wire),
|
||||||
)*
|
)*
|
||||||
];
|
];
|
||||||
pub(crate) const EXPERIMENTAL_CLIENT_METHOD_PARAM_TYPES: &[&str] = &[
|
pub(crate) const EXPERIMENTAL_CLIENT_METHOD_PARAM_TYPES: &[&str] = &[
|
||||||
@@ -470,7 +483,7 @@ macro_rules! client_response_payload_from_impl {
|
|||||||
}
|
}
|
||||||
|
|
||||||
client_request_definitions! {
|
client_request_definitions! {
|
||||||
Initialize {
|
Initialize => "initialize" {
|
||||||
params: v1::InitializeParams,
|
params: v1::InitializeParams,
|
||||||
serialization: None,
|
serialization: None,
|
||||||
response: v1::InitializeResponse,
|
response: v1::InitializeResponse,
|
||||||
@@ -1177,25 +1190,25 @@ client_request_definitions! {
|
|||||||
},
|
},
|
||||||
|
|
||||||
/// DEPRECATED APIs below
|
/// DEPRECATED APIs below
|
||||||
GetConversationSummary {
|
GetConversationSummary => "getConversationSummary" {
|
||||||
params: v1::GetConversationSummaryParams,
|
params: v1::GetConversationSummaryParams,
|
||||||
serialization: None,
|
serialization: None,
|
||||||
response: v1::GetConversationSummaryResponse,
|
response: v1::GetConversationSummaryResponse,
|
||||||
},
|
},
|
||||||
GitDiffToRemote {
|
GitDiffToRemote => "gitDiffToRemote" {
|
||||||
params: v1::GitDiffToRemoteParams,
|
params: v1::GitDiffToRemoteParams,
|
||||||
serialization: None,
|
serialization: None,
|
||||||
response: v1::GitDiffToRemoteResponse,
|
response: v1::GitDiffToRemoteResponse,
|
||||||
},
|
},
|
||||||
/// DEPRECATED in favor of GetAccount
|
/// DEPRECATED in favor of GetAccount
|
||||||
GetAuthStatus {
|
GetAuthStatus => "getAuthStatus" {
|
||||||
params: v1::GetAuthStatusParams,
|
params: v1::GetAuthStatusParams,
|
||||||
serialization: global("account-auth"),
|
serialization: global("account-auth"),
|
||||||
response: v1::GetAuthStatusResponse,
|
response: v1::GetAuthStatusResponse,
|
||||||
},
|
},
|
||||||
// Legacy fuzzy search cancellation is intentionally concurrent: clients reuse a
|
// Legacy fuzzy search cancellation is intentionally concurrent: clients reuse a
|
||||||
// cancellation token so a newer request can cancel an older in-flight search.
|
// cancellation token so a newer request can cancel an older in-flight search.
|
||||||
FuzzyFileSearch {
|
FuzzyFileSearch => "fuzzyFileSearch" {
|
||||||
params: FuzzyFileSearchParams,
|
params: FuzzyFileSearchParams,
|
||||||
serialization: None,
|
serialization: None,
|
||||||
response: FuzzyFileSearchResponse,
|
response: FuzzyFileSearchResponse,
|
||||||
@@ -1798,6 +1811,81 @@ mod tests {
|
|||||||
RequestId::Integer(REQUEST_ID)
|
RequestId::Integer(REQUEST_ID)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn decode_client_request_through_json(
|
||||||
|
request: &JSONRPCRequest,
|
||||||
|
) -> std::result::Result<ClientRequest, String> {
|
||||||
|
serde_json::to_value(request)
|
||||||
|
.and_then(serde_json::from_value)
|
||||||
|
.map_err(|err| err.to_string())
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn jsonrpc_request_conversion_preserves_serde_enum_decoding() {
|
||||||
|
let requests = [
|
||||||
|
JSONRPCRequest {
|
||||||
|
id: RequestId::Integer(1),
|
||||||
|
method: "thread/archive".to_string(),
|
||||||
|
params: Some(json!({"threadId": "thread-1"})),
|
||||||
|
trace: Some(codex_protocol::protocol::W3cTraceContext {
|
||||||
|
traceparent: Some("traceparent".to_string()),
|
||||||
|
tracestate: Some("tracestate".to_string()),
|
||||||
|
}),
|
||||||
|
},
|
||||||
|
// Required params preserve distinct omitted and explicit-null errors.
|
||||||
|
JSONRPCRequest {
|
||||||
|
id: RequestId::Integer(2),
|
||||||
|
method: "thread/archive".to_string(),
|
||||||
|
params: None,
|
||||||
|
trace: None,
|
||||||
|
},
|
||||||
|
JSONRPCRequest {
|
||||||
|
id: RequestId::Integer(3),
|
||||||
|
method: "thread/archive".to_string(),
|
||||||
|
params: Some(serde_json::Value::Null),
|
||||||
|
trace: None,
|
||||||
|
},
|
||||||
|
// Optional unit params preserve omitted, null, and empty-object behavior.
|
||||||
|
JSONRPCRequest {
|
||||||
|
id: RequestId::Integer(4),
|
||||||
|
method: "memory/reset".to_string(),
|
||||||
|
params: None,
|
||||||
|
trace: None,
|
||||||
|
},
|
||||||
|
JSONRPCRequest {
|
||||||
|
id: RequestId::Integer(5),
|
||||||
|
method: "memory/reset".to_string(),
|
||||||
|
params: Some(serde_json::Value::Null),
|
||||||
|
trace: None,
|
||||||
|
},
|
||||||
|
JSONRPCRequest {
|
||||||
|
id: RequestId::Integer(6),
|
||||||
|
method: "memory/reset".to_string(),
|
||||||
|
params: Some(json!({})),
|
||||||
|
trace: None,
|
||||||
|
},
|
||||||
|
JSONRPCRequest {
|
||||||
|
id: RequestId::Integer(7),
|
||||||
|
method: "getConversationSummary".to_string(),
|
||||||
|
params: Some(json!({
|
||||||
|
"conversationId": "67e55044-10b1-426f-9247-bb680e5fe0c8"
|
||||||
|
})),
|
||||||
|
trace: None,
|
||||||
|
},
|
||||||
|
JSONRPCRequest {
|
||||||
|
id: RequestId::Integer(8),
|
||||||
|
method: "unknown/method".to_string(),
|
||||||
|
params: Some(json!({})),
|
||||||
|
trace: None,
|
||||||
|
},
|
||||||
|
];
|
||||||
|
|
||||||
|
for request in requests {
|
||||||
|
let expected = decode_client_request_through_json(&request);
|
||||||
|
let actual = ClientRequest::try_from(request).map_err(|err| err.to_string());
|
||||||
|
assert_eq!(actual, expected);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn client_request_serialization_scope_covers_keyed_families() {
|
fn client_request_serialization_scope_covers_keyed_families() {
|
||||||
let thread_id = "thread-1".to_string();
|
let thread_id = "thread-1".to_string();
|
||||||
|
|||||||
@@ -256,14 +256,7 @@ async fn enqueue_incoming_message(
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn serialize_outgoing_message(outgoing_message: OutgoingMessage) -> Option<String> {
|
fn serialize_outgoing_message(outgoing_message: OutgoingMessage) -> Option<String> {
|
||||||
let value = match serde_json::to_value(outgoing_message) {
|
match serde_json::to_string(&outgoing_message) {
|
||||||
Ok(value) => value,
|
|
||||||
Err(err) => {
|
|
||||||
error!("Failed to convert OutgoingMessage to JSON value: {err}");
|
|
||||||
return None;
|
|
||||||
}
|
|
||||||
};
|
|
||||||
match serde_json::to_string(&value) {
|
|
||||||
Ok(json) => Some(json),
|
Ok(json) => Some(json),
|
||||||
Err(err) => {
|
Err(err) => {
|
||||||
error!("Failed to serialize JSONRPCMessage: {err}");
|
error!("Failed to serialize JSONRPCMessage: {err}");
|
||||||
@@ -295,6 +288,32 @@ mod tests {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn serialize_outgoing_message_preserves_wire_shape() {
|
||||||
|
let message = OutgoingMessage::AppServerNotification(ServerNotificationEnvelope {
|
||||||
|
notification: ServerNotification::ConfigWarning(ConfigWarningNotification {
|
||||||
|
summary: "summary".to_string(),
|
||||||
|
details: None,
|
||||||
|
path: None,
|
||||||
|
range: None,
|
||||||
|
}),
|
||||||
|
emitted_at_ms: Some(1_234),
|
||||||
|
});
|
||||||
|
|
||||||
|
let json = serialize_outgoing_message(message).expect("message should serialize");
|
||||||
|
assert_eq!(
|
||||||
|
serde_json::from_str::<serde_json::Value>(&json).expect("message should be valid JSON"),
|
||||||
|
json!({
|
||||||
|
"method": "configWarning",
|
||||||
|
"params": {
|
||||||
|
"summary": "summary",
|
||||||
|
"details": null,
|
||||||
|
},
|
||||||
|
"emittedAtMs": 1_234,
|
||||||
|
})
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
#[tokio::test]
|
#[tokio::test]
|
||||||
async fn enqueue_incoming_request_returns_overload_error_when_queue_is_full() {
|
async fn enqueue_incoming_request_returns_overload_error_when_queue_is_full() {
|
||||||
let connection_id = ConnectionId(42);
|
let connection_id = ConnectionId(42);
|
||||||
|
|||||||
@@ -64,8 +64,8 @@ pub(crate) fn typed_request_span(
|
|||||||
connection_id: ConnectionId,
|
connection_id: ConnectionId,
|
||||||
session: &ConnectionSessionState,
|
session: &ConnectionSessionState,
|
||||||
) -> Span {
|
) -> Span {
|
||||||
let method = request.method();
|
let method = request.method_name();
|
||||||
let span = app_server_request_span_template(&method, "in-process", request.id(), connection_id);
|
let span = app_server_request_span_template(method, "in-process", request.id(), connection_id);
|
||||||
|
|
||||||
let client_info = initialize_client_info_from_typed_request(request);
|
let client_info = initialize_client_info_from_typed_request(request);
|
||||||
record_client_info(
|
record_client_info(
|
||||||
@@ -78,7 +78,7 @@ pub(crate) fn typed_request_span(
|
|||||||
.or(session.client_version()),
|
.or(session.client_version()),
|
||||||
);
|
);
|
||||||
|
|
||||||
attach_parent_context(&span, &method, request.id(), /*parent_trace*/ None);
|
attach_parent_context(&span, method, request.id(), /*parent_trace*/ None);
|
||||||
span
|
span
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -88,15 +88,9 @@ use crate::models_refresh_worker::ModelsRefreshWorker;
|
|||||||
|
|
||||||
const CONNECTION_RPC_DRAIN_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 30);
|
const CONNECTION_RPC_DRAIN_TIMEOUT: Duration = Duration::from_secs(/*secs*/ 30);
|
||||||
|
|
||||||
fn deserialize_client_request(
|
fn deserialize_client_request(request: JSONRPCRequest) -> Result<ClientRequest, JSONRPCErrorError> {
|
||||||
request: &JSONRPCRequest,
|
ClientRequest::try_from(request)
|
||||||
) -> Result<ClientRequest, JSONRPCErrorError> {
|
|
||||||
serde_json::to_value(request)
|
|
||||||
.map_err(|err| invalid_request(format!("Invalid request: {err}")))
|
.map_err(|err| invalid_request(format!("Invalid request: {err}")))
|
||||||
.and_then(|request_json| {
|
|
||||||
serde_json::from_value(request_json)
|
|
||||||
.map_err(|err| invalid_request(format!("Invalid request: {err}")))
|
|
||||||
})
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) struct MessageProcessor {
|
pub(crate) struct MessageProcessor {
|
||||||
@@ -543,7 +537,7 @@ impl MessageProcessor {
|
|||||||
Arc::clone(&self.outgoing),
|
Arc::clone(&self.outgoing),
|
||||||
request_context.clone(),
|
request_context.clone(),
|
||||||
async {
|
async {
|
||||||
let codex_request = deserialize_client_request(&request);
|
let codex_request = deserialize_client_request(request);
|
||||||
let result = match codex_request {
|
let result = match codex_request {
|
||||||
Ok(codex_request) => {
|
Ok(codex_request) => {
|
||||||
// Websocket callers finalize outbound readiness in lib.rs after mirroring
|
// Websocket callers finalize outbound readiness in lib.rs after mirroring
|
||||||
|
|||||||
Reference in new issue
Block a user