langgenius/dify
1
0
Fork 0
mirror of https://github.com/langgenius/dify.git synced 2026-07-21 12:05:30 -04:00
Code Issues 775 Packages Projects Releases 10 Wiki Activity

Race Condition in Output Moderation Processing #20743

New Issue
Closed
opened 2026-02-21 20:08:46 -05:00 by yindo · 2 comments
yindo commented 2026-02-21 20:08:46 -05:00
Owner
Copy Link
Copy Source

Originally created by @yutingcaicyt on GitHub (Dec 2, 2025).

Problem Description

There's a race condition in the output moderation processing pipeline that can lead to inappropriate content being sent to users without proper filtering due to timing issues between synchronous and asynchronous moderation processes.

Current Behavior

The system employs two-stage moderation:

  1. Asynchronous incremental moderation: An OutputModeration background thread that monitors output chunks in real-time as they are generated
  2. Synchronous final moderation: Performed when QueueMessageEndEvent is received, reviewing the complete message content

The issue occurs in the EasyUIBasedGenerateTaskPipeline class:

  1. When QueueMessageEndEvent is processed:

  2. However, the asynchronous moderation thread (OutputModeration.worker()) may still be running and could discover problematic content after the QueueMessageEndEvent has been processed.

  3. When the async thread later publishes a QueueMessageReplaceEvent (to replace inappropriate content), this event is never processed because the queue listener has already been stopped.

Impact

This race condition can result in:

  • Inappropriate content being delivered to users without filtering
  • QueueMessageReplaceEvent being ignored when published after QueueMessageEndEvent
  • Inconsistent moderation behavior depending on timing
  • Potential compliance violations due to unfiltered content reaching end users

Expected Behavior

All moderation events, including those published after QueueMessageEndEvent, should be properly processed to ensure content compliance before the final response is sent to users. The system should guarantee that both synchronous and asynchronous moderation processes complete before terminating the queue listener.

Suggested Solutions

  1. Delay queue termination: Ensure all asynchronous moderation threads complete their processing before stopping the queue listener
  2. Implement graceful shutdown: Add coordination mechanism between the main processing thread and asynchronous moderation threads
  3. Buffer final events: Allow a grace period after QueueMessageEndEvent to process any pending QueueMessageReplaceEvent events
  4. Modify publishing logic: Update MessageBasedAppQueueManager._publish() to delay stopping the listener until all moderation threads signal completion

Code References

Key files involved:

Additional Context

This issue particularly affects scenarios where:

  • The synchronous moderation times out on long responses
  • Asynchronous moderation detects issues in the final portions of a response
  • Network latency causes delayed moderation results

The problem represents a critical gap in the content safety system that could lead to serious compliance issues.

Originally created by @yutingcaicyt on GitHub (Dec 2, 2025). ## Problem Description There's a race condition in the output moderation processing pipeline that can lead to inappropriate content being sent to users without proper filtering due to timing issues between synchronous and asynchronous moderation processes. ## Current Behavior The system employs two-stage moderation: 1. **Asynchronous incremental moderation**: An [OutputModeration](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/moderation/output_moderation.py#L22-L140) background thread that monitors output chunks in real-time as they are generated 2. **Synchronous final moderation**: Performed when [QueueMessageEndEvent](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/entities/queue_entities.py#L302-L308) is received, reviewing the complete message content The issue occurs in the [EasyUIBasedGenerateTaskPipeline](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py#L65-L525) class: 1. When [QueueMessageEndEvent](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/entities/queue_entities.py#L302-L308) is processed: - The synchronous final moderation is executed on the complete message content - A [MessageEndStreamResponse](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/entities/task_entities.py#L137-L145) is yielded - The queue listener is stopped via [stop_listen()](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/apps/base_app_queue_manager.py#L75-L80) in [MessageBasedAppQueueManager](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/apps/message_based_app_queue_manager.py#L13-L46) 2. However, the asynchronous moderation thread ([OutputModeration.worker()](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/moderation/output_moderation.py#L90-L127)) may still be running and could discover problematic content after the [QueueMessageEndEvent](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/entities/queue_entities.py#L302-L308) has been processed. 3. When the async thread later publishes a [QueueMessageReplaceEvent](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/entities/queue_entities.py#L263-L277) (to replace inappropriate content), this event is never processed because the queue listener has already been stopped. ## Impact This race condition can result in: - Inappropriate content being delivered to users without filtering - [QueueMessageReplaceEvent](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/entities/queue_entities.py#L263-L277) being ignored when published after [QueueMessageEndEvent](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/entities/queue_entities.py#L302-L308) - Inconsistent moderation behavior depending on timing - Potential compliance violations due to unfiltered content reaching end users ## Expected Behavior All moderation events, including those published after [QueueMessageEndEvent](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/entities/queue_entities.py#L302-L308), should be properly processed to ensure content compliance before the final response is sent to users. The system should guarantee that both synchronous and asynchronous moderation processes complete before terminating the queue listener. ## Suggested Solutions 1. **Delay queue termination**: Ensure all asynchronous moderation threads complete their processing before stopping the queue listener 2. **Implement graceful shutdown**: Add coordination mechanism between the main processing thread and asynchronous moderation threads 3. **Buffer final events**: Allow a grace period after [QueueMessageEndEvent](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/entities/queue_entities.py#L302-L308) to process any pending [QueueMessageReplaceEvent](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/entities/queue_entities.py#L263-L277) events 4. **Modify publishing logic**: Update [MessageBasedAppQueueManager._publish()](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/apps/message_based_app_queue_manager.py#L23-L46) to delay stopping the listener until all moderation threads signal completion ## Code References Key files involved: - [core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py) - Main event processing pipeline - [core/app/apps/message_based_app_queue_manager.py](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/app/apps/message_based_app_queue_manager.py) - Queue management with premature stop_listen() call - [core/moderation/output_moderation.py](file:///Users/caiyuting/Desktop/电信相关/xiaonengtisheng/it-zhian-dify/api/core/moderation/output_moderation.py) - Asynchronous moderation thread implementation ## Additional Context This issue particularly affects scenarios where: - The synchronous moderation times out on long responses - Asynchronous moderation detects issues in the final portions of a response - Network latency causes delayed moderation results The problem represents a critical gap in the content safety system that could lead to serious compliance issues.
yindo added the 🐞 bug label 2026-02-21 20:08:46 -05:00
yindo closed this issue 2026-02-21 20:08:46 -05:00
yindo commented 2026-02-21 20:08:46 -05:00
Author
Owner
Copy Link
Copy Source

@dosubot[bot] commented on GitHub (Dec 2, 2025):

You're correct—there is a race condition in the output moderation pipeline. The main event loop in EasyUIBasedGenerateTaskPipeline processes events and, upon receiving QueueMessageEndEvent, performs final moderation and then breaks the loop, sending the final response. However, the asynchronous moderation thread (OutputModeration.worker) may still be running and can publish QueueMessageReplaceEvent after the loop has ended, which means those moderation events are ignored and inappropriate content could slip through. There is currently no explicit synchronization (like joining or waiting for the moderation thread) before the event loop ends, so late moderation events are missed and moderation behavior can be inconsistent (source).

None of the recent PRs or issues address this specific race condition. The best way to mitigate this is to ensure the main event loop either waits for the moderation thread to finish (by joining it) before sending the final response, or to keep the event loop alive until all moderation events are processed. This would guarantee that all QueueMessageReplaceEvent events are handled before the response is finalized (source).

If you need a workaround before a permanent fix, consider adding explicit thread synchronization in the pipeline to wait for the moderation thread to complete before breaking the event loop. This will help ensure moderation is always applied before the message is finalized.

To reply, just mention @dosu.

How did I do? Good | Irrelevant | Incorrect | Verbose | Hallucination | Report 🐛 | Other

Chat with Dosu Open in Cursor Join Discord Share on X

@dosubot[bot] commented on GitHub (Dec 2, 2025): <!-- Answer --> You're correct—there is a race condition in the output moderation pipeline. The main event loop in `EasyUIBasedGenerateTaskPipeline` processes events and, upon receiving `QueueMessageEndEvent`, performs final moderation and then breaks the loop, sending the final response. However, the asynchronous moderation thread (`OutputModeration.worker`) may still be running and can publish `QueueMessageReplaceEvent` after the loop has ended, which means those moderation events are ignored and inappropriate content could slip through. There is currently no explicit synchronization (like joining or waiting for the moderation thread) before the event loop ends, so late moderation events are missed and moderation behavior can be inconsistent [(source)](https://github.com/langgenius/dify/blob/f48522e923369a28a1efaa4e4fa7325229147d0c/api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py#L256-L545). None of the recent PRs or issues address this specific race condition. The best way to mitigate this is to ensure the main event loop either waits for the moderation thread to finish (by joining it) before sending the final response, or to keep the event loop alive until all moderation events are processed. This would guarantee that all `QueueMessageReplaceEvent` events are handled before the response is finalized [(source)](https://github.com/langgenius/dify/blob/f48522e923369a28a1efaa4e4fa7325229147d0c/api/core/moderation/output_moderation.py#L43-L129). If you need a workaround before a permanent fix, consider adding explicit thread synchronization in the pipeline to wait for the moderation thread to complete before breaking the event loop. This will help ensure moderation is always applied before the message is finalized. <!-- Dosu Comment Footer --> *To reply, just mention [@dosu](https://go.dosu.dev/dosubot).* --- <sup>How did I do? [Good](https://app.dosu.dev/response-feedback/853e2ea1-043a-4492-a26d-99ae8f5bc795?feedback_type=great_response) | [Irrelevant](https://app.dosu.dev/response-feedback/853e2ea1-043a-4492-a26d-99ae8f5bc795?feedback_type=irrelevant_answer) | [Incorrect](https://app.dosu.dev/response-feedback/853e2ea1-043a-4492-a26d-99ae8f5bc795?feedback_type=incorrect_sources) | [Verbose](https://app.dosu.dev/response-feedback/853e2ea1-043a-4492-a26d-99ae8f5bc795?feedback_type=too_verbose) | [Hallucination](https://app.dosu.dev/response-feedback/853e2ea1-043a-4492-a26d-99ae8f5bc795?feedback_type=hallucination) | [Report 🐛](https://app.dosu.dev/response-feedback/853e2ea1-043a-4492-a26d-99ae8f5bc795?feedback_type=bug_report) | [Other](https://app.dosu.dev/response-feedback/853e2ea1-043a-4492-a26d-99ae8f5bc795?feedback_type=other)</sup> [![Chat with Dosu](https://dosu.dev/dosu-chat-badge.svg)](https://app.dosu.dev/6088a4f4-073a-43f3-b3d4-4f06fb0c05a0/ask?utm_source=github)&nbsp;[![Open in Cursor](https://dosu.dev/dosu-in-cursor.svg)](https://cursor.com/link/prompt?text=You%27re%20correct%E2%80%94there%20is%20a%20race%20condition%20in%20the%20output%20moderation%20pipeline.%20The%20main%20event%20loop%20in%20%60EasyUIBasedGenerateTaskPipeline%60%20processes%20events%20and%2C%20upon%20receiving%20%60QueueMessageEndEvent%60%2C%20performs%20final%20moderation%20and%20then%20breaks%20the%20loop%2C%20sending%20the%20final%20response.%20However%2C%20the%20asynchronous%20moderation%20thread%20%28%60OutputModeration.worker%60%29%20may%20still%20be%20running%20and%20can%20publish%20%60QueueMessageReplaceEvent%60%20after%20the%20loop%20has%20ended%2C%20which%20means%20those%20moderation%20events%20are%20ignored%20and%20inappropriate%20content%20could%20slip%20through.%20There%20is%20currently%20no%20explicit%20synchronization%20%28like%20joining%20or%20waiting%20for%20the%20moderation%20thread%29%20before%20the%20event%20loop%20ends%2C%20so%20late%20moderation%20events%20are%20missed%20and%20moderation%20behavior%20can%20be%20inconsistent%20%5B%28source%29%5D%28https%3A//github.com/langgenius/dify/blob/f48522e923369a28a1efaa4e4fa7325229147d0c/api/core/app/task_pipeline/easy_ui_based_generate_task_pipeline.py%23L256-L545%29.%0A%0ANone%20of%20the%20recent%20PRs%20or%20issues%20address%20this%20specific%20race%20condition.%20The%20best%20way%20to%20mitigate%20this%20is%20to%20ensure%20the%20main%20event%20loop%20either%20waits%20for%20the%20moderation%20thread%20to%20finish%20%28by%20joining%20it%29%20before%20sending%20the%20final%20response%2C%20or%20to%20keep%20the%20event%20loop%20alive%20until%20all%20moderation%20events%20are%20processed.%20This%20would%20guarantee%20that%20all%20%60QueueMessageReplaceEvent%60%20events%20are%20handled%20before%20the%20response%20is%20finalized%20%5B%28source%29%5D%28https%3A//github.com/langgenius/dify/blob/f48522e923369a28a1efaa4e4fa7325229147d0c/api/core/moderation/output_moderation.py%23L43-L129%29.%0A%0AIf%20you%20need%20a%20workaround%20before%20a%20permanent%20fix%2C%20consider%20adding%20explicit%20thread%20synchronization%20in%20the%20pipeline%20to%20wait%20for%20the%20moderation%20thread%20to%20complete%20before%20breaking%20the%20event%20loop.%20This%20will%20help%20ensure%20moderation%20is%20always%20applied%20before%20the%20message%20is%20finalized.)&nbsp;[![Join Discord](https://img.shields.io/badge/join-5865F2?logo=discord&logoColor=white&label=)](https://go.dosu.dev/discord-bot)&nbsp;[![Share on X](https://img.shields.io/badge/X-share-black)](https://twitter.com/intent/tweet?text=%40dosu_ai%20helped%20me%20solve%20this%20issue!&url=https%3A//github.com/langgenius/dify/issues/29033)
yindo commented 2026-02-21 20:08:46 -05:00
Author
Owner
Copy Link
Copy Source

@crazywoola commented on GitHub (Dec 2, 2025):

Please do not modify the template.

@crazywoola commented on GitHub (Dec 2, 2025): Please do not modify the template.
Sign in to join this conversation.
No Branch/Tag Specified
Branches Tags
Labels
Clear labels
Knowledge Pipeline
Proposal
WIP
cloud
dependencies
feat:webapp
good first issue
outdated
performance
pull-request
Mirrored from GitHub Pull Request
 python:uv
refactor
size:L
stale
status: accepting prs
status: blocked
status: in discussion
stt & tts
vulnerablity
web
wont-fix
🌊 feat:workflow
🌚 invalid
🐞 bug
👻 feat:rag
💪 enhancement
📚 documentation
🔨 feat:tools
🙋‍♂️ question
🙏 help wanted
🤔 cant-reproduce
🤖 feat:agent
No Label 🐞 bug
Milestone
No items
No Milestone
Projects
Clear projects
No project
Assignees
Clear assignees
yindo
No Assignees
1 Participants
Notifications
Due Date
No due date set.
Dependencies

No dependencies set.

Reference: langgenius/dify#20743
Delete Branch "%!s()"

Deleting a branch is permanent. Although the deleted branch may continue to exist for a short time before it actually gets removed, it CANNOT be undone in most cases. Continue?