fix: user-facing duplicate handling workflow for bucket-based connectors (backport of #2077)#2096
Merged
edwinjosechittilappilly merged 7 commits intoJul 16, 2026
Conversation
…hare all" toggle (#2025) * feature Flag Share all * style: apply biome auto-fixes [skip ci] --------- Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com>
* fix: COS shared-toggle owner not propagated on bucket sync + stale on dedupe skip Two related bugs kept "Make documents available to all users" from actually sharing documents: 1. connector_sync's bucket_filter branch (the path the COS connector UI's "Ingest N Buckets" button always hits) never forwarded body.shared to either of its two sync_specific_files calls (new files / changed files), so newly-ingested files always ended up privately owned regardless of the toggle. 2. ConnectorFileProcessor.process_item's duplicate-filename and unchanged-hash skip branches returned before resolve_shared_owner_fields() ever ran, so re-syncing a bucket whose files were already indexed left their ownership exactly as it was on the prior sync. Fix (2) via _reconcile_shared_owner(): a metadata-only update_by_query scoped to owner==self OR ownerless (same boundary as delete_document_by_filename), called from both skip paths instead of silently leaving stale ownership in place. Also fixes two pre-existing test_shared_flag.py tests that had drifted from connector_sync's current signature (missing request/session/rbac args). * test: guard is_dev_ibm_cos_enabled's default against re-regressing to enabled Found during review: is_dev_ibm_cos_enabled() previously defaulted OPENRAG_DEV_IBM_COS to "true" when unset (now fixed to "false", matching is_dev_azure_blob_enabled's pattern), which made the Enterprise/SaaS-only IBM COS connector available in every deployment by default. The existing tests only monkeypatched the function away, so they wouldn't have caught this. Add a test asserting the unset-env-var default. * style: ruff autofix (auto) --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
Root cause: FileBrowserDialog (the "Browse Files" individual-file picker) never read the "Make documents available to all users" toggle or ingest settings — it built its sync request with only selected_files, so shared silently defaulted to false on the backend. The whole-bucket ("folder") ingest path in SharedBucketView already did this correctly.
Changes (frontend-only, no backend changes needed — it already threads shared uniformly):
- frontend/components/file-browser-dialog.tsx: added showShared/ingestSettings props, and now includes settings and shared in the sync request body, mirroring the bucket-ingest path.
- frontend/components/connectors/shared-bucket-view.tsx: passes its existing showSharedToggle/ingestSettings state down into FileBrowserDialog.
…based connectors (#2077) * fix japanese file names Root cause: COS/Azure/S3 connectors build document.id as the raw, non-ASCII object key ("bucket::報告書.pdf"), and that value was reused verbatim as document_id in the default Langflow ingestion path — the only ingestion path that puts non-ASCII text into ASCII-only HTTP headers and an unbounded-length OpenSearch identifier. Google Drive (opaque ASCII ID) and manual upload (content hash) were structurally immune; COS/Azure/S3 were the only paths exposed. Fix applied (matches the already-proven split used by the non-Langflow ingestion path): - document_index_writer.py — DocumentIndexContext gained a connector_file_id field, now written onto every indexed chunk when present. - langflow_file_service.py — _resolve_document_id now hashes any supplied connector_file_id into a stable ASCII document_id, instead of using it verbatim; connector_file_id threads through run_ingestion_flow → _configure_ingest_callback → the JWT context. - processors.py — the Langflow-path connector branch now passes connector_file_id=document.id (not document_id=document.id), and the post-ingest existence check / metadata sync now match on connector_file_id. - connectors/service.py — stale comments corrected (dedupe/ACL-sync already matched both fields, no logic change needed there). - Added 13 new unit tests (test_resolve_document_id_connector_file_id.py, test_document_index_writer_connector_file_id.py, test_bucket_connector_unicode_file_ids.py) — confirmed they fail on the old code and pass on the fix. - Ran the full unit suite: same 14 pre-existing failures before and after (unrelated to this change — auth/settings/encryption tests), no new regressions. * style: ruff autofix (auto) * fix connector File name issue * ingest fix * Update processors.py * review fix * fix connector dialog on rewrite * overwrite dialog box * Update connectors.py --------- Co-authored-by: autofix-ci[bot] <114827586+autofix-ci[bot]@users.noreply.github.com>
github-actions
Bot
added
frontend
Contributor
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
WalkthroughConnector ingestion now performs duplicate pre-checks for bucket and file selections, supports overwrite or skip handling, propagates shared-ingest settings, queues sync tasks, and retries incompatible OpenSearch aggregations using keyword fields. ChangesConnector ingestion updates
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant User
participant SharedBucketView
participant connector_check_duplicates
participant DuplicateHandlingDialog
participant SyncMutation
User->>SharedBucketView: Start ingestion
SharedBucketView->>connector_check_duplicates: Submit selected files or bucket filter
connector_check_duplicates-->>SharedBucketView: Return duplicate and non-duplicate files
SharedBucketView->>DuplicateHandlingDialog: Show duplicate choices
User->>DuplicateHandlingDialog: Choose overwrite or skip
DuplicateHandlingDialog->>SyncMutation: Submit selected files with replace_duplicates
SyncMutation-->>SharedBucketView: Return task_ids
Possibly related issues
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
Contributor
|
React Doctor found 1 new issue in 1 file · 1 warning · score 84 / 100 (Needs work) · 0 fixed · vs 1 warning
Reviewed by React Doctor for commit |
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
| ingestSettings?: IngestSettings; | ||
| } | ||
|
|
||
| export function FileBrowserDialog({ |
Contributor
There was a problem hiding this comment.
React Doctor · react-doctor/no-giant-component (warning)
Component "FileBrowserDialog" is 363 lines long, which is hard to read & change. Split it into a few smaller components.
Fix → Pull each section into its own component so the parent is easier to read, test, and change.
github-actions
Bot
added
bug
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
🧹 Nitpick comments (3)
enhancements/connectors/ibm_cos/connector.py (1)
71-74: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winRefactor availability gating to use run-mode helpers.
Based on learnings, the repository is migrating away from the legacy
IBM_AUTH_ENABLEDflag for Enterprise/SaaS gating. You should use the run-mode helpers fromutils.run_mode_utilsinstead, updating both the implementation and its tests.
enhancements/connectors/ibm_cos/connector.py#L71-L74: replaceIBM_AUTH_ENABLEDwithis_run_mode_saas() or is_run_mode_on_prem()in theis_availablecheck.tests/unit/connectors/test_ibm_cos_connector.py#L20-L31: update the monkeypatches to mockis_run_mode_saasandis_run_mode_on_preminstead ofIBM_AUTH_ENABLED.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@enhancements/connectors/ibm_cos/connector.py` around lines 71 - 74, The is_available gating in enhancements/connectors/ibm_cos/connector.py#L71-L74 must stop using IBM_AUTH_ENABLED and instead allow availability when is_run_mode_saas() or is_run_mode_on_prem() is true, while preserving the development bypass. Update the corresponding monkeypatches in tests/unit/connectors/test_ibm_cos_connector.py#L20-L31 to mock is_run_mode_saas and is_run_mode_on_prem rather than IBM_AUTH_ENABLED.Source: Learnings
frontend/components/cloud-picker/ingest-settings.tsx (1)
232-246: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove commented-out dead code.
This
Table Structuresection is commented out. Removing it will keep the file clean.♻️ Proposed refactor
- {/* <div className="flex gap-2 items-center justify-between"> - <div> - <div className="text-sm font-semibold pb-2">Table Structure</div> - <div className="text-sm text-muted-foreground"> - Capture table structure during ingest. - </div> - </div> - <Switch - id="table-structure" - checked={currentSettings.tableStructure} - onCheckedChange={(checked) => - handleSettingsChange({ tableStructure: checked }) - } - /> - </div> */}🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@frontend/components/cloud-picker/ingest-settings.tsx` around lines 232 - 246, Remove the commented-out Table Structure JSX block near the ingest settings UI, including its wrapper, descriptive text, Switch, and handlers; leave the surrounding active UI unchanged.src/models/processors.py (1)
839-890: 🚀 Performance & Scalability | 🔵 Trivial | 🏗️ Heavy liftPer-file
update_by_queryon every duplicate/unchanged skip adds write load on every routine re-sync.
_reconcile_shared_ownerunconditionally issues a synchronousupdate_by_queryper filename alias on every file that hits the skip path — which is the common case for a periodic re-sync of an already-fully-ingested bucket. Previously this path was pure read-only; now every unchanged file in a large bucket adds a write (and its OpenSearch-side cost) even when thesharedsetting never changed between syncs. For connectors with large, frequently-resynced buckets this could add meaningful sustained write load.Consider either gating this behind a per-sync "has
sharedchanged since last run" check, batching it into a singleupdate_by_queryper sync run (scoped byconnector_type+user_id) instead of per file, or at least passingwait_for_completion=Falseto avoid blockingprocess_itemon the write.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/models/processors.py` around lines 839 - 890, Reduce write overhead from _reconcile_shared_owner by avoiding synchronous per-alias update_by_query calls on every unchanged or duplicate skip. Prefer batching reconciliation once per sync, scoped to the connector and user, or gate it on detecting that the shared setting changed since the previous run; if retaining per-file updates, make them asynchronous with wait_for_completion=False and ensure process_item is not blocked.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@frontend/components/connectors/shared-bucket-view.tsx`:
- Around line 198-217: The sync request currently sends connection_id from
shared-bucket-view.tsx, but ConnectorSyncBody and the backend sync handler do
not preserve or use it, while file-browser-dialog.tsx omits it. Add
connection_id to ConnectorSyncBody, send the selected connection ID from both
runSelectedFilesSync flows in shared-bucket-view.tsx and
file-browser-dialog.tsx, and update connector_sync to reuse that ID for
duplicate checks and ingest instead of falling back to the first working
connection.
In `@src/api/connectors.py`:
- Around line 774-789: Update the docstring of
_classify_bucket_connector_duplicates to describe existence-only classification:
treat already-existing remote blobs as duplicates and unseen blobs as
non-duplicates, without comparing modified_time or claiming to mirror
modified-time reconciliation. Keep the documented handling of duplicate and new
blobs consistent with the function’s actual return behavior.
- Around line 790-802: Update the file-listing flow around connector.list_files
so it does not mutate the shared connector.bucket_names returned by
connection_manager.get_connector(). Pass bucket_filter through the list_files
API for each paginated request, or create a per-request connector copy, while
preserving pagination and filtering behavior.
- Around line 78-86: Update _is_unmapped_keyword_agg_error to return true only
when the error message contains the exact “Text fields are not optimised”
mapping-error text; remove the broad fielddata substring check so unrelated
fielddata failures are propagated normally.
---
Nitpick comments:
In `@enhancements/connectors/ibm_cos/connector.py`:
- Around line 71-74: The is_available gating in
enhancements/connectors/ibm_cos/connector.py#L71-L74 must stop using
IBM_AUTH_ENABLED and instead allow availability when is_run_mode_saas() or
is_run_mode_on_prem() is true, while preserving the development bypass. Update
the corresponding monkeypatches in
tests/unit/connectors/test_ibm_cos_connector.py#L20-L31 to mock is_run_mode_saas
and is_run_mode_on_prem rather than IBM_AUTH_ENABLED.
In `@frontend/components/cloud-picker/ingest-settings.tsx`:
- Around line 232-246: Remove the commented-out Table Structure JSX block near
the ingest settings UI, including its wrapper, descriptive text, Switch, and
handlers; leave the surrounding active UI unchanged.
In `@src/models/processors.py`:
- Around line 839-890: Reduce write overhead from _reconcile_shared_owner by
avoiding synchronous per-alias update_by_query calls on every unchanged or
duplicate skip. Prefer batching reconciliation once per sync, scoped to the
connector and user, or gate it on detecting that the shared setting changed
since the previous run; if retaining per-file updates, make them asynchronous
with wait_for_completion=False and ensure process_item is not blocked.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 9fc0fa60-1e66-418b-b834-db6f15a32048
📒 Files selected for processing (19)
.env.exampleenhancements/connectors/ibm_cos/connector.pyfrontend/app/api/queries/useGetSettingsQuery.tsfrontend/components/cloud-picker/ingest-settings.tsxfrontend/components/connectors/shared-bucket-view.tsxfrontend/components/file-browser-dialog.tsxfrontend/lib/brand.tssrc/api/connectors.pysrc/api/settings/endpoints.pysrc/api/settings/models.pysrc/config/settings.pysrc/models/processors.pytests/unit/api/test_connector_change_detection.pytests/unit/config/test_show_shared_upload_toggle.pytests/unit/connectors/test_connector_file_type_validation.pytests/unit/connectors/test_ibm_cos_connector.pytests/unit/test_connector_sync_bucket_filter_shared.pytests/unit/test_reconcile_shared_owner.pytests/unit/test_shared_flag.py
Comment on lines
+198
to
217
| const runSelectedFilesSync = ( | ||
| files: BucketDuplicateFile[], | ||
| replaceDuplicates = false, | ||
| ) => { | ||
| syncMutation.mutate( | ||
| { | ||
| onSuccess: (result) => { | ||
| invalidate(); | ||
| if (result.task_ids?.length) { | ||
| // The container path may return two tasks (new files + changed files); | ||
| // track them all. | ||
| trackIngestTasks(result.task_ids); | ||
| onDone(); | ||
| } else { | ||
| toast.info( | ||
| result.message ?? "No files found in the selected buckets.", | ||
| ); | ||
| } | ||
| }, | ||
| onError: (err) => { | ||
| trackProcessFailure({ | ||
| processType: "Ingestion", | ||
| process: "Document Upload", | ||
| category: "Knowledge", | ||
| source: "connector", | ||
| connector_type: connector.type, | ||
| resultValue: err instanceof Error ? err.message : "Sync failed", | ||
| }); | ||
| toast.error(err instanceof Error ? err.message : "Sync failed"); | ||
| connectorType: connector.type, | ||
| body: { | ||
| connection_id: connector.connectionId!, | ||
| selected_files: files, | ||
| settings: ingestSettings, | ||
| replace_duplicates: replaceDuplicates, | ||
| shared: showSharedToggle | ||
| ? (ingestSettings.shared ?? false) | ||
| : undefined, | ||
| }, | ||
| }, | ||
| { onSuccess: onSyncSuccess, onError: onSyncError }, | ||
| ); | ||
| }; |
Contributor
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Confirm whether ConnectorSyncBody/connector_sync ever reads a connection hint,
# and whether multiple active connections per connector_type are supported.
rg -n 'connection_id' src/api/connectors.py
rg -n -A5 'class ConnectorSyncBody' src/api/connectors.pyRepository: langflow-ai/openrag
Length of output: 5067
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the backend sync body and resolution logic around connection_id.
sed -n '600,920p' src/api/connectors.py
# Inspect the two frontend call sites in context.
sed -n '170,235p' frontend/components/connectors/shared-bucket-view.tsx
sed -n '130,210p' frontend/components/file-browser-dialog.tsxRepository: langflow-ai/openrag
Length of output: 16970
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the sync handler and any pydantic config that could change handling of extra fields.
sed -n '840,940p' src/api/connectors.py
rg -n "extra\s*=" src/api/connectors.py src -g'*.py'
rg -n "class Config|model_config" src/api/connectors.py src -g'*.py'Repository: langflow-ai/openrag
Length of output: 4904
Preserve the connection used for duplicate checks in sync requests ConnectorSyncBody still has no connection_id, so connector_sync can fall back to the first working connection and run the overwrite/skip ingest against a different bucket than the one that was previewed. shared-bucket-view.tsx is already sending a field the backend ignores, and file-browser-dialog.tsx doesn’t send one at all; thread connection_id through the sync body and reuse it in the sync handler.
📍 Affects 2 files
frontend/components/connectors/shared-bucket-view.tsx#L198-L217(this comment)frontend/components/file-browser-dialog.tsx#L152-L186
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@frontend/components/connectors/shared-bucket-view.tsx` around lines 198 -
217, The sync request currently sends connection_id from shared-bucket-view.tsx,
but ConnectorSyncBody and the backend sync handler do not preserve or use it,
while file-browser-dialog.tsx omits it. Add connection_id to ConnectorSyncBody,
send the selected connection ID from both runSelectedFilesSync flows in
shared-bucket-view.tsx and file-browser-dialog.tsx, and update connector_sync to
reuse that ID for duplicate checks and ingest instead of falling back to the
first working connection.
Comment on lines
+78
to
+86
| def _is_unmapped_keyword_agg_error(err: Exception) -> bool: | ||
| """True when a terms aggregation failed because the target field is an | ||
| analyzed `text` field without fielddata enabled — the error OpenSearch | ||
| raises for connector_file_id on indices that predate its addition to the | ||
| explicit `keyword` mapping in config/settings.py.""" | ||
| msg = str(err) | ||
| return "Text fields are not optimised" in msg or "fielddata" in msg | ||
|
|
||
|
|
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Narrow the unmapped-agg check. or "fielddata" in msg is too broad and can retry on unrelated fielddata errors; that can still fall through to the outer empty-result path and hide a real failure. Restrict this helper to the actual "Text fields are not optimised" mapping error.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/api/connectors.py` around lines 78 - 86, Update
_is_unmapped_keyword_agg_error to return true only when the error message
contains the exact “Text fields are not optimised” mapping-error text; remove
the broad fielddata substring check so unrelated fielddata failures are
propagated normally.
Comment on lines
+774
to
+789
| async def _classify_bucket_connector_duplicates( | ||
| connector, | ||
| connector_type: str, | ||
| bucket_filter: list[str], | ||
| session_manager, | ||
| user_id: str, | ||
| jwt_token: str | None, | ||
| ) -> dict[str, Any]: | ||
| """Preview a bucket_filter sync: classify remote blobs new/changed/unchanged | ||
| without ingesting anything, mirroring the reconciliation in connector_sync's | ||
| bucket_filter branch. "changed" blobs are reported as duplicates (they would | ||
| overwrite an already-indexed version); "unchanged" blobs are silently | ||
| dropped (the real sync would skip them too); "new" blobs are returned as | ||
| ``non_duplicate_files`` so the caller can sync just those when the user | ||
| chooses to skip duplicates. | ||
| """ |
Contributor
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Docstring contradicts the function's actual (and correct) behavior.
The docstring says this classifies blobs "new/changed/unchanged" mirroring connector_sync's modified-time reconciliation, but the implementation (and the inline comment at lines 821-827) explicitly does existence-only classification, deliberately ignoring modified_time. Future maintainers reading just the docstring will misunderstand the semantics.
📝 Suggested fix
- """Preview a bucket_filter sync: classify remote blobs new/changed/unchanged
- without ingesting anything, mirroring the reconciliation in connector_sync's
- bucket_filter branch. "changed" blobs are reported as duplicates (they would
- overwrite an already-indexed version); "unchanged" blobs are silently
- dropped (the real sync would skip them too); "new" blobs are returned as
- ``non_duplicate_files`` so the caller can sync just those when the user
- chooses to skip duplicates.
- """
+ """Preview a bucket_filter sync: classify remote blobs as duplicate/non-duplicate
+ purely by connector_file_id existence (same semantics as the OAuth connector
+ duplicate check) — NOT the new/changed/unchanged modified-time gate the real
+ bucket_filter sync in connector_sync uses. "duplicate" blobs already exist
+ under this connector_type; "non_duplicate" blobs are returned so the caller
+ can sync just those when the user chooses to skip duplicates.
+ """📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| async def _classify_bucket_connector_duplicates( | |
| connector, | |
| connector_type: str, | |
| bucket_filter: list[str], | |
| session_manager, | |
| user_id: str, | |
| jwt_token: str | None, | |
| ) -> dict[str, Any]: | |
| """Preview a bucket_filter sync: classify remote blobs new/changed/unchanged | |
| without ingesting anything, mirroring the reconciliation in connector_sync's | |
| bucket_filter branch. "changed" blobs are reported as duplicates (they would | |
| overwrite an already-indexed version); "unchanged" blobs are silently | |
| dropped (the real sync would skip them too); "new" blobs are returned as | |
| ``non_duplicate_files`` so the caller can sync just those when the user | |
| chooses to skip duplicates. | |
| """ | |
| async def _classify_bucket_connector_duplicates( | |
| connector, | |
| connector_type: str, | |
| bucket_filter: list[str], | |
| session_manager, | |
| user_id: str, | |
| jwt_token: str | None, | |
| ) -> dict[str, Any]: | |
| """Preview a bucket_filter sync: classify remote blobs as duplicate/non-duplicate | |
| purely by connector_file_id existence (same semantics as the OAuth connector | |
| duplicate check) — NOT the new/changed/unchanged modified-time gate the real | |
| bucket_filter sync in connector_sync uses. "duplicate" blobs already exist | |
| under this connector_type; "non_duplicate" blobs are returned so the caller | |
| can sync just those when the user chooses to skip duplicates. | |
| """ |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/api/connectors.py` around lines 774 - 789, Update the docstring of
_classify_bucket_connector_duplicates to describe existence-only classification:
treat already-existing remote blobs as duplicates and unseen blobs as
non-duplicates, without comparing modified_time or claiming to mirror
modified-time reconciliation. Keep the documented handling of duplicate and new
blobs consistent with the function’s actual return behavior.
Comment on lines
+790
to
+802
| original_buckets = connector.bucket_names | ||
| connector.bucket_names = bucket_filter | ||
| try: | ||
| all_files: list[dict[str, Any]] = [] | ||
| page_token = None | ||
| while True: | ||
| result = await connector.list_files(page_token=page_token) | ||
| all_files.extend(result.get("files", [])) | ||
| page_token = result.get("next_page_token") | ||
| if not page_token: | ||
| break | ||
| finally: | ||
| connector.bucket_names = original_buckets |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Check whether get_connector()/connection_manager.get_connector() caches connector instances
rg -n -A8 'async def get_connector' src/connectors/service.py
rg -n -A15 'def get_connector' src -g '*.py' | grep -A15 'connection_manager'Repository: langflow-ai/openrag
Length of output: 4343
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo '--- connection_manager cache ---'
sed -n '420,500p' src/connectors/connection_manager.py
echo
echo '--- connector_service get_connector ---'
sed -n '1,80p' src/connectors/service.py
echo
echo '--- connector list_files and bucket_names usage ---'
rg -n -A8 -B8 'bucket_names|async def list_files|def list_files' src/connectors -g '*.py'
echo
echo '--- connector_sync bucket_filter branch ---'
rg -n -A30 -B10 'bucket_filter' src/api/connectors.pyRepository: langflow-ai/openrag
Length of output: 40353
Avoid mutating shared connector.bucket_names
connection_manager.get_connector() returns a cached authenticated connector per connection_id, so this try/finally can race with another duplicate-check or sync on the same connection and list files with the wrong bucket_filter. Pass the filter into list_files or use a per-request connector copy.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/api/connectors.py` around lines 790 - 802, Update the file-listing flow
around connector.list_files so it does not mutate the shared
connector.bucket_names returned by connection_manager.get_connector(). Pass
bucket_filter through the list_files API for each paginated request, or create a
per-request connector copy, while preserving pagination and filtering behavior.
mpawlow
reviewed
Jul 15, 2026
mpawlow
left a comment
mpawlow
left a comment
Collaborator
There was a problem hiding this comment.
Code Review 1
- See Major PR comment (1a) for assessment / evaluation
- See Normal PR comments (1b) (1c) (1f) (1g) for consideration
- See Minor PR comments (1d) (1e) (1h) for optional consideration
⚠️ Note: Most of these PR review comments are carried over from PR #2077- ✅ Verified: All relevant changes from PR #2077 are cleanly applied to this PR #2096.
- Given that this PR's changes exactly represents what will be deployed to SaaS, let me know if it makes sense to address the remaining issues in a subsequent PR 👀
| return orphan_ids | ||
|
|
||
|
|
||
| class ConnectorSyncBody(BaseModel): |
Collaborator
There was a problem hiding this comment.
(1a) [Major] connection_id is not threaded through the real sync request, only the duplicate-check preview
Problem
ConnectorCheckDuplicatesBody(used by/check-duplicates) has aconnection_idfield, so the duplicate-check preview runs against a specific connection when the user has more than one active connection of the sameconnector_type.ConnectorSyncBody(used by the actualconnector_syncingest call) has noconnection_idfield at all — it always falls back to "the first connection that actually works."shared-bucket-view.tsxsendsconnection_idin its sync body anyway, but sinceConnectorSyncBodydoesn't declare that field, Pydantic's defaultextra="ignore"behavior silently drops it.file-browser-dialog.tsx'ssubmitSyncdoesn't even attempt to send it.
Background Information
- This is a pre-existing gap in
connector_sync's connection-resolution logic (src/api/connectors.py:1026picks the first authenticating connection), not something newly introduced by the backport — but fix: introduces a user-facing duplicate handling workflow for bucket-based connectors #2077's new duplicate-preview flow is the first place in the codebase that lets a user preview against a specific connection and then act on that preview via a follow-up sync call, so it's the first place the mismatch is user-visible: a user with two S3/COS/Azure connections could see duplicates computed against connection A, choose "overwrite," and have the actual overwrite happen against connection B instead. This was also flagged independently by CodeRabbit on PR fix: user-facing duplicate handling workflow for bucket-based connectors (backport of #2077) #2096.
Code References
src/api/connectors.py:612-629(ConnectorSyncBody— noconnection_id)src/api/connectors.py:632-639(ConnectorCheckDuplicatesBody— hasconnection_id)src/api/connectors.py:1026,1075-1090(connector_syncconnection resolution — first-working fallback only)frontend/components/connectors/shared-bucket-view.tsx:180,206(sendsconnection_idthat the backend ignores)frontend/components/file-browser-dialog.tsx:152-186(submitSync— noconnection_idsent at all)
Potential Solution
- Add
connection_id: str | None = NonetoConnectorSyncBody, and inconnector_syncpreferbody.connection_id(mirroring theif body.connection_id: ... else: first-workingpattern already inconnector_check_duplicates) before falling back to "first working connection."
Alternative Solutions
- Have the duplicate-check response echo back the resolved
connection_id, and require the frontend to pass it through unmodified on the follow-up sync call (makes the coupling explicit at the API boundary).
Collaborator
Author
There was a problem hiding this comment.
@coderabbitai can you create an issue wrt to the review comments above and assign it to me and @mpawlow
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
repo="langflow-ai/openrag"
body_file="$(mktemp)"
trap 'rm -f "$body_file"' EXIT
cat >"$body_file" <<'EOF'
## Summary
Ensure that the connector used for duplicate preview is also used for the subsequent ingestion request.
Today, `ConnectorCheckDuplicatesBody` accepts `connection_id`, so `/check-duplicates` can inspect a specific active connection. The real `connector_sync` endpoint, however, does not declare or honor `connection_id`; it resolves the first active connection that authenticates. Because Pydantic ignores unknown fields by default, a `connection_id` sent by the bucket UI is silently discarded. The file-browser flow does not send it at all.
With multiple active connections of the same connector type, users can preview duplicates from connection A and then overwrite or ingest files from connection B.
## Required changes
- Add an optional `connection_id` field to `ConnectorSyncBody` in `src/api/connectors.py`.
- Update `connector_sync` connection resolution to prefer and validate `body.connection_id`, matching the behavior of `connector_check_duplicates`; retain the first-working-connection behavior only as the fallback when it is omitted.
- Thread the resolved/selected connection ID through both follow-up sync flows:
- `frontend/components/connectors/shared-bucket-view.tsx`
- `frontend/components/file-browser-dialog.tsx`
- Add regression coverage for multiple active connections, including that a supplied connection ID is used for the actual sync and that the duplicate preview and sync target the same connection.
## Rationale
Duplicate handling must act on the source that was previewed. Otherwise, a user can make an overwrite/skip decision based on one bucket connection while the application ingests from another connection of the same type.
## Acceptance criteria
- A sync request with a valid `connection_id` uses that exact active, authenticated connection.
- An invalid, inactive, or unauthenticated requested connection produces an appropriate error rather than silently switching to another connection.
- When no `connection_id` is provided, existing first-working-connection fallback behavior remains intact.
- The bucket shared-ingest and file-browser duplicate workflows send the same connection ID to both duplicate checking and sync.
- Automated tests cover the multi-connection scenario.
## Backlinks
- PR: https://github.com/langflow-ai/openrag/pull/2096
- Review discussion: https://github.com/langflow-ai/openrag/pull/2096#discussion_r3587743320
- Requested by: `@edwinjosechittilappilly`
EOF
gh issue create \
--repo "$repo" \
--title "Preserve connector connection_id from duplicate preview through sync" \
--body-file "$body_file" \
--assignee edwinjosechittilappilly \
--assignee mpawlowLength of output: 2989
@edwinjosechittilappilly, created and assigned the follow-up issue to you and @mpawlow:
#2111 — Preserve connector connection_id from duplicate preview through sync
| ``non_duplicate_files`` so the caller can sync just those when the user | ||
| chooses to skip duplicates. | ||
| """ | ||
| original_buckets = connector.bucket_names |
Collaborator
There was a problem hiding this comment.
(1b) [Normal] Duplicate-check preview mutates a cached, shared connector's bucket_names without per-request isolation
Problem
_classify_bucket_connector_duplicatestemporarily overwritesconnector.bucket_names = bucket_filter, callsconnector.list_files(...), then restores it in afinallyblock.connector_service.get_connector(connection_id)(viaconnection_manager) returns a cached connector instance keyed byconnection_id(self.active_connectors[connection_id]), not a fresh instance per request. If two requests for the same connection race (e.g. two browser tabs, or a preview check overlapping with a real sync), one request's temporarybucket_namesmutation can leak into the other'slist_files()call, causing it to list the wrong buckets.
Background Information
- This mutate/restore-in-
finallypattern is not unique to the new code — the same pattern already exists inconnector_sync's ownbucket_filterbranch (src/api/connectors.py:1198-1209) and elsewhere in the file (e.g.:2381-2386), so this PR extends an existing architectural shortcut rather than introducing a new one.
Code References
src/api/connectors.py:790-801(_classify_bucket_connector_duplicates)src/api/connectors.py:1198-1209(same pattern inconnector_sync, pre-existing)src/connectors/connection_manager.py:431-440(get_connectorconnector caching byconnection_id)
Potential Solution
- Pass
bucket_filteras a parameter intoconnector.list_files(bucket_filter=..., page_token=...)instead of mutating shared instance state, so no shared mutable state is touched across concurrent requests.
Alternative Solutions
- Acquire a per-
connection_idasyncio lock around the mutate/restore window as a smaller, lower-risk interim fix if changing thelist_filessignature is out of scope for this backport.
| return connector_type in ["google_drive", "sharepoint", "onedrive"] | ||
|
|
||
|
|
||
| def _is_unmapped_keyword_agg_error(err: Exception) -> bool: |
Collaborator
There was a problem hiding this comment.
(1c) [Normal] _is_unmapped_keyword_agg_error's broad substring match can mask unrelated aggregation failures
Problem
_is_unmapped_keyword_agg_errorreturnsTrueon"Text fields are not optimised" in msg or "fielddata" in msg. The bare"fielddata"substring check is broader than the specific legacy-mapping error this helper is meant to detect, so an unrelated OpenSearch error that happens to mention "fielddata" would also trigger the.keyword-field retry path instead of surfacing as a real failure.
Background Information
- CodeRabbit flagged this as Major; but it is Normal because the retry re-runs the same query with a renamed field — if the actual problem is unrelated, the retry itself will very likely still raise (there's no swallowing
exceptaround the retry call), so the most common failure mode is a wasted retry rather than a silently wrong result. The narrower risk is a transient/unrelated fielddata error that happens to succeed on retry, returning results computed against a re-issued query that may not reflect the original triggering condition.
Code References
src/api/connectors.py:78-84(_is_unmapped_keyword_agg_error)src/api/connectors.py:127-137(first call site / retry)src/api/connectors.py:291-297(second call site / retry)
Potential Solution
- Narrow the check to the specific mapping-error text (
"Text fields are not optimised"/ the associated OpenSearch error type), dropping the standalone"fielddata"substring match.
Alternative Solutions
- Match on the specific OpenSearch exception type/
error.typefield (e.g.illegal_argument_exceptionwith acaused_bymapping-related reason) instead of a raw string substring match, which is generally more robust against message-text drift across OpenSearch versions.
| } | ||
|
|
||
|
|
||
| async def _classify_bucket_connector_duplicates( |
Collaborator
There was a problem hiding this comment.
(1d) [Minor] Docstring on _classify_bucket_connector_duplicates contradicts its actual (correct) behavior
Problem
- The docstring claims the function classifies blobs as "new/changed/unchanged" mirroring
connector_sync's modified-time reconciliation. The implementation — and its own inline comment a few lines below — explicitly does existence-only classification (any blob whose ID already exists is a "duplicate", regardless of modification time). A future maintainer reading only the docstring will misunderstand the semantics.
Code References
src/api/connectors.py:774-789(docstring)src/api/connectors.py:820-827(inline comment describing the actual, correct behavior)
Potential Solution
- Update the docstring to describe existence-only classification, as CodeRabbit's suggested diff does — align it with the inline comment already in the function body.
| chooses to skip duplicates. | ||
| """ | ||
| original_buckets = connector.bucket_names | ||
| connector.bucket_names = bucket_filter |
Collaborator
There was a problem hiding this comment.
(1f) [Normal] Existence-based duplicate check triggers the confirm dialog (and a full re-embed) on every routine bucket re-sync, even with zero remote changes
Problem
- Because
_classify_bucket_connector_duplicatesflags any previously-ingested blob as a duplicate (see 1d), re-clicking "Ingest" on a bucket that was already fully synced — with no changes at the source — always surfaces the overwrite dialog. If the user picks "Overwrite duplicates" (the more prominent, styled button), every unchanged file gets re-ingested/re-embedded unnecessarily.
Background Information
- Identified in an earlier code review of the source PR fix: introduces a user-facing duplicate handling workflow for bucket-based connectors #2077 (thread still unresolved/not outdated there) and independently re-confirmed here as still present in fix: user-facing duplicate handling workflow for bucket-based connectors (backport of #2077) #2096's code, since
_classify_bucket_connector_duplicatesis byte-identical between the two PRs. Contrast with the pre-existingbucket_filterreconciliation inconnector_sync(src/api/connectors.py:1198-1238), which already does propermodified_time-based new/changed/unchanged classification and cleanly no-ops when nothing changed. This turns what should be a rare, meaningful prompt (real conflicting content) into dialog fatigue for the routine "re-sync my bucket" workflow, and risks users reflexively clicking "Overwrite" out of habit, wasting embedding/OCR compute on unchanged documents.
Code References
src/api/connectors.py:790-827(existence-based classification)src/api/connectors.py:1198-1238(existing, correct modified-time-based reconciliation it should mirror)
Potential Solution
- Have
_classify_bucket_connector_duplicatesreuse the existingmodified_time-based reconciliation helper already used inconnector_sync'sbucket_filterbranch, and only surface true "changed" blobs asduplicate_files; drop "unchanged" blobs from both lists so the dialog only appears when there's a real conflict.
|
|
||
| const checkData = | ||
| (await checkResponse.json()) as BucketDuplicateCheckResponse; | ||
| const duplicateNames = checkData.duplicate_names || []; |
Collaborator
There was a problem hiding this comment.
(1g) [Normal] Redundant full bucket listing on every "no duplicates" ingest
Problem
ingestSelectedinSharedBucketViewcallscheck-duplicates, which lists every object across the selected buckets (_classify_bucket_connector_duplicates, unbounded pagination). WhenduplicateCount === 0, the frontend discards that listing and callsrunBucketSync(), which re-triggersconnector_sync'sbucket_filterbranch — listing the same buckets from scratch again. For large buckets this doubles listing latency/cost on every ingest click, when the check step already has the completenon_duplicate_fileslist in hand.
Background Information
- Identified in an earlier code review of the source PR fix: introduces a user-facing duplicate handling workflow for bucket-based connectors #2077 (thread still unresolved/not outdated there); re-verified here that
runBucketSync()is still called unconditionally on theduplicateCount === 0path in fix: user-facing duplicate handling workflow for bucket-based connectors (backport of #2077) #2096'sshared-bucket-view.tsx:260-261, andnon_duplicate_filesis available oncheckDatabut unused in that branch.
Potential Solution
- When
duplicateCount === 0, callrunSelectedFilesSync(checkData.non_duplicate_files, false)directly instead ofrunBucketSync(), avoiding the second full listing.
| duplicateFiles: BucketDuplicateFile[]; | ||
| nonDuplicateFiles: BucketDuplicateFile[]; | ||
| } | null>(null); | ||
| const isOverwriteConfirmedRef = useRef(false); |
Collaborator
There was a problem hiding this comment.
(1h) [Minor] Fragile ref-based workaround for dialog close/overwrite race, with no test coverage
Problem
isOverwriteConfirmedRefexists to suppresshandleDuplicateDialogOpenChange's "skip" branch from re-firing whenDuplicateHandlingDialog's overwrite handler callsonOpenChange(false)right afteronOverwrite(). It works, but depends onpendingDuplicatesstill being non-null in a stale closure at the momentonOpenChange(false)fires synchronously afteronOverwrite(), before React flushes the state update clearing it. Any future change toDuplicateHandlingDialog's call order could silently break this, and no frontend/e2e test exercises either the overwrite or skip path.
Background Information
- Identified in an earlier code review of the source PR fix: introduces a user-facing duplicate handling workflow for bucket-based connectors #2077 (thread still unresolved/not outdated there); re-verified here that
isOverwriteConfirmedRefis still present and used identically in fix: user-facing duplicate handling workflow for bucket-based connectors (backport of #2077) #2096'sshared-bucket-view.tsx:97,284,292,296.
Potential Solution
- Have a single source of truth for "why did the dialog close" — e.g., have the overwrite handler own closing the dialog and clearing
pendingDuplicatesin one state update, and have the dialog's own Cancel/backdrop-close path call a distinctonSkipcallback rather than overloadingonOpenChange. Add a Playwright test covering: no-duplicates auto-sync, duplicates→overwrite, duplicates→skip, duplicates→dismiss-without-choice.
2 tasks
mpawlow
approved these changes
Jul 16, 2026
…te-handling-bucket-connectors # Conflicts: # frontend/components/connectors/shared-bucket-view.tsx # frontend/components/file-browser-dialog.tsx # src/api/connectors.py
github-actions
Bot
added
bug
github-actions
Bot
added
bug
github-actions
Bot
deleted the
backport/2077-duplicate-handling-bucket-connectors
branch
July 16, 2026 19:25
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Backports #2077 from
release-saas-ga-0.6.2tomain.Depends on #2095 (backport of #2066), #2094 (#2050), #2093 (#2035), #2092 (#2025) — stacked because it substantially rewrites the ingest flow in
frontend/components/file-browser-dialog.tsxandfrontend/components/connectors/shared-bucket-view.tsx, both touched by earlier backports. Merge order: #2092 → #2093 → #2094 → #2095 → this PR.Note on conflict resolution:
mainhas anonIngestSuccesscallback prop onFileBrowserDialog(added independently of the release branch) that release's version of this file doesn't have. Resolved by taking release's file (duplicate-check flow) as the base and re-threadingonIngestSuccessthroughsubmitSyncso it still fires with the mutation result on successful ingest, matching its original main behavior. Verified withtsc --noEmit— no errors in either touched file.Test plan
onIngestSuccessstill fires (task tracking / invalidation) on successSummary by CodeRabbit