diff --git a/codex-rs/message-board/src/local.rs b/codex-rs/message-board/src/local.rs index 6d580e6fa6..4e8bf86177 100644 --- a/codex-rs/message-board/src/local.rs +++ b/codex-rs/message-board/src/local.rs @@ -52,6 +52,7 @@ mod paging; mod pool; mod queries; pub(crate) mod setup; +mod subscriptions; #[cfg(test)] #[path = "local/pools_tests.rs"] @@ -422,27 +423,8 @@ impl LocalAgentMessageBoard { .map(Uuid::parse_str) .transpose() .map_err(storage_error)?; - // Keep active subscriptions readable by older binaries. Opt-outs only - // prevent implicit subscription when this agent participates again. - let statements = match request.change { - SubscriptionChange::Subscribe => [ - "DELETE FROM subscription_opt_outs WHERE board=? AND target=? AND agent=?", - "INSERT OR IGNORE INTO subscriptions(board,target,agent) VALUES(?,?,?)", - ], - SubscriptionChange::Unsubscribe => [ - "DELETE FROM subscriptions WHERE board=? AND target=? AND agent=?", - "INSERT OR IGNORE INTO subscription_opt_outs(board,target,agent) VALUES(?,?,?)", - ], - }; - for statement in statements { - sqlx::query(statement) - .bind(self.identity.to_string()) - .bind(target_key(&request.target)?) - .bind(target_agent.to_string()) - .execute(&mut *tx) - .await - .map_err(storage_error)?; - } + self.change_subscription(&mut tx, &request.target, target_agent, request.change) + .await?; tx.commit().await.map_err(storage_error)?; Ok(SubscriptionState { channel_name: channel, diff --git a/codex-rs/message-board/src/local/subscriptions.rs b/codex-rs/message-board/src/local/subscriptions.rs new file mode 100644 index 0000000000..74ff80d9a9 --- /dev/null +++ b/codex-rs/message-board/src/local/subscriptions.rs @@ -0,0 +1,95 @@ +//! Atomic replacement and discovery of a caller's own channel subscriptions. +//! Thread subscriptions and other agents are unaffected; no notifications are emitted. + +use super::LocalAgentMessageBoard; +use super::connection::Access; +use super::invalid; +use super::storage_error; +use super::target_key; +use super::validate_channel; +use crate::SubscriptionChange; +use crate::SubscriptionTarget; +use codex_protocol::ThreadId; +use codex_protocol::error::Result; +use sqlx::Sqlite; +use sqlx::Transaction; +use std::collections::BTreeMap; + +impl LocalAgentMessageBoard { + /// Atomically replaces the caller's membership in the supplied channels. + /// Channels outside this set keep their subscriptions and opt-outs. + /// A read-only live connection permits this self-only operation; a fixed view does not. + pub async fn set_subscriptions( + &self, + caller: ThreadId, + channels: &BTreeMap<&str, SubscriptionChange>, + ) -> Result<()> { + if self.mode.view().is_some() { + Access::ReadOnly.require_write(self.identity, "board.manage_subscriptions")?; + } + self.host.agent_path(caller).await?; + for channel in channels.keys() { + validate_channel(channel)?; + } + let names = + serde_json::to_string(&channels.keys().collect::>()).map_err(storage_error)?; + let mut tx = self.begin_write().await?; + let membership = sqlx::query_as::<_, (String, bool)>( + "SELECT c.name, EXISTS( + SELECT 1 FROM subscriptions AS s WHERE s.board=c.board + AND s.target=json_object('Channel',c.name) AND s.agent=? + ) FROM channels AS c WHERE c.board=? AND c.name IN (SELECT value FROM json_each(?))", + ) + .bind(caller.to_string()) + .bind(self.identity.to_string()) + .bind(names) + .fetch_all(&mut *tx) + .await + .map_err(storage_error)?; + if membership.len() != channels.len() { + return Err(invalid("channel not found in this board")); + } + for (name, subscribed) in membership { + let change = channels[name.as_str()]; + if subscribed == (change == SubscriptionChange::Subscribe) { + continue; + } + self.change_subscription(&mut tx, &SubscriptionTarget::Channel(name), caller, change) + .await?; + } + tx.commit().await.map_err(storage_error) + } + + /// Update explicit membership and its matching implicit-subscription opt-out atomically. + pub(super) async fn change_subscription( + &self, + tx: &mut Transaction<'_, Sqlite>, + target: &SubscriptionTarget, + agent: ThreadId, + change: SubscriptionChange, + ) -> Result<()> { + let board = self.identity.to_string(); + let target = target_key(target)?; + let agent = agent.to_string(); + let statements = match change { + SubscriptionChange::Subscribe => [ + "DELETE FROM subscription_opt_outs WHERE board=? AND target=? AND agent=?", + "INSERT OR IGNORE INTO subscriptions(board,target,agent) VALUES(?,?,?)", + ], + SubscriptionChange::Unsubscribe => [ + "DELETE FROM subscriptions WHERE board=? AND target=? AND agent=?", + "INSERT OR IGNORE INTO subscription_opt_outs(board,target,agent) VALUES(?,?,?)", + ], + }; + for statement in statements { + sqlx::query(statement) + .bind(&board) + .bind(&target) + .bind(&agent) + .execute(&mut **tx) + .await + .map_err(storage_error)?; + } + Ok(()) + } +} diff --git a/codex-rs/message-board/tests/local_board.rs b/codex-rs/message-board/tests/local_board.rs index 4b1d8b6f09..e273f7c2e8 100644 --- a/codex-rs/message-board/tests/local_board.rs +++ b/codex-rs/message-board/tests/local_board.rs @@ -1939,6 +1939,89 @@ async fn silent_appends_preserve_evidence_and_explicit_subscriptions() Ok(()) } +#[tokio::test] +async fn bulk_subscriptions_are_atomic_and_scoped_to_the_caller_and_supplied_channels() -> Result<()> +{ + use SubscriptionChange::Subscribe; + use SubscriptionChange::Unsubscribe; + use std::collections::BTreeMap; + + let dir = tempfile::tempdir()?; + let sqlite = SqliteConfig::new_for_testing(dir.path().to_path_buf().try_into().unwrap()); + let owner = ThreadId::new(); + let reviewer = ThreadId::new(); + let host = Arc::new(Host { + clock: AtomicI64::default(), + fail_clock: AtomicBool::default(), + agent_path_calls: AtomicUsize::default(), + members: [ + (owner, AgentPath::root()), + (reviewer, AgentPath::root().join("reviewer").unwrap()), + ] + .into(), + active: AtomicBool::new(true), + fail_notifications: AtomicBool::default(), + notifications: Mutex::default(), + }); + let board = LocalBoard::new(sqlite, owner.into()); + let writer = board.read_write().open(host.clone()).await?; + let reader = board.read_only().open(host.clone()).await?; + for name in ["build", "release"] { + writer + .create_channel( + owner, + CreateChannelRequest { + channel_name: name.into(), + description: None, + subscription: Subscribe, + }, + ) + .await?; + } + reader + .set_subscriptions( + reviewer, + &BTreeMap::from([("build", Subscribe), ("release", Subscribe)]), + ) + .await?; + let both = vec!["build".to_owned(), "release".to_owned()]; + assert!( + reader + .set_subscriptions( + reviewer, + &BTreeMap::from([("build", Unsubscribe), ("missing", Subscribe)]) + ) + .await + .is_err() + ); + assert_eq!( + reader + .subscribed_channels(reviewer, PageRequest::default()) + .await? + .results, + both + ); + reader + .set_subscriptions(reviewer, &BTreeMap::from([("build", Unsubscribe)])) + .await?; + assert_eq!( + reader + .subscribed_channels(reviewer, PageRequest::default()) + .await? + .results, + vec!["release".to_owned()] + ); + assert_eq!( + reader + .subscribed_channels(owner, PageRequest::default()) + .await? + .results, + both + ); + assert!(host.notifications.lock().unwrap().is_empty()); + Ok(()) +} + #[tokio::test] async fn local_board_setup_survives_reopen() -> Result<()> { use serde_json::json;