zlei6/ragflow
1
0
Fork 0
mirror of https://github.com/infiniflow/ragflow.git synced 2026-07-16 04:37:21 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: prevent duplicate 'skills' and '.knowledgebase' folders caused by race conditions (#16568)

Browse Source
This commit is contained in:
euvre
2026-07-03 12:06:45 +08:00
committed by GitHub
parent 83540185e1
commit 918229613a
4 changed files with 203 additions and 62 deletions
Download Patch File Download Diff File Expand all files Collapse all files

119
api/db/services/file_service.py
View File

@@ -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:

63
internal/dao/file.go
View File

@@ -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
}

54
internal/service/file.go
View File

@@ -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"

29
web/src/pages/files/move-dialog.tsx
View File

@@ -31,19 +31,22 @@ export function MoveDialog({ hideModal, onOk, loading }: IModalProps<any>) {
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);
});
}
},