diff --git a/crates/buzz-relay/src/handlers/artifact_sync.rs b/crates/buzz-relay/src/handlers/artifact_sync.rs index fac385d635..9c3a1afaa8 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync.rs +++ b/crates/buzz-relay/src/handlers/artifact_sync.rs @@ -345,7 +345,7 @@ pub async fn handle_sync( return; } }; - let reader = match conn.auth_state_snapshot() { + let (reader, bound) = match conn.auth_state_snapshot() { AuthState::Authenticated(ctx) => { if !ctx.scopes.is_empty() && !ctx.scopes.contains(&Scope::MessagesRead) { refuse(&sub_id, "restricted: insufficient scope", &conn, &state).await; @@ -359,7 +359,7 @@ pub async fn handle_sync( return; } } - ctx.pubkey.to_bytes().to_vec() + (ctx.pubkey.to_bytes().to_vec(), ctx.channel_ids) } _ => { refuse(&sub_id, "auth-required: not authenticated", &conn, &state).await; @@ -428,7 +428,15 @@ pub async fn handle_sync( drop(permit); run_live( - &sub_id, owner, &filter, &reader, in_set, &handle, &conn, &state, + &sub_id, + owner, + &filter, + &reader, + bound.as_deref(), + in_set, + &handle, + &conn, + &state, ) .await; } @@ -645,12 +653,15 @@ async fn try_send_owned( /// Rules 3–5 after `EOSE`: read each changed head, one read at a time, and /// send it. A dropped frame or a resync request ends the sync. `in_set` is /// the client's copy as this sync sent it: the `d` of each artifact in scope. +/// `bound` is a channel-scoped token's channels: the reader reads no head +/// outside them, even where its pubkey could. #[allow(clippy::too_many_arguments)] async fn run_live( sub_id: &str, owner: u64, filter: &ArtifactSyncFilter, reader: &[u8], + bound: Option<&[Uuid]>, mut in_set: HashSet, handle: &SyncHandle, conn: &ConnectionState, @@ -672,7 +683,11 @@ async fn run_live( .artifact_head(conn.tenant.community(), reader, d, &filter.channels) .await { - Ok(head) => head, + Ok(head) => head.filter(|(head, _)| { + bound.is_none_or(|allowed| { + head.channel_id.is_some_and(|id| allowed.contains(&id)) + }) + }), Err(e) => { warn!(conn_id = %conn.conn_id, "artifact sync head read failed: {e}"); close(sub_id, owner, "error: sync read failed", conn, state).await; @@ -690,8 +705,11 @@ async fn run_live( continue; } // A head the reader cannot read leaves only the source removal. + // Without one, nothing the reader may read can remove the + // artifact from its copy, so the copy is no longer exact. let Some(event) = head.map(|(head, _)| head.event).or(removal) else { - continue; + close(sub_id, owner, "resync: artifact left scope", conn, state).await; + return; }; #[cfg(test)] super::close::test_seam::pause_at(&test_seam::LIVE_SEND_PAUSE).await; diff --git a/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs b/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs index b54a5a3125..f33b709e56 100644 --- a/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs +++ b/crates/buzz-relay/src/handlers/artifact_sync_postgres_tests.rs @@ -384,6 +384,49 @@ async fn membership_removal_resyncs_the_whole_sync() { .is_empty()); } +/// Rule 3 under a channel-scoped token: the token bounds what the reader +/// can read, so a move to a channel outside it sends the source removal, +/// never the destination head, even though the pubkey can read that +/// channel. Mutation: drop the token bound on the live head read → the +/// moved head is sent → RED. +#[tokio::test] +#[ignore = "requires Postgres and Redis"] +async fn scoped_token_gets_only_the_removal_for_a_move_outside_it() { + let f = fixture().await; + let owner = Keys::generate(); + let source = create_channel(&f.state, &f.tenant, &owner, &[]).await; + let destination = create_channel(&f.state, &f.tenant, &owner, &[]).await; + let d = Uuid::new_v4(); + let created = revision(&f, &owner, source, d, "create", None).await; + + let (conn, mut rx) = connect(&f, &owner); + match &mut *conn.auth_state.lock().unwrap() { + crate::connection::AuthState::Authenticated(ctx) => { + ctx.channel_ids = Some(vec![source]); + } + _ => unreachable!("test connections are authenticated"), + } + let task = start_sync(&f, &conn, sync_filter(&[source])); + assert_eq!(initial_set(&mut rx).await, vec![created.id.to_hex()]); + + let moved = revision(&f, &owner, destination, d, "move", Some(&created)).await; + let message = frame(&mut rx).await; + assert_eq!( + (message[0].as_str(), message[2]["kind"].as_u64()), + (Some("EVENT"), Some(45011)), + "{message}" + ); + assert_ne!(message[2]["id"].as_str(), Some(moved.id.to_hex().as_str())); + no_frame(&mut rx).await; + + super::close::handle_close("s".into(), Arc::clone(&conn), Arc::clone(&f.state)).await; + assert_eq!(frame(&mut rx).await, json!(["CLOSED", "s", ""])); + tokio::time::timeout(Duration::from_secs(5), task) + .await + .expect("sync task ends on CLOSE") + .unwrap(); +} + /// Rule 2: an unreadable or unknown channel gets `CLOSED restricted` and /// no `EOSE`, with the same reason for both. #[tokio::test]