mirror of
https://github.com/open-webui/open-webui.git
synced 2026-09-06 08:18:53 +00:00
fix(backend): clear orphaned Redis task IDs on startup and explicit stop (#22525)
- Add clear_all_tasks() function to clear orphaned task IDs from Redis and memory - Call clear_all_tasks() in FastAPI lifespan handler on startup - Enhance stop_task() to explicitly clean up both Redis cache and in-memory state - Ensure cleanup happens even if background task is dead - Prevents UI from being bricked after server restarts Fixes #22525
This commit is contained in:
parent
66c9bf57da
commit
c0924206d7
2 changed files with 56 additions and 0 deletions
|
|
@ -559,6 +559,7 @@ from open_webui.tasks import (
|
|||
create_task,
|
||||
stop_task,
|
||||
list_tasks,
|
||||
clear_all_tasks,
|
||||
) # Import from tasks.py
|
||||
|
||||
from open_webui.utils.redis import get_sentinels_from_env
|
||||
|
|
@ -638,6 +639,9 @@ async def lifespan(app: FastAPI):
|
|||
async_mode=True,
|
||||
)
|
||||
|
||||
# Clear orphaned tasks on startup to prevent UI from being bricked
|
||||
await clear_all_tasks(app.state.redis)
|
||||
|
||||
if app.state.redis is not None:
|
||||
app.state.redis_task_command_listener = asyncio.create_task(redis_task_command_listener(app))
|
||||
|
||||
|
|
|
|||
|
|
@ -149,6 +149,9 @@ async def stop_task(redis, task_id: str):
|
|||
if redis:
|
||||
# Look up the item_id before cleanup so we can remove the set entry too
|
||||
item_id = await redis.hget(REDIS_TASKS_KEY, task_id)
|
||||
if item_id:
|
||||
item_id = item_id.decode('utf-8') if isinstance(item_id, bytes) else item_id
|
||||
|
||||
# PUBSUB: All instances check if they have this task, and stop if so.
|
||||
await redis_send_command(
|
||||
redis,
|
||||
|
|
@ -159,7 +162,17 @@ async def stop_task(redis, task_id: str):
|
|||
)
|
||||
# Always clean Redis directly — hdel/srem are idempotent, safe even
|
||||
# if the done_callback on the owning process also fires cleanup.
|
||||
# This ensures orphaned tasks are removed even if the background task is dead.
|
||||
await redis_cleanup_task(redis, task_id, item_id or None)
|
||||
|
||||
# Also clean up local in-memory state
|
||||
tasks.pop(task_id, None)
|
||||
if item_id and item_id in item_tasks:
|
||||
if task_id in item_tasks[item_id]:
|
||||
item_tasks[item_id].remove(task_id)
|
||||
if not item_tasks[item_id]:
|
||||
item_tasks.pop(item_id, None)
|
||||
|
||||
return {'status': True, 'message': f'Task {task_id} stopped.'}
|
||||
|
||||
task = tasks.pop(task_id, None)
|
||||
|
|
@ -208,3 +221,42 @@ async def get_active_chat_ids(redis, chat_ids: List[str]) -> List[str]:
|
|||
if await has_active_tasks(redis, chat_id):
|
||||
active.append(chat_id)
|
||||
return active
|
||||
|
||||
|
||||
async def clear_all_tasks(redis: Optional[Redis]):
|
||||
"""
|
||||
Clear all active tasks from memory and Redis cache.
|
||||
This should be called on server startup to prevent orphaned task IDs
|
||||
from bricking the UI after a restart.
|
||||
"""
|
||||
log.info('Clearing all orphaned tasks from cache...')
|
||||
|
||||
# Clear in-memory task dictionaries
|
||||
tasks.clear()
|
||||
item_tasks.clear()
|
||||
|
||||
# Clear Redis cache if available
|
||||
if redis:
|
||||
try:
|
||||
# Delete all task-related keys
|
||||
await redis.delete(REDIS_TASKS_KEY)
|
||||
|
||||
# Find and delete all item task sets
|
||||
pattern = f'{REDIS_ITEM_TASKS_KEY}:*'
|
||||
cursor = 0
|
||||
deleted_count = 0
|
||||
|
||||
# Use SCAN to iterate through keys matching the pattern
|
||||
while True:
|
||||
cursor, keys = await redis.scan(cursor, match=pattern, count=100)
|
||||
if keys:
|
||||
await redis.delete(*keys)
|
||||
deleted_count += len(keys)
|
||||
if cursor == 0:
|
||||
break
|
||||
|
||||
log.info(f'Cleared {deleted_count} item task sets from Redis')
|
||||
except Exception as e:
|
||||
log.error(f'Error clearing tasks from Redis: {e}')
|
||||
|
||||
log.info('All orphaned tasks cleared successfully')
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue