mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-24 07:34:10 +08:00
ea29bcc6b9
* feat: implement Redis job queue and fakeredis support
- Refactored the job queue service to support Redis-backed management for cross-worker scaling.
- Added environment variables for configuration:
- `LANGFLOW_JOB_QUEUE_TYPE=redis`
- `LANGFLOW_REDIS_QUEUE_DB=1`
- Updated job ownership methods to be asynchronous for improved concurrency handling.
- Enhanced Redis cache service with namespacing via key prefixes.
- Introduced `fakeredis` for in-memory Redis simulation in testin>
- Added comprehensive unit tests for Redis job queue components.
* fix: run experimental warning for RedisCache usage only once
- Introduced a mechanism to emit a one-time warning for the RedisCache experimental feature during server runtime.
- The warning is logged only if no other worker has already emitted it, ensuring clarity for users regarding the experimental status of RedisCache.
- The implementation includes a temporary file check to prevent multiple warnings across different processes.
* docs: document environment variables for worker management
- Added documentation for LANGFLOW_GUNICORN_PRELOAD to explain preloading for better performance.
- Detailed the use of LANGFLOW_JOB_QUEUE_TYPE for specifying backends (e.g., Redis).
- Included LANGFLOW_REDIS_QUEUE_DB to define the database index for job queues.
- Updated the "High-Load Environments" guide with these optimal configurations.
* docs: updated 'High-load and multi-worker environments' section
* feat: enhance RedisJobQueueService with consumer wrapper management
- Introduced a caching mechanism for Redis stream consumers to optimize job data retrieval.
- Added methods to manage consumer wrappers, ensuring they are reused across sequential polls.
- Implemented cleanup logic to cancel and clear consumer wrappers during job cleanup and service stop.
- Expanded unit tests to verify consumer wrapper reuse and cleanup behavior.
* fix: ensure Redis keys are deleted during job cleanup even on cancellation
- Updated the cleanup_job method in RedisJobQueueService to guarantee Redis keys are removed even if the job cleanup is interrupted by a CancelledError.
- Added a new unit test to verify that Redis keys are deleted correctly when cleanup is called during task cancellation.
* fix: manage connection check task in RedisJobQueueService
- Added handling for the connection check task in the stop method to ensure it is properly cancelled and awaited if still running.
- This change improves resource management and prevents potential issues during service shutdown.
* fix: handle unpublished sentinel requeue on cancellation in RedisJobQueueService
- Updated the job processing logic to ensure that if a job is cancelled during the xadd operation, the unpublished sentinel is requeued instead of being dropped.
- Introduced a new unit test to verify this behavior, ensuring robustness in job handling during cancellations.
* fix: improve Redis job queue service with enhanced configuration and cleanup
- Added atexit cleanup to remove stale temporary files for RedisCache.
- Refactored Redis job queue service to use shared constants for stream prefixes, improving maintainability.
- Updated type hints for better clarity and consistency in RedisQueueWrapper and RedisJobQueueService.
- Enhanced error handling with configurable backoff for transient read failures.
* fix: enhance Redis job queue service with maxlen configuration for xadd
- Updated the xadd method in RedisJobQueueService to include maxlen and approximate parameters, improving stream management and preventing excessive memory usage.
* fix: enhance job ownership retrieval in RedisJobQueueService
- Updated the get_job_owner method to refresh the Redis key TTL on successful lookups, ensuring long-running jobs maintain their ownership anchor.
- Improved code clarity by extracting the owner key into a variable and adding detailed docstring explanations for better understanding of the TTL management.
* fix: improve job ownership handling in cleanup_job method
- Enhanced the cleanup_job method in RedisJobQueueService to accurately capture job ownership before deleting Redis keys, preventing potential data corruption in multi-worker scenarios.
- Added comments for clarity on ownership logic and its implications during job cleanup.
* fix: optimize TTL management in RedisJobQueueService
- Introduced periodic TTL refresh logic in the _bridge_to_redis method to enhance Redis stream management, reducing round-trips and improving throughput.
- Added constants for TTL refresh events and seconds to maintain clarity and configurability.
- Updated event handling to ensure TTL is refreshed appropriately based on event count and time elapsed.
* fix: improve event handling in flow response management
- Removed unnecessary error handling for missing event tasks in get_flow_events_response, allowing for smoother operation when no task exists.
- Updated create_flow_response to handle optional event_task parameter, ensuring proper cleanup during disconnections.
- Added unit tests to verify behavior when event tasks are missing, enhancing robustness in streaming scenarios.
* fix: enhance RedisQueueWrapper with startup grace period and stream observation
- Added a startup grace period to prevent premature end-of-stream signals when the producer has not yet issued its first XADD.
- Introduced a flag to track whether the stream has been observed, improving the handling of early polling scenarios.
- Updated logic to ensure proper handling of stream existence checks and logging for better debugging during job processing.
* fix: implement cleanup for old cross-worker job queues in RedisJobQueueService
- Added a new method to clean up done cross-worker consumer wrappers that are not owned by the current worker, ensuring proper resource management.
- Enhanced the existing cleanup logic to prevent memory leaks by explicitly pruning stale entries from the consumer wrappers dictionary.
- Improved logging to provide better visibility into the cleanup process for cross-worker jobs.
* fix: enhance job handling in JobQueueService with guarded task execution
- Updated the start_job method to accept a Coroutine type for task_coro, ensuring type safety.
- Introduced a new _guarded_task method to wrap job coroutines, guaranteeing that unhandled exceptions emit an error event and write a sentinel to the Redis Stream, improving reliability in job processing.
- Enhanced documentation to clarify the behavior of the new task handling mechanism and its implications for cross-worker consumers.
* fix: improve client disconnection handling in create_flow_response
- Added logging for scenarios where a client disconnects without an associated event_task, clarifying that the producer will continue running until the build completes.
- Documented the limitation regarding cross-worker passive disconnects and the need for a Redis side-channel for proper cancellation, enhancing observability in the logs.
* fix: enhance RedisQueueWrapper to manage initial read state and buffer behavior
- Introduced a flag to track the completion of the first XREAD call, ensuring proper buffer management during the initial read phase.
- Updated the empty method to reflect the state of the buffer accurately, preventing premature exits from the drain loop until the first read is complete.
- Improved documentation to clarify the behavior of the new flag and its impact on job processing.
* fix: clarify RedisQueueWrapper behavior in tests for buffer state and no-op operations
- Updated test documentation to explain the behavior of the empty() method in relation to the first XREAD completion, ensuring accurate understanding of buffer state during job processing.
- Adjusted assertions in tests to reflect the intended behavior of the RedisQueueWrapper, specifically regarding the no-op nature of put_nowait and its impact on the internal buffer.
* fix: update dependencies in pyproject.toml and uv.lock
- Added fakeredis dependency with a minimum version of 2.0.0 to both pyproject.toml and uv.lock files.
- Ensured proper formatting and comments in pyproject.toml for clarity on onnxruntime version constraints.
* fix: enhance RedisQueueWrapper and job queue handling for improved cancellation and buffer management
- Introduced a structural protocol `_CancellableQueue` to ensure queues can handle cancellation properly during client disconnects.
- Updated `RedisQueueWrapper` to implement this protocol, allowing for graceful cancellation of background tasks.
- Added a maximum size limit to the internal buffer to prevent unbounded memory usage and ensure backpressure on slow consumers.
- Implemented a done callback to handle unexpected fill task crashes, ensuring consumers are not left hanging indefinitely.
- Enhanced unit tests to verify compliance with the new protocol and the behavior of the buffer under various conditions.
* fix: RedisQueueWrapper sentinel delivery, error time-bound, and cancel response
- Deliver end-of-stream sentinel on fill-task cancellation (_on_fill_done now
handles both cancelled and exception paths so consumers are never left hanging)
- Add _error_start time-bound to xread and exists() error loops: after
_STARTUP_GRACE_S seconds of continuous Redis errors the sentinel is delivered
instead of retrying forever
- Advance _last_id cursor only after buffer.put() succeeds so cancellation mid-put
does not silently skip that message in the Redis cursor
- Return False from cancel_flow_build when event_task is None (cross-worker path)
so the HTTP response correctly reports success=False instead of false success
* [autofix.ci] apply automated fixes
---------
Co-authored-by: ogabrielluiz <gabriel@langflow.org>
Co-authored-by: Jordan Frazier <jordan.frazier@datastax.com>
Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>