openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-13 11:47:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(app-server): clear subscriber snapshots on teardown

Browse Source
This commit is contained in:
Fabian Ponce
2026-06-23 14:44:44 +00:00
parent b78df5cce7
commit 47768aec2c
2 changed files with 20 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
codex-rs/app-server/src/request_processors/thread_processor_tests.rs
View File

@@ -1255,6 +1255,16 @@ mod thread_processor_behavior_tests {
)
.await
.expect("connection should be live");
let connection_ids = manager
.subscribe_to_connection_ids(thread_id)
.await
.expect("thread should have a connection-ids watcher");
let has_connections = manager
.subscribe_to_has_connections(thread_id)
.await
.expect("thread should have a has-connections watcher");
assert_eq!(connection_ids.borrow().as_ref(), &[connection]);
assert!(*has_connections.borrow());
{
let state = manager.thread_state(thread_id).await;
let mut state = state.lock().await;
@@ -1273,6 +1283,8 @@ mod thread_processor_behavior_tests {
manager.remove_thread_state(thread_id).await;
assert_eq!(cancel_rx.await, Ok(()));
assert!(connection_ids.borrow().is_empty());
assert!(!*has_connections.borrow());
let state = manager.thread_state(thread_id).await;
let subscribed_connection_ids = manager.subscribed_connection_ids(thread_id).await;

12
codex-rs/app-server/src/thread_state.rs
View File

@@ -429,10 +429,14 @@ impl ThreadStateManager {
pub(crate) async fn remove_thread_state(&self, thread_id: ThreadId) {
let thread_state = {
let mut state = self.state.lock().await;
let thread_state = state
.threads
.remove(&thread_id)
.map(|thread_entry| thread_entry.state);
let thread_state = state.threads.remove(&thread_id).map(|mut thread_entry| {
// Watch receivers retain their last value after the sender is dropped. Publish
// a terminal empty snapshot so an in-flight listener cannot dispatch to stale
// subscribers while teardown cancellation is being observed.
thread_entry.connection_ids.clear();
thread_entry.publish_connections();
thread_entry.state
});
state.thread_ids_by_connection.retain(|_, thread_ids| {
thread_ids.remove(&thread_id);
!thread_ids.is_empty()