diff --git a/api/db/services/file_service.py b/api/db/services/file_service.py index 598a77d9ed..8c80466d7d 100644 --- a/api/db/services/file_service.py +++ b/api/db/services/file_service.py @@ -75,6 +75,15 @@ class FileService(CommonService): files = files.paginate(page_number, items_per_page) res_files = list(files.dicts()) + # Deduplicate by file ID as a safety net against any leftover duplicate rows + # (e.g. duplicate 'skills' or '.knowledgebase' folders created by race conditions). + seen_ids = set() + unique_files = [] + for file in res_files: + if file["id"] not in seen_ids: + seen_ids.add(file["id"]) + unique_files.append(file) + res_files = unique_files for file in res_files: if file["type"] == FileType.FOLDER.value: file["size"] = cls.get_folder_size(file["id"]) @@ -268,7 +277,9 @@ class FileService(CommonService): @classmethod @DB.connection_context() def new_a_file_from_kb(cls, tenant_id, name, parent_id, ty=FileType.FOLDER.value, size=0, location=""): - # Create a new file from dataset + # Create a new file from dataset, or return the existing one. + # Includes deduplication to handle race conditions where concurrent + # requests may have created duplicate entries. # Args: # tenant_id: Tenant ID # name: File name @@ -277,9 +288,22 @@ class FileService(CommonService): # size: File size # location: File location # Returns: - # Created file dictionary - for file in cls.query(tenant_id=tenant_id, parent_id=parent_id, name=name): - return file.to_dict() + # Created or existing file dictionary + existing = list(cls.model.select().where((cls.model.tenant_id == tenant_id) & (cls.model.parent_id == parent_id) & (cls.model.name == name)).order_by(cls.model.create_time.asc())) + if existing: + if len(existing) > 1: + logger.warning( + "Found %d duplicate entries named '%s' under parent %s, keeping only the first", + len(existing), + name, + parent_id, + ) + keep_id = existing[0].id + with DB.atomic(): + for dup in existing[1:]: + File.update(parent_id=keep_id).where(File.parent_id == dup.id).execute() + cls.delete_by_id(dup.id) + return existing[0].to_dict() file = { "id": get_uuid(), "parent_id": parent_id, @@ -297,11 +321,26 @@ class FileService(CommonService): @classmethod @DB.connection_context() def init_skills_folder(cls, root_id, tenant_id): - # Initialize skills folder if not exists + # Initialize skills folder if not exists. + # Deduplicates duplicate entries that may have been created + # by concurrent race conditions (TOCTOU). # Args: # root_id: Root folder ID # tenant_id: Tenant ID - for _ in cls.model.select().where((cls.model.name == SKILLS_FOLDER_NAME) & (cls.model.parent_id == root_id)): + existing = list(cls.model.select().where((cls.model.name == SKILLS_FOLDER_NAME) & (cls.model.parent_id == root_id) & (cls.model.tenant_id == tenant_id)).order_by(cls.model.create_time.asc())) + if existing: + if len(existing) > 1: + logger.warning( + "Found %d duplicate '%s' folders under root %s, keeping only the first", + len(existing), + SKILLS_FOLDER_NAME, + root_id, + ) + keep_id = existing[0].id + with DB.atomic(): + for dup in existing[1:]: + cls.model.update(parent_id=keep_id).where(cls.model.parent_id == dup.id).execute() + cls.delete_by_id(dup.id) return file_id = get_uuid() file = { @@ -319,11 +358,28 @@ class FileService(CommonService): @classmethod @DB.connection_context() def init_knowledgebase_docs(cls, root_id, tenant_id): - # Initialize dataset documents + # Initialize dataset documents. + # Deduplicates duplicate entries that may have been created + # by concurrent race conditions (TOCTOU). # Args: # root_id: Root folder ID # tenant_id: Tenant ID - for _ in cls.model.select().where((cls.model.name == KNOWLEDGEBASE_FOLDER_NAME) & (cls.model.parent_id == root_id)): + existing = list( + cls.model.select().where((cls.model.name == KNOWLEDGEBASE_FOLDER_NAME) & (cls.model.parent_id == root_id) & (cls.model.tenant_id == tenant_id)).order_by(cls.model.create_time.asc()) + ) + if existing: + if len(existing) > 1: + logger.warning( + "Found %d duplicate '%s' folders under root %s, keeping only the first", + len(existing), + KNOWLEDGEBASE_FOLDER_NAME, + root_id, + ) + keep_id = existing[0].id + with DB.atomic(): + for dup in existing[1:]: + cls.model.update(parent_id=keep_id).where(cls.model.parent_id == dup.id).execute() + cls.delete_by_id(dup.id) return folder = cls.new_a_file_from_kb(tenant_id, KNOWLEDGEBASE_FOLDER_NAME, root_id) @@ -479,8 +535,7 @@ class FileService(CommonService): try: if str(doc.kb_id) != str(kb.id): logging.warning( - "Existing document id collision detected for %s: belongs to kb_id=%s, incoming kb_id=%s. " - "Skipping update to avoid cross-KB overwrite.", + "Existing document id collision detected for %s: belongs to kb_id=%s, incoming kb_id=%s. Skipping update to avoid cross-KB overwrite.", doc_id, doc.kb_id, kb.id, @@ -523,7 +578,6 @@ class FileService(CommonService): blob = read_potential_broken_pdf(blob) settings.STORAGE_IMPL.put(kb.id, location, blob) - img = thumbnail_img(filename, blob) thumbnail_location = "" if img is not None: @@ -678,13 +732,14 @@ class FileService(CommonService): written to the log. """ from urllib.parse import urlparse + parsed = urlparse(url) port_suffix = f":{parsed.port}" if parsed.port else "" redacted = f"{parsed.scheme}://{parsed.hostname}{port_suffix}" return assert_url_is_safe(redacted, allowed_schemes=FileService._ALLOWED_SCHEMES) @staticmethod - def upload_info(user_id, file, url: str|None=None): + def upload_info(user_id, file, url: str | None = None): def structured(filename, filetype, blob, content_type): nonlocal user_id if filetype == FileType.PDF.value: @@ -701,7 +756,7 @@ class FileService(CommonService): "mime_type": content_type, "created_by": user_id, "created_at": time.time(), - "preview_url": None + "preview_url": None, } if url: @@ -743,24 +798,17 @@ class FileService(CommonService): host_pins[_next_hostname] = _next_ip current_url = _next_url else: - raise ValueError( - f"Exceeded {_MAX_CRAWL_REDIRECTS} redirects fetching {url!r}" - ) + raise ValueError(f"Exceeded {_MAX_CRAWL_REDIRECTS} redirects fetching {url!r}") # Build a single MAP rule string covering every validated hostname # in the redirect chain. Chromium uses the pinned IP for each, # skipping DNS entirely and eliminating the rebinding window. _map_rules = ",".join(f"MAP {h} {ip}" for h, ip in host_pins.items()) - from crawl4ai import ( - AsyncWebCrawler, - BrowserConfig, - CrawlerRunConfig, - DefaultMarkdownGenerator, - PruningContentFilter, - CrawlResult - ) + from crawl4ai import AsyncWebCrawler, BrowserConfig, CrawlerRunConfig, DefaultMarkdownGenerator, PruningContentFilter, CrawlResult + filename = re.sub(r"\?.*", "", url.split("/")[-1]) + async def adownload(): browser_config = BrowserConfig( headless=True, @@ -768,20 +816,12 @@ class FileService(CommonService): extra_args=[f"--host-resolver-rules={_map_rules}"], ) async with AsyncWebCrawler(config=browser_config) as crawler: - crawler_config = CrawlerRunConfig( - markdown_generator=DefaultMarkdownGenerator( - content_filter=PruningContentFilter() - ), - pdf=True, - screenshot=False - ) + crawler_config = CrawlerRunConfig(markdown_generator=DefaultMarkdownGenerator(content_filter=PruningContentFilter()), pdf=True, screenshot=False) # Use the final resolved URL so the browser starts at the # redirect destination rather than re-following the chain. - result: CrawlResult = await crawler.arun( - url=current_url, - config=crawler_config - ) + result: CrawlResult = await crawler.arun(url=current_url, config=crawler_config) return result + page = asyncio.run(adownload()) if page.pdf: if filename.split(".")[-1].lower() != "pdf": @@ -796,15 +836,16 @@ class FileService(CommonService): @staticmethod def get_files(files: Union[None, list[dict]], raw: bool = False, layout_recognize: str = None) -> Union[list[str], tuple[list[str], list[dict]]]: if not files: - return [] + return [] + def image_to_base64(file): - return "data:{};base64,{}".format(file["mime_type"], - base64.b64encode(FileService.get_blob(file["created_by"], file["id"])).decode("utf-8")) + return "data:{};base64,{}".format(file["mime_type"], base64.b64encode(FileService.get_blob(file["created_by"], file["id"])).decode("utf-8")) + with ThreadPoolExecutor(max_workers=5) as exe: threads = [] imgs = [] for file in files: - if file["mime_type"].find("image") >=0: + if file["mime_type"].find("image") >= 0: if raw: imgs.append(FileService.get_blob(file["created_by"], file["id"])) else: diff --git a/internal/dao/file.go b/internal/dao/file.go index 38c6d514af..9531cc8aa7 100644 --- a/internal/dao/file.go +++ b/internal/dao/file.go @@ -17,6 +17,8 @@ package dao import ( + "fmt" + "log" "ragflow/internal/entity" "strings" @@ -367,21 +369,53 @@ func generateUUID() string { return strings.ReplaceAll(id, "-", "") } +// reparentAndDeleteFolder safely removes a duplicate folder by first +// reparenting any child records to the kept folder, then hard-deleting +// the duplicate row. This prevents orphaned children when cleaning up +// duplicates created by race conditions. +func reparentAndDeleteFolder(dupID, keepID string) error { + // Reparent any child files/folders from the duplicate to the kept folder + if err := DB.Model(&entity.File{}). + Where("parent_id = ?", dupID). + Update("parent_id", keepID).Error; err != nil { + return fmt.Errorf("failed to reparent children from %s to %s: %w", dupID, keepID, err) + } + + // Hard-delete the duplicate folder row + if err := DB.Unscoped().Where("id = ?", dupID).Delete(&entity.File{}).Error; err != nil { + return fmt.Errorf("failed to delete duplicate folder %s: %w", dupID, err) + } + + return nil +} + // DatasetFolderName is the folder name for dataset const DatasetFolderName = ".knowledgebase" -// InitDatasetDocs initializes dataset documents for tenant -// This matches Python's FileService.init_dataset_docs method +// InitDatasetDocs initializes dataset documents for tenant. +// This matches Python's FileService.init_dataset_docs method. +// Deduplicates duplicate entries that may have been created by +// concurrent race conditions (TOCTOU). func (dao *FileDAO) InitDatasetDocs(rootID, tenantID string, file2DocumentDAO *File2DocumentDAO) error { - var count int64 - err := DB.Model(&entity.File{}). - Where("name = ? AND parent_id = ?", DatasetFolderName, rootID). - Count(&count).Error + var existing []*entity.File + err := DB.Where("name = ? AND parent_id = ? AND tenant_id = ?", DatasetFolderName, rootID, tenantID). + Order("create_time ASC"). + Find(&existing).Error if err != nil { return err } - if count > 0 { + if len(existing) > 0 { + if len(existing) > 1 { + log.Printf("[WARN] Found %d duplicate '%s' folders under root %s, keeping only the first", + len(existing), DatasetFolderName, rootID) + keepID := existing[0].ID + for _, dup := range existing[1:] { + if err := reparentAndDeleteFolder(dup.ID, keepID); err != nil { + log.Printf("[ERROR] Failed to deduplicate folder %s: %v", dup.ID, err) + } + } + } return nil } @@ -420,15 +454,26 @@ func (dao *FileDAO) InitDatasetDocs(rootID, tenantID string, file2DocumentDAO *F return nil } -// newAFileFromDataset creates a new file from knowledgebase +// newAFileFromDataset creates a new file from knowledgebase, or returns the existing one. +// Deduplicates duplicate entries that may have been created by race conditions. func (dao *FileDAO) newAFileFromDataset(tenantID, name, parentID string) (*entity.File, error) { var existingFiles []*entity.File - err := DB.Where("tenant_id = ? AND parent_id = ? AND name = ?", tenantID, parentID, name).Find(&existingFiles).Error + err := DB.Where("tenant_id = ? AND parent_id = ? AND name = ?", tenantID, parentID, name).Order("create_time ASC").Find(&existingFiles).Error if err != nil { return nil, err } if len(existingFiles) > 0 { + if len(existingFiles) > 1 { + log.Printf("[WARN] Found %d duplicate entries named '%s' under parent %s, keeping only the first", + len(existingFiles), name, parentID) + keepID := existingFiles[0].ID + for _, dup := range existingFiles[1:] { + if err := reparentAndDeleteFolder(dup.ID, keepID); err != nil { + log.Printf("[ERROR] Failed to deduplicate file entry %s: %v", dup.ID, err) + } + } + } return existingFiles[0], nil } diff --git a/internal/service/file.go b/internal/service/file.go index 4ffd6b93f1..f9ad808171 100644 --- a/internal/service/file.go +++ b/internal/service/file.go @@ -95,6 +95,11 @@ func (s *FileService) ListFiles(tenantID, pfID string, page, pageSize int, order if err := s.initDatasetDocs(pfID, tenantID); err != nil { return nil, fmt.Errorf("failed to initialize dataset docs: %w", err) } + + // Initialize skills folder (matching Python init_skills_folder logic) + if err := s.initSkillsFolder(pfID, tenantID); err != nil { + return nil, fmt.Errorf("failed to initialize skills folder: %w", err) + } } // Check if parent folder exists @@ -114,9 +119,15 @@ func (s *FileService) ListFiles(tenantID, pfID string, page, pageSize int, order return nil, fmt.Errorf("File not found!") } - // Process files to add additional info + // Process files to add additional info, deduplicating by ID as a safety net + // against any leftover duplicate rows (e.g. duplicate 'skills' or '.knowledgebase' folders). fileResponses := make([]map[string]interface{}, 0, len(files)) + seenIDs := make(map[string]struct{}) for _, file := range files { + if _, ok := seenIDs[file.ID]; ok { + continue + } + seenIDs[file.ID] = struct{}{} fileInfo := s.toFileInfo(file) // If folder, calculate size and check for child folders @@ -158,6 +169,47 @@ func (s *FileService) initDatasetDocs(rootID, tenantID string) error { // DatasetFolderName is the folder name for dataset const DatasetFolderName = ".knowledgebase" +// SkillsFolderName is the folder name for skills +const SkillsFolderName = "skills" + +// initSkillsFolder initializes the skills folder under the root folder. +// Deduplicates duplicate entries that may have been created by +// concurrent race conditions (TOCTOU). +func (s *FileService) initSkillsFolder(rootID, tenantID string) error { + existing := s.fileDAO.Query(SkillsFolderName, rootID, tenantID) + if len(existing) > 0 { + if len(existing) > 1 { + common.Logger.Warn(fmt.Sprintf( + "Found %d duplicate '%s' folders under root %s, keeping only the first", + len(existing), SkillsFolderName, rootID, + )) + keepID := existing[0].ID + for _, dup := range existing[1:] { + children, _ := s.fileDAO.ListAllFilesByParentID(dup.ID) + for _, child := range children { + s.fileDAO.UpdateByID(child.ID, map[string]interface{}{"parent_id": keepID}) + } + if delErr := s.fileDAO.Delete(dup.ID); delErr != nil { + common.Logger.Warn(fmt.Sprintf("Failed to delete duplicate skills folder %s: %v", dup.ID, delErr)) + } + } + } + return nil + } + + folder := &entity.File{ + ID: s.generateUUID(), + ParentID: rootID, + TenantID: tenantID, + CreatedBy: tenantID, + Name: SkillsFolderName, + Type: FileTypeFolder, + Size: 0, + SourceType: "", + } + return s.fileDAO.Insert(folder) +} + // FileSourceDataset represents dataset as file source const FileSourceDataset = "knowledgebase" diff --git a/web/src/pages/files/move-dialog.tsx b/web/src/pages/files/move-dialog.tsx index b02c728c7f..4f8d06e1d8 100644 --- a/web/src/pages/files/move-dialog.tsx +++ b/web/src/pages/files/move-dialog.tsx @@ -31,19 +31,22 @@ export function MoveDialog({ hideModal, onOk, loading }: IModalProps) { const ret = await fetchList(id as string); if (ret.code === 0) { setTreeData((tree) => { - return tree.concat( - ret.data.files - .filter((x: IFile) => x.type === 'folder') - .map((x: IFile) => ({ - id: x.id, - parentId: x.parent_id, - title: x.name, - isLeaf: - typeof x.has_child_folder === 'boolean' - ? !x.has_child_folder - : false, - })), - ); + const existingIds = new Set(tree.map((n) => n.id)); + const newNodes = ret.data.files + .filter( + (x: IFile) => + x.type === 'folder' && !existingIds.has(x.id), + ) + .map((x: IFile) => ({ + id: x.id, + parentId: x.parent_id, + title: x.name, + isLeaf: + typeof x.has_child_folder === 'boolean' + ? !x.has_child_folder + : false, + })); + return tree.concat(newNodes); }); } },