From 5afaad79ffc1dcc0e98877f16b9b0785fbbc41b3 Mon Sep 17 00:00:00 2001 From: zijiren <84728412+zijiren233@users.noreply.github.com> Date: Sun, 16 Aug 2026 02:51:18 +0800 Subject: [PATCH] fix(realtime): separate presence events from disconnects (#416) ## Summary - Route `UserLeft` through room presence delivery only. - Reserve disconnect signals and access-revocation events for terminating realtime sessions. - Keep cross-replica presence coverage and explicitly release E2E test infrastructure. ## Testing - `make nextest` - `cargo nextest run -p synctv-api-common --no-fail-fast` - `cargo nextest run -p synctv-realtime --no-fail-fast` - `cargo nextest run -p synctv-api --test websocket_test websocket_e2e::test_ws_cross_replica_same_user_partial_disconnect_does_not_emit_user_left --run-ignored ignored-only --no-fail-fast` --- synctv-api-common/src/impls/messaging.rs | 24 ------------------- .../src/impls/messaging/event_policy.rs | 10 -------- .../src/impls/messaging/tests.rs | 6 ++--- synctv-api/tests/websocket_test.rs | 7 ++++-- synctv-realtime/src/sync/redis_pubsub.rs | 6 ++--- 5 files changed, 10 insertions(+), 43 deletions(-) diff --git a/synctv-api-common/src/impls/messaging.rs b/synctv-api-common/src/impls/messaging.rs index 4ac7a63c..e0bab5e8 100644 --- a/synctv-api-common/src/impls/messaging.rs +++ b/synctv-api-common/src/impls/messaging.rs @@ -1517,25 +1517,6 @@ impl StreamMessageHandler { break; } } - Ok(RealtimeEvent::UserLeft { ref user_id, ref room_id, .. }) => { - if self.principal.user_id() == Some(*user_id) && *room_id == self.room_id { - tracing::info!( - actor = %self.username, - room_id = %self.room_id, - "Received cross-replica UserLeft event, disconnecting" - ); - // UserLeft was already published by the leave_room - // or delete_room API call. Skip the redundant - // broadcast in cleanup(). - self.send_realtime_termination( - stream, - "Your room membership has ended", - RealtimeTerminationCode::RoomMembershipRevoked, - ); - self.skip_cleanup_user_left.store(true, std::sync::atomic::Ordering::Relaxed); - break; - } - } Ok(RealtimeEvent::RoomDeleted { ref room_id, .. }) => { if *room_id == self.room_id { self.send_realtime_termination( @@ -2622,11 +2603,6 @@ impl StreamMessageHandler { "Your room membership has ended", RealtimeTerminationCode::RoomMembershipRevoked, )), - Ok(RealtimeEvent::UserLeft { user_id: uid, room_id: rid, .. }) - if user_id == Some(*uid) && *rid == room_id => Some(realtime_termination_server_message( - "Your room membership has ended", - RealtimeTerminationCode::RoomMembershipRevoked, - )), _ => None, }; let should_disconnect = termination.is_some() diff --git a/synctv-api-common/src/impls/messaging/event_policy.rs b/synctv-api-common/src/impls/messaging/event_policy.rs index 40e19ba6..39f186d5 100644 --- a/synctv-api-common/src/impls/messaging/event_policy.rs +++ b/synctv-api-common/src/impls/messaging/event_policy.rs @@ -51,11 +51,6 @@ pub fn admin_event_requires_skip_cleanup( user_id: uid, room_id: rid, .. - } - | RealtimeEvent::UserLeft { - user_id: uid, - room_id: rid, - .. } => user_id == Some(*uid) && rid == room_id, _ => false, } @@ -91,11 +86,6 @@ pub fn watch_admin_event_matches( user_id: uid, room_id: rid, .. - } - | RealtimeEvent::UserLeft { - user_id: uid, - room_id: rid, - .. } => user_id == Some(*uid) && rid == room_id, RealtimeEvent::RoomDeleted { room_id: rid, .. } | RealtimeEvent::RoomBanned { room_id: rid, .. } diff --git a/synctv-api-common/src/impls/messaging/tests.rs b/synctv-api-common/src/impls/messaging/tests.rs index 20d51130..983e91c7 100644 --- a/synctv-api-common/src/impls/messaging/tests.rs +++ b/synctv-api-common/src/impls/messaging/tests.rs @@ -9624,7 +9624,7 @@ fn test_disconnect_signal_requires_skip_cleanup_only_for_room_scoped_or_redundan } #[test] -fn test_admin_event_requires_skip_cleanup_only_for_room_scoped_or_redundant_exits() { +fn test_admin_event_requires_skip_cleanup_only_for_room_scoped_revocations() { let rid = room_id(); let uid = user_id(); let now = synctv_core::SystemClock.now(); @@ -9650,7 +9650,7 @@ fn test_admin_event_requires_skip_cleanup_only_for_room_scoped_or_redundant_exit Some(uid), &rid, )); - assert!(super::admin_event_requires_skip_cleanup( + assert!(!super::admin_event_requires_skip_cleanup( &RealtimeEvent::UserLeft { event_id: "evt-3".to_string(), room_id: rid, @@ -9760,7 +9760,7 @@ fn test_watch_admin_event_matches_access_revocation_events() { Some(uid), &rid, )); - assert!(super::watch_admin_event_matches( + assert!(!super::watch_admin_event_matches( &RealtimeEvent::UserLeft { event_id: "evt-3".to_string(), room_id: rid, diff --git a/synctv-api/tests/websocket_test.rs b/synctv-api/tests/websocket_test.rs index a48d9ce9..814fb3a6 100644 --- a/synctv-api/tests/websocket_test.rs +++ b/synctv-api/tests/websocket_test.rs @@ -2601,8 +2601,8 @@ mod websocket_e2e { async fn test_ws_cross_replica_same_user_partial_disconnect_does_not_emit_user_left() { let infra = TestInfra::new().await; - let server1 = setup_e2e_server_with_node(&infra, "presence_replica_1").await; - let server2 = setup_e2e_server_with_node(&infra, "presence_replica_2").await; + let mut server1 = setup_e2e_server_with_node(&infra, "presence_replica_1").await; + let mut server2 = setup_e2e_server_with_node(&infra, "presence_replica_2").await; let (owner_id, owner_token) = register_test_user( &server1.user_service, @@ -2705,6 +2705,9 @@ mod websocket_e2e { .close(None) .await .expect("close remaining replica-2 connection"); + server1.shutdown().await; + server2.shutdown().await; + infra.cleanup().await; } #[tokio::test] diff --git a/synctv-realtime/src/sync/redis_pubsub.rs b/synctv-realtime/src/sync/redis_pubsub.rs index 6b3394a7..3b7a121f 100644 --- a/synctv-realtime/src/sync/redis_pubsub.rs +++ b/synctv-realtime/src/sync/redis_pubsub.rs @@ -1764,16 +1764,14 @@ impl RedisPubSub { self.handle_remote_event(Some(room_id), &event).await; - // Forward kick/leave events to admin channel for cross-replica disconnect handling. - // UserLeft is included so other replicas disconnect the user's connections - // from the room (same behavior as KickUserFromRoom but with correct semantics). + // Forward access-revocation events to the admin channel for cross-replica + // disconnect handling. UserLeft remains room-scoped presence state. if matches!( &event, RealtimeEvent::KickPublisher { .. } | RealtimeEvent::KickUserFromRoom { .. } | RealtimeEvent::RoomBanned { .. } | RealtimeEvent::RoomOwnerInactive { .. } - | RealtimeEvent::UserLeft { .. } ) { super::events::publish_admin_event( &self.admin_event_tx,