fix: Sync Functionality for ACLs#931
Conversation
I noticed during onboarding that even once documents were ingested the frontend continued to poll the backend for task status. This is unnecessary, keeps the backend server busy when it doesn't need to be and fills the logs with polling. This addition stops the onboarding cards from polling when they are completed.
I found the page at /admin, it appears to be an abandoned page for uploading files that has been superseded by /knowledge, so I removed it.
Add end-to-end sync support for cloud connectors. Frontend: introduce useSyncConnector/useSyncAllConnectors hooks and wire sync buttons into the knowledge page and actions dropdown with pending state and toast feedback; invalidate search queries after sync. Backend: add get_synced_file_ids_for_connector and change connector_sync to only re-sync files already indexed in OpenSearch (prefer document_id, fallback to filename) to avoid reimporting deleted items; implement sync_all_connectors to batch-sync existing cloud connector files and emit telemetry. Infrastructure: expose DOCUMENT_ID and SOURCE_URL env vars in docker-compose and Helm values/templates and add them to LANGFLOW variables. Langflow flow: update flows/ingestion_flow.json to add secret inputs and dynamic document_id/source_url form fields and adjust node visibility/positions and defaults. This ensures targeted re-syncs of previously indexed cloud files and provides UI controls to trigger syncs.
This comment has been minimized.
This comment has been minimized.
Introduce OPENSEARCH_INDEX_NAME (default: documents) to .env.example and wire it into docker-compose so services (including Langflow) receive the index name; update the integration workflow to watch flows/**. Large documentation revisions: clarify onboarding/setup/prereqs, OAuth redirect URIs and webhook guidance, list supported ingestion file types, consolidate troubleshooting links, and reorganize/configure many environment and Langflow-related settings in the configuration reference. Also include multiple minor frontend/backend updates and add SharePoint utils and TUI startup checks to support connector/startup behavior.
|
Build successful! ✅ |
There was a problem hiding this comment.
Pull request overview
This pull request implements comprehensive sync functionality for cloud storage connectors (Google Drive, OneDrive, SharePoint), enabling users to update both file content and Access Control Lists (ACLs) for previously ingested documents. The changes span backend services, frontend UI components, configuration files, and the Langflow ingestion workflow.
Changes:
- Added new
/connectors/sync-allAPI endpoint and enhanced existing sync endpoints to intelligently sync only previously indexed files (preventing re-sync of deleted files) while updating content and ACLs - Implemented document_id and source_url tracking through the ingestion pipeline, with support for pre-deletion of existing chunks before re-ingestion to prevent duplicates
- Added sync UI controls to the frontend knowledge page and individual file action dropdowns, with task status notifications
Reviewed changes
Copilot reviewed 12 out of 12 changed files in this pull request and generated 6 comments.
Show a summary per file
| File | Description |
|---|---|
| src/api/connectors.py | Added get_synced_file_ids_for_connector() helper and sync_all_connectors() endpoint; modified connector_sync() to use OpenSearch-based filtering |
| src/connectors/service.py | Added filename_filter parameter to sync_connector_files() with comprehensive documentation |
| src/connectors/langflow_connector_service.py | Added pre-deletion logic to remove existing chunks before re-ingestion; passes document_id and source_url to ingestion flow |
| src/services/langflow_file_service.py | Added document_id and source_url parameters to run_ingestion_flow() and passes them as global variables |
| src/main.py | Registered new /connectors/sync-all route with authentication |
| frontend/components/knowledge-actions-dropdown.tsx | Added sync button for cloud connector files with loading states |
| frontend/app/knowledge/page.tsx | Added "Sync" button for syncing all cloud connectors; passes connector_type to action dropdown |
| frontend/app/api/mutations/useSyncConnector.ts | New mutation hooks for syncing individual and all connectors with query invalidation |
| docker-compose.yml | Added DOCUMENT_ID and SOURCE_URL environment variables; updated LANGFLOW_VARIABLES_TO_GET_FROM_ENVIRONMENT list |
| helm/openrag/values.yaml | Updated variablesToGetFromEnvironment to include DOCUMENT_ID and SOURCE_URL |
| helm/openrag/templates/langflow/deployment.yaml | Added DOCUMENT_ID and SOURCE_URL environment variable definitions |
| flows/ingestion_flow.json | Added document_id and source_url fields to form builder; configured nodes to accept these new environment variables; updated node positions and visibility settings |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| onSettled: () => { | ||
| // Invalidate and refetch search queries after a delay to allow sync to complete | ||
| setTimeout(() => { | ||
| queryClient.invalidateQueries({ queryKey: ["search"] }); | ||
| }, 2000); |
There was a problem hiding this comment.
Same issue as above: using a fixed 2-second delay before invalidating queries is fragile and doesn't account for the actual completion of sync operations. Consider implementing proper task status polling instead of using an arbitrary timeout.
| try: | ||
| from config.settings import INDEX_NAME | ||
| opensearch_client = self.session_manager.get_user_opensearch_client(owner_user_id, jwt_token) | ||
| delete_body = {"query": {"term": {"filename": processed_filename}}} |
There was a problem hiding this comment.
The delete query body is constructed manually here, but the codebase has a standard utility function build_filename_delete_body() in utils.opensearch_queries that should be used instead for consistency. Using the utility function ensures all delete-by-query operations follow the same pattern.
| from config.settings import INDEX_NAME | ||
| opensearch_client = self.session_manager.get_user_opensearch_client(owner_user_id, jwt_token) | ||
| delete_body = {"query": {"term": {"filename": processed_filename}}} | ||
| delete_result = await opensearch_client.delete_by_query(index=INDEX_NAME, body=delete_body) |
There was a problem hiding this comment.
The conflicts="proceed" parameter is missing from this delete_by_query call. Other delete operations in the codebase (e.g., src/api/documents.py:81-84) include this parameter to handle concurrent modifications gracefully. Without it, the operation may fail if there are version conflicts during deletion.
| delete_result = await opensearch_client.delete_by_query(index=INDEX_NAME, body=delete_body) | |
| delete_result = await opensearch_client.delete_by_query(index=INDEX_NAME, body=delete_body, conflicts="proceed") |
| "size": 10000 | ||
| } | ||
| }, | ||
| "unique_filenames": { | ||
| "terms": { | ||
| "field": "filename", | ||
| "size": 10000 |
There was a problem hiding this comment.
The aggregation size is hardcoded to 10000, which could be a scalability issue. If a user has more than 10000 unique files from a connector, some files will be silently excluded from the sync operation. Consider using a composite aggregation or pagination approach to handle larger numbers of files, or at minimum add a warning log when the bucket count approaches this limit.
| - FILESIZE=0 | ||
| - SELECTED_EMBEDDING_MODEL=${SELECTED_EMBEDDING_MODEL:-} | ||
| - LANGFLOW_VARIABLES_TO_GET_FROM_ENVIRONMENT=JWT,OPENRAG-QUERY-FILTER,OPENSEARCH_PASSWORD,OPENSEARCH_URL,OPENSEARCH_INDEX_NAME,DOCLING_SERVE_URL,OWNER,OWNER_NAME,OWNER_EMAIL,CONNECTOR_TYPE,FILENAME,MIMETYPE,FILESIZE,SELECTED_EMBEDDING_MODEL,OPENAI_API_KEY,ANTHROPIC_API_KEY,WATSONX_API_KEY,WATSONX_ENDPOINT,WATSONX_PROJECT_ID,OLLAMA_BASE_URL | ||
| - LANGFLOW_VARIABLES_TO_GET_FROM_ENVIRONMENT=JWT,OPENRAG-QUERY-FILTER,OPENSEARCH_PASSWORD,OPENSEARCH_URL,DOCLING_SERVE_URL,OWNER,OWNER_NAME,OWNER_EMAIL,CONNECTOR_TYPE,DOCUMENT_ID,SOURCE_URL,FILENAME,MIMETYPE,FILESIZE,SELECTED_EMBEDDING_MODEL,OPENAI_API_KEY,ANTHROPIC_API_KEY,WATSONX_API_KEY,WATSONX_ENDPOINT,WATSONX_PROJECT_ID,OLLAMA_BASE_URL,OPENSEARCH_INDEX_NAME |
There was a problem hiding this comment.
The order of items in LANGFLOW_VARIABLES_TO_GET_FROM_ENVIRONMENT has changed, with OPENSEARCH_INDEX_NAME moved from an earlier position to the end. While this shouldn't affect functionality (environment variables are looked up by name), it's inconsistent with the ordering in other configuration files. Consider keeping a consistent ordering across all configuration files (helm/openrag/values.yaml has it in the original position) to make maintenance easier.
| - LANGFLOW_VARIABLES_TO_GET_FROM_ENVIRONMENT=JWT,OPENRAG-QUERY-FILTER,OPENSEARCH_PASSWORD,OPENSEARCH_URL,DOCLING_SERVE_URL,OWNER,OWNER_NAME,OWNER_EMAIL,CONNECTOR_TYPE,DOCUMENT_ID,SOURCE_URL,FILENAME,MIMETYPE,FILESIZE,SELECTED_EMBEDDING_MODEL,OPENAI_API_KEY,ANTHROPIC_API_KEY,WATSONX_API_KEY,WATSONX_ENDPOINT,WATSONX_PROJECT_ID,OLLAMA_BASE_URL,OPENSEARCH_INDEX_NAME | |
| - LANGFLOW_VARIABLES_TO_GET_FROM_ENVIRONMENT=JWT,OPENRAG-QUERY-FILTER,OPENSEARCH_PASSWORD,OPENSEARCH_URL,OPENSEARCH_INDEX_NAME,DOCLING_SERVE_URL,OWNER,OWNER_NAME,OWNER_EMAIL,CONNECTOR_TYPE,DOCUMENT_ID,SOURCE_URL,FILENAME,MIMETYPE,FILESIZE,SELECTED_EMBEDDING_MODEL,OPENAI_API_KEY,ANTHROPIC_API_KEY,WATSONX_API_KEY,WATSONX_ENDPOINT,WATSONX_PROJECT_ID,OLLAMA_BASE_URL |
| onSettled: () => { | ||
| // Invalidate and refetch search queries after a delay to allow sync to complete | ||
| setTimeout(() => { | ||
| queryClient.invalidateQueries({ queryKey: ["search"] }); | ||
| }, 2000); |
There was a problem hiding this comment.
Using a fixed 2-second delay before invalidating queries is fragile and doesn't account for the actual completion of sync operations, which can take varying amounts of time. Since the backend returns task_ids for tracking sync progress, consider implementing proper task status polling instead of using an arbitrary timeout. The sync operation could complete in less than 2 seconds (wasting time) or take much longer (showing stale data).
Include allowed_users/allowed_groups in search results and frontend types/UI; update chunks page to display access control. Harden OpenSearch interactions: use filename.keyword in aggregations, safer _source access, and stronger logging. Add knn_vector mapping checks/handling to avoid known OpenSearch k-NN NullPointerException, verify dimensions, and skip or warn on mapping issues. Improve embedding generation with tenacity retries and exponential backoff, add IBM/watsonx rate-limit handling (sequential with delays), and better concurrency controls. Misc: hide long embedding vectors from _source, small logging and stability improvements throughout the multimodel OpenSearch component and search service.
Add support for document ACLs (allowed_users, allowed_groups) end-to-end: expose ALLOWED_USERS and ALLOWED_GROUPS env vars and include them in LANGFLOW_VARIABLES_TO_GET_FROM_ENVIRONMENT (docker-compose.yml); extend the ingestion flow UI to accept dynamic allowed_users/allowed_groups inputs and update node metadata/positions; normalize metadata in OpenSearch ingestion components so allowed_users/allowed_groups arriving as JSON strings are parsed into lists (flows/components/opensearch.py and flows/components/opensearch_multimodel.py). Also import (and comment) vector_store_connection in multimodel component, add/adjust raw_search/tool metadata and update related timestamps/code_hash. Minor frontend, connector, service and helm/value changes to propagate the new ACL fields and related defaults.
Update useSyncAllConnectors and useSyncConnector to remove the delayed setTimeout that invalidated the "search" query and instead immediately call queryClient.invalidateQueries({ queryKey: ["tasks"], exact: false }). This ensures new sync jobs show up in the task list right away instead of waiting for a delayed search refetch. (file: frontend/app/api/mutations/useSyncConnector.ts)
This pull request introduces several improvements and refactorings across configuration, build, testing, and documentation. The most significant changes add support for customizable ports, streamline build and test workflows, and enhance diagnostic capabilities for JWT authentication with OpenSearch. Documentation and environment variable handling are also updated for clarity and flexibility.
Configuration improvements:
FRONTEND_PORTandLANGFLOW_PORTenvironment variables to.env.exampleand updateddocker-compose.ymlto support customizable ports for frontend and Langflow services. This allows users to resolve port conflicts more easily. [1] [2] [3]docker-compose.yml, including new variables likeDOCUMENT_ID,SOURCE_URL,OPENSEARCH_HOST,OPENSEARCH_PORT,OPENSEARCH_URL, andDOCLING_SERVE_URL. TheLANGFLOW_VARIABLES_TO_GET_FROM_ENVIRONMENTlist is updated to include these for improved flexibility.Build and test workflow simplification:
build-multiarch.yml) to useubuntu-latestrunners instead ofubuntu-latest-16-cores, simplifying the build environment and improving compatibility. [1] [2] [3] [4]docker composeinstead ofdocker-composefor improved compatibility and future-proofing.Testing and diagnostics enhancements:
test_jwt_opensearchfunction to theMakefilefor testing JWT authentication against OpenSearch, and integrated it intotest-ci,test-ci-local, and as a standalonetest-os-jwttarget. This provides a more robust and reusable diagnostic tool for JWT issues. [1] [2] [3] [4]factory-resetand other Makefile targets for safer and clearer cleanup operations, including better user prompts and more robust directory and file removal logic.Documentation updates:
_partial-integrate-chat.mdxfile, which contained detailed instructions and code snippets for integrating with Langflow's API, possibly to streamline or replace with updated documentation elsewhere.These changes collectively improve configuration flexibility, build and test reliability, diagnostic tooling, and documentation accuracy.