mirror of
https://github.com/open-webui/open-webui.git
synced 2026-11-04 07:47:10 +00:00
refac: invalidate socket sessions from the user model
Role changes and user deletion now tear down the user's Socket.IO sessions in the model layer, where the row is written, rather than in the admin user routes. Sessions are looked up in the shared session pool as well as in this process's room map, and the client reconnects when the server closes its socket.
This commit is contained in:
parent
8faaf2cd1e
commit
412ecbac7c
4 changed files with 35 additions and 20 deletions
|
|
@ -582,13 +582,7 @@ class UsersTable:
|
|||
return result.scalar()
|
||||
|
||||
async def update_user_role_by_id(self, id: str, role: str, db: AsyncSession | None = None) -> UserModel | None:
|
||||
async with get_async_db_context(db) as session:
|
||||
user = await session.get(User, id)
|
||||
if not user:
|
||||
return None
|
||||
user.role = role
|
||||
await session.commit()
|
||||
return UserModel.model_validate(user)
|
||||
return await self.update_user_by_id(id, {'role': role}, db=db)
|
||||
|
||||
async def update_user_status_by_id(
|
||||
self, id: str, form_data: UserStatus, db: AsyncSession | None = None
|
||||
|
|
@ -664,10 +658,18 @@ class UsersTable:
|
|||
user = await session.get(User, id)
|
||||
if not user:
|
||||
return None
|
||||
role_changed = 'role' in updated and updated['role'] != user.role
|
||||
for key, value in updated.items():
|
||||
setattr(user, key, value)
|
||||
await session.commit()
|
||||
return UserModel.model_validate(user)
|
||||
updated_user = UserModel.model_validate(user)
|
||||
|
||||
if role_changed:
|
||||
# Deferred import: socket.main imports this module.
|
||||
from open_webui.socket.main import disconnect_user_sessions
|
||||
|
||||
await disconnect_user_sessions(id)
|
||||
return updated_user
|
||||
|
||||
# settings update helper
|
||||
async def update_user_settings_by_id(
|
||||
|
|
@ -686,6 +688,7 @@ class UsersTable:
|
|||
async def delete_user_by_id(self, id: str, db: AsyncSession | None = None) -> bool:
|
||||
from open_webui.models.chats import Chats
|
||||
from open_webui.models.groups import Groups
|
||||
from open_webui.socket.main import disconnect_user_sessions
|
||||
|
||||
# Remove User from Groups
|
||||
await Groups.remove_user_from_all_groups(id)
|
||||
|
|
@ -697,7 +700,9 @@ class UsersTable:
|
|||
return False # chats deletion failed
|
||||
await session.execute(delete(User).where(User.id == id))
|
||||
await session.commit()
|
||||
return True
|
||||
|
||||
await disconnect_user_sessions(id)
|
||||
return True
|
||||
|
||||
async def get_user_api_key_by_id(self, id: str, db: AsyncSession | None = None) -> str | None:
|
||||
async with get_async_db_context(db) as session:
|
||||
|
|
|
|||
|
|
@ -36,7 +36,6 @@ from open_webui.models.access_grants import AccessGrants
|
|||
from open_webui.models.knowledge import Knowledges
|
||||
from open_webui.models.models import Models
|
||||
from open_webui.models.tools import Tools
|
||||
from open_webui.socket.main import disconnect_user_sessions
|
||||
from open_webui.utils.access_control import get_permissions, has_permission
|
||||
from open_webui.utils.auth import (
|
||||
get_admin_user,
|
||||
|
|
@ -940,10 +939,7 @@ async def update_user_by_id(
|
|||
updated_user = user
|
||||
|
||||
if updated_user:
|
||||
# If the role changed, disconnect all socket sessions so stale
|
||||
# privileges cached in SESSION_POOL are invalidated.
|
||||
if updated_user.role != user.role:
|
||||
await disconnect_user_sessions(user_id)
|
||||
await publish_event(
|
||||
request,
|
||||
EVENTS.USER_ROLE_UPDATED,
|
||||
|
|
@ -1011,7 +1007,6 @@ async def delete_user_by_id(
|
|||
result = await Auths.delete_auth_by_id(user_id, db=db)
|
||||
|
||||
if result:
|
||||
await disconnect_user_sessions(user_id)
|
||||
await publish_event(
|
||||
request,
|
||||
EVENTS.USER_DELETED,
|
||||
|
|
|
|||
|
|
@ -327,13 +327,22 @@ async def disconnect_user_sessions(user_id: str):
|
|||
fresh data from the database.
|
||||
"""
|
||||
try:
|
||||
session_ids = get_session_ids_from_room(f'user:{user_id}')
|
||||
for sid in session_ids:
|
||||
# SESSION_POOL reaches other processes; the room map also holds sockets whose pool entry
|
||||
# was dropped without disconnecting them.
|
||||
session_ids = {sid for sid, entry in SESSION_POOL.items() if entry.get('id') == user_id}
|
||||
session_ids.update(get_session_ids_from_room(f'user:{user_id}'))
|
||||
except Exception:
|
||||
# Non-fatal: whatever changed the user is already committed.
|
||||
log.exception('Failed to look up sessions for user %s', user_id)
|
||||
return
|
||||
|
||||
for sid in session_ids:
|
||||
try:
|
||||
await sio.disconnect(sid)
|
||||
if session_ids:
|
||||
log.info('Disconnected %s session(s) for user %s', len(session_ids), user_id)
|
||||
except Exception as e:
|
||||
log.warning(f'Failed to disconnect sessions for user {user_id}: {e}')
|
||||
except Exception as e:
|
||||
log.warning('Failed to disconnect session %s for user %s: %s', sid, user_id, e)
|
||||
if session_ids:
|
||||
log.info('Requested disconnect of %s session(s) for user %s', len(session_ids), user_id)
|
||||
|
||||
|
||||
@sio.on('usage')
|
||||
|
|
|
|||
|
|
@ -266,6 +266,12 @@
|
|||
heartbeatInterval = null;
|
||||
}
|
||||
|
||||
// socket.io never retries a server-initiated disconnect, and the server uses one to drop
|
||||
// sessions holding stale user data. Reconnecting re-reads that data from the database.
|
||||
if (reason === 'io server disconnect') {
|
||||
_socket.connect();
|
||||
}
|
||||
|
||||
if (details) {
|
||||
console.log('Additional details:', details);
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue