feat: prevent duplicate log clearing tasks with Redis locks and optimistic UI updates

This commit addresses the reviewer's concern about duplicate task submissions
by implementing a multi-layered prevention strategy.

## Backend improvements:

1. **Redis-based task deduplication**
   - Added Redis lock mechanism in ConversationService.clear_conversations()
   - Lock key format: `clear_conversations:{app_id}:{mode}[:{conversation_ids_hash}]`
   - Lock expires in 10 minutes, preventing stale locks
   - Returns 409 Conflict if clearing task already in progress

2. **New exception handling**
   - Added ConversationClearInProgressError exception
   - Controllers return 409 status with clear error message
   - Lock is automatically released on task completion or max retries

3. **Celery task improvements**
   - Task accepts lock_key parameter for cleanup
   - Releases lock on successful completion
   - Releases lock after max retries to prevent deadlock

## Frontend improvements:

1. **Optimistic UI updates**
   - List immediately cleared from UI when user clicks "Clear All"
   - For selective deletion, removed items are filtered out instantly
   - Prevents user confusion and repeat submissions

2. **Improved user experience**
   - No need to wait for backend processing to see results
   - Clear visual feedback that operation started
   - Backend revalidation happens after optimistic update

## How it works:

1. User clicks "Clear All" → List immediately disappears (optimistic update)
2. Frontend sends DELETE request → Backend checks Redis lock
3. If lock exists → Return 409 error (task already running)
4. If no lock → Acquire lock and queue Celery task
5. Celery task processes deletion in background
6. Task releases lock when done or after max retries
7. Frontend revalidates data after request completes

This prevents duplicate tasks even in edge cases like:
- Cross-tab submissions
- Slow network + multiple clicks
- Page refresh during operation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: connermo

api/controllers/console/app/conversation.py

@@ -149,6 +149,8 @@ def get(self, app_model):
     @get_app_model(mode=AppMode.COMPLETION)
     @edit_permission_required
     def delete(self, app_model):
+        from services.errors.conversation import ConversationClearInProgressError
+
         current_user, _ = current_account_with_tenant()
         parser = reqparse.RequestParser()
         parser.add_argument("conversation_ids", type=list, location="json", required=False, default=None)
@@ -160,11 +162,13 @@ def delete(self, app_model):
             [str(id) for id in conversation_ids_raw] if conversation_ids_raw and len(conversation_ids_raw) > 0 else None
         )
 
-        result = ConversationService.clear_conversations(
-            app_model=app_model, user=current_user, conversation_ids=conversation_ids
-        )
-
-        return result, 202
+        try:
+            result = ConversationService.clear_conversations(
+                app_model=app_model, user=current_user, conversation_ids=conversation_ids
+            )
+            return result, 202
+        except ConversationClearInProgressError as e:
+            return {"message": str(e), "code": "task_in_progress"}, 409
 
 
 @console_ns.route("/apps/<uuid:app_id>/completion-conversations/<uuid:conversation_id>")
@@ -394,6 +398,8 @@ def get(self, app_model):
     @get_app_model(mode=[AppMode.CHAT, AppMode.AGENT_CHAT, AppMode.ADVANCED_CHAT])
     @edit_permission_required
     def delete(self, app_model):
+        from services.errors.conversation import ConversationClearInProgressError
+
         current_user, _ = current_account_with_tenant()
         parser = reqparse.RequestParser()
         parser.add_argument("conversation_ids", type=list, location="json", required=False, default=None)
@@ -405,11 +411,13 @@ def delete(self, app_model):
             [str(id) for id in conversation_ids_raw] if conversation_ids_raw and len(conversation_ids_raw) > 0 else None
         )
 
-        result = ConversationService.clear_conversations(
-            app_model=app_model, user=current_user, conversation_ids=conversation_ids
-        )
-
-        return result, 202
+        try:
+            result = ConversationService.clear_conversations(
+                app_model=app_model, user=current_user, conversation_ids=conversation_ids
+            )
+            return result, 202
+        except ConversationClearInProgressError as e:
+            return {"message": str(e), "code": "task_in_progress"}, 409
 
 
 @console_ns.route("/apps/<uuid:app_id>/chat-conversations/<uuid:conversation_id>")

api/services/conversation_service.py

@@ -1,4 +1,5 @@
 import contextlib
+import hashlib
 import logging
 from collections.abc import Callable, Sequence
 from typing import Any, Union
@@ -11,12 +12,14 @@
 from core.variables.types import SegmentType
 from core.workflow.nodes.variable_assigner.common.impl import conversation_variable_updater_factory
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from factories import variable_factory
 from libs.datetime_utils import naive_utc_now
 from libs.infinite_scroll_pagination import InfiniteScrollPagination
 from models import Account, ConversationVariable
 from models.model import App, Conversation, EndUser, Message
 from services.errors.conversation import (
+    ConversationClearInProgressError,
     ConversationNotExistsError,
     ConversationVariableNotExistsError,
     ConversationVariableTypeMismatchError,
@@ -232,6 +235,9 @@ def clear_conversations(
 
         Returns:
             dict with task info and estimated counts
+
+        Raises:
+            ConversationClearInProgressError: If a clearing task is already in progress
         """
         # Validate conversation ownership if specific IDs provided
         if conversation_ids:
@@ -244,39 +250,62 @@ def clear_conversations(
         else:
             mode = "chat"  # covers chat, agent-chat, advanced-chat
 
-        # Queue the Celery task
-        task = clear_conversations_task.delay(
-            app_id=app_model.id,
-            conversation_mode=mode,
-            conversation_ids=conversation_ids,
-            user_id=user.id if user else None,
-            user_type="account" if isinstance(user, Account) else "end_user" if isinstance(user, EndUser) else None,
-        )
-
-        # Get estimated counts for response
+        # Generate unique lock key to prevent duplicate tasks
+        # Key format: clear_conversations:{app_id}:{mode}[:{conversation_ids_hash}]
+        lock_key = f"clear_conversations:{app_model.id}:{mode}"
         if conversation_ids:
-            conversation_count = len(conversation_ids)
-        else:
-            # Estimate total conversations for this app
-            conversation_count = (
-                db.session.query(Conversation)
-                .filter(
-                    Conversation.app_id == app_model.id,
-                    Conversation.mode == mode,
-                    Conversation.from_source == ("api" if isinstance(user, EndUser) else "console"),
-                    Conversation.from_end_user_id == (user.id if isinstance(user, EndUser) else None),
-                    Conversation.from_account_id == (user.id if isinstance(user, Account) else None),
-                    Conversation.is_deleted == False,
-                )
-                .count()
+            # For selective deletion, include hash of conversation IDs
+            ids_hash = hashlib.md5(",".join(sorted(conversation_ids)).encode()).hexdigest()[:8]
+            lock_key += f":{ids_hash}"
+
+        # Try to acquire lock (expires in 10 minutes)
+        # nx=True means set only if key doesn't exist
+        lock_acquired = redis_client.set(lock_key, "1", ex=600, nx=True)
+        if not lock_acquired:
+            # Another clearing task is already in progress
+            raise ConversationClearInProgressError(
+                "A conversation clearing task is already in progress for this app. Please wait for it to complete."
+            )
+
+        try:
+            # Queue the Celery task with lock_key for cleanup
+            task = clear_conversations_task.delay(
+                app_id=app_model.id,
+                conversation_mode=mode,
+                conversation_ids=conversation_ids,
+                user_id=user.id if user else None,
+                user_type="account" if isinstance(user, Account) else "end_user" if isinstance(user, EndUser) else None,
+                lock_key=lock_key,
             )
 
-        return {
-            "task_id": task.id,
-            "status": "queued",
-            "estimated_conversations": conversation_count,
-            "mode": "selective" if conversation_ids else "all",
-        }
+            # Get estimated counts for response
+            if conversation_ids:
+                conversation_count = len(conversation_ids)
+            else:
+                # Estimate total conversations for this app
+                conversation_count = (
+                    db.session.query(Conversation)
+                    .filter(
+                        Conversation.app_id == app_model.id,
+                        Conversation.mode == mode,
+                        Conversation.from_source == ("api" if isinstance(user, EndUser) else "console"),
+                        Conversation.from_end_user_id == (user.id if isinstance(user, EndUser) else None),
+                        Conversation.from_account_id == (user.id if isinstance(user, Account) else None),
+                        Conversation.is_deleted == False,
+                    )
+                    .count()
+                )
+
+            return {
+                "task_id": task.id,
+                "status": "queued",
+                "estimated_conversations": conversation_count,
+                "mode": "selective" if conversation_ids else "all",
+            }
+        except Exception:
+            # Release lock if task queueing failed
+            redis_client.delete(lock_key)
+            raise
 
     @classmethod
     def get_conversational_variable(

api/services/errors/conversation.py

@@ -19,3 +19,9 @@ class ConversationVariableNotExistsError(BaseServiceError):
 
 class ConversationVariableTypeMismatchError(BaseServiceError):
     pass
+
+
+class ConversationClearInProgressError(BaseServiceError):
+    """Raised when a conversation clear operation is already in progress for the same app/mode."""
+
+    pass

api/tasks/clear_conversation_task.py

@@ -5,6 +5,7 @@
 from sqlalchemy.orm import sessionmaker
 
 from extensions.ext_database import db
+from extensions.ext_redis import redis_client
 from extensions.ext_storage import storage
 from models import (
     Conversation,
@@ -32,6 +33,7 @@ def clear_conversations_task(
     conversation_ids: list[str] | None = None,
     user_id: str | None = None,
     user_type: str | None = None,
+    lock_key: str | None = None,
 ):
     """
     Celery task to clear conversations and related data.
@@ -42,6 +44,7 @@ def clear_conversations_task(
         conversation_ids: Optional list of specific conversation IDs to clear
         user_id: The user ID for permission validation
         user_type: 'account' or 'end_user'
+        lock_key: Redis lock key to release after completion
     """
     start_time = time.time()
     total_deleted = {
@@ -160,6 +163,14 @@ def clear_conversations_task(
         execution_time = time.time() - start_time
         logger.info("Conversation cleanup completed in %.2fs. Deleted: %s", execution_time, total_deleted)
 
+        # Release Redis lock on successful completion
+        if lock_key:
+            try:
+                redis_client.delete(lock_key)
+                logger.info("Released lock: %s", lock_key)
+            except Exception as e:
+                logger.warning("Failed to release lock %s: %s", lock_key, e)
+
         return {
             "status": "completed",
             "execution_time": execution_time,
@@ -178,6 +189,13 @@ def clear_conversations_task(
             raise self.retry(exc=exc, countdown=60 * (2**self.request.retries))
         except MaxRetriesExceededError:
             logger.exception("Max retries exceeded for conversation cleanup, app_id=%s", app_id)
+            # Release lock after max retries
+            if lock_key:
+                try:
+                    redis_client.delete(lock_key)
+                    logger.info("Released lock after max retries: %s", lock_key)
+                except Exception as e:
+                    logger.warning("Failed to release lock %s: %s", lock_key, e)
             raise
 
 

web/app/components/app/log/index.tsx

@@ -107,7 +107,8 @@ const Logs: FC<ILogsProps> = ({ appDetail }) => {
     if (conversationIds) {
       // Remove specific conversation IDs from selection
       setSelectedItems(prev => prev.filter(id => !conversationIds.includes(id)))
-    } else {
+    }
+    else {
       // Clear all selections
       setSelectedItems([])
     }
@@ -129,13 +130,13 @@ const Logs: FC<ILogsProps> = ({ appDetail }) => {
         {total === undefined
           ? <Loading type='app' />
           : total > 0
-            ? <List 
-                logs={isChatMode ? chatConversations : completionConversations} 
-                appDetail={appDetail} 
-                onRefresh={isChatMode ? mutateChatList : mutateCompletionList}
-                selectedItems={selectedItems}
-                onSelectionChange={setSelectedItems}
-              />
+            ? <List
+              logs={isChatMode ? chatConversations : completionConversations}
+              appDetail={appDetail}
+              onRefresh={isChatMode ? mutateChatList : mutateCompletionList}
+              selectedItems={selectedItems}
+              onSelectionChange={setSelectedItems}
+            />
             : <EmptyElement appUrl={`${appDetail.site.app_base_url}${basePath}/${getWebAppType(appDetail.mode)}/${appDetail.site.access_token}`} />
         }
         {/* Show Pagination only if the total is more than the limit */}

web/service/log.ts

@@ -88,6 +88,29 @@ export const clearChatConversations = async ({ appId, conversationIds }: { appId
       ? { conversation_ids: conversationIds }
       : { conversation_ids: [] }
 
+    // Optimistic update: immediately clear the list from UI
+    // This prevents users from thinking the operation failed and clicking again
+    mutate(
+      key =>
+        typeof key === 'object' && key !== null && 'url' in key
+        && key.url === `/apps/${appId}/chat-conversations`,
+      (data: any) => {
+        if (!data)
+          return { data: [], total: 0 }
+        // For selective deletion, filter out deleted conversations
+        if (conversationIds && conversationIds.length > 0) {
+          return {
+            ...data,
+            data: data.data?.filter((conv: any) => !conversationIds.includes(conv.id)) || [],
+            total: Math.max(0, (data.total || 0) - conversationIds.length),
+          }
+        }
+        // For "clear all", return empty list
+        return { data: [], total: 0 }
+      },
+      { revalidate: false },
+    )
+
     const result = await del<any>(`/apps/${appId}/chat-conversations`, { body })
 
     // Clear localStorage to prevent 404 errors
@@ -96,7 +119,7 @@ export const clearChatConversations = async ({ appId, conversationIds }: { appId
     // Clear SWR caches to force reload of conversation lists
     await Promise.all([
       // Clear log list caches (key is an object with url and params)
-      // Force cache invalidation with populateCache: false
+      // Now revalidate after the deletion is confirmed
       mutate(
         key =>
           typeof key === 'object' && key !== null && 'url' in key
@@ -141,6 +164,29 @@ export const clearCompletionConversations = async ({ appId, conversationIds }: {
       ? { conversation_ids: conversationIds }
       : { conversation_ids: [] }
 
+    // Optimistic update: immediately clear the list from UI
+    // This prevents users from thinking the operation failed and clicking again
+    mutate(
+      key =>
+        typeof key === 'object' && key !== null && 'url' in key
+        && key.url === `/apps/${appId}/completion-conversations`,
+      (data: any) => {
+        if (!data)
+          return { data: [], total: 0 }
+        // For selective deletion, filter out deleted conversations
+        if (conversationIds && conversationIds.length > 0) {
+          return {
+            ...data,
+            data: data.data?.filter((conv: any) => !conversationIds.includes(conv.id)) || [],
+            total: Math.max(0, (data.total || 0) - conversationIds.length),
+          }
+        }
+        // For "clear all", return empty list
+        return { data: [], total: 0 }
+      },
+      { revalidate: false },
+    )
+
     const result = await del<any>(`/apps/${appId}/completion-conversations`, { body })
 
     // Clear localStorage to prevent 404 errors
@@ -149,7 +195,7 @@ export const clearCompletionConversations = async ({ appId, conversationIds }: {
     // Clear SWR caches to force reload of conversation lists
     await Promise.all([
       // Clear log list caches (key is an object with url and params)
-      // Force cache invalidation with populateCache: false
+      // Now revalidate after the deletion is confirmed
       mutate(
         key =>
           typeof key === 'object' && key !== null && 'url' in key