meta: validate entry existence in doBatchUnlink transaction (#6603)

Signed-off-by: jiefenghuang <jiefeng@juicedata.io>
Co-authored-by: jiefenghuang <jiefeng@juicedata.io>

Author: Xuyuchao-juice

pkg/meta/base.go

@@ -1676,6 +1676,7 @@ func (m *baseMeta) Rmdir(ctx Context, parent Ino, name string, skipCheckTrash ..
 	return st
 }
 
+// BatchUnlink delete multiple files in the same directory (case-sensitive filenames)
 func (m *baseMeta) BatchUnlink(ctx Context, parent Ino, entries []*Entry, count *uint64, skipCheckTrash bool) syscall.Errno {
 	if len(entries) == 0 {
 		return 0

pkg/meta/redis.go

@@ -1775,23 +1775,34 @@ func (m *redisMeta) doBatchUnlink(ctx Context, parent Ino, entries []*Entry, len
 			return syscall.EPERM
 		}
 
+		entryKey := m.entryKey(parent)
 		entryInfos = make([]*entryInfo, 0, len(entries))
 		now := time.Now()
-
-		if len(entries) > 0 {
-			for _, entry := range entries {
-				if entry.Attr.Typ == TypeDirectory {
-					continue
-				}
-				info := entryInfo{
-					name:  string(entry.Name),
-					inode: entry.Inode,
-					typ:   entry.Attr.Typ,
-					trash: trash,
-					buf:   m.packEntry(entry.Attr.Typ, entry.Inode),
-				}
-				entryInfos = append(entryInfos, &info)
+		enames := make([]string, 0, len(entries))
+		for _, entry := range entries {
+			enames = append(enames, string(entry.Name))
+		}
+		vals, err := tx.HMGet(ctx, entryKey, enames...).Result()
+		if err != nil {
+			return err
+		}
+		for idx, entry := range entries {
+			val := vals[idx]
+			if val == nil {
+				continue
+			}
+			buf := []byte(val.(string))
+			typ, ino := m.parseEntry(buf)
+			if entry.Inode != ino || typ == TypeDirectory || (entry.Attr != nil && entry.Attr.Typ != typ) {
+				continue
 			}
+			entryInfos = append(entryInfos, &entryInfo{
+				name:  string(entry.Name),
+				inode: ino,
+				typ:   typ,
+				trash: trash,
+				buf:   buf,
+			})
 		}
 
 		inodesSet := make(map[Ino]struct{}, len(entryInfos))
@@ -1906,9 +1917,6 @@ func (m *redisMeta) doBatchUnlink(ctx Context, parent Ino, entries []*Entry, len
 		stats := make(map[string]int64)                     // key -> delta
 
 		for _, info := range entryInfos {
-			if info.typ == TypeDirectory {
-				continue
-			}
 			names = append(names, info.name)
 			if info.attr == nil {
 				continue

pkg/meta/sql.go

@@ -2638,10 +2638,6 @@ func (m *dbMeta) doBatchUnlink(ctx Context, parent Ino, entries []*Entry, length
 	}
 	delNodes := make(map[Ino]*dNode)
 	var totalLength, totalSpace, totalInodes int64
-	if userGroupQuotas != nil {
-		*userGroupQuotas = make([]userGroupQuotaDelta, 0, len(entries))
-	}
-	// main transaction: validate, collect metadata, update inode/link counts, and prepare DB mutations
 	err := m.txn(func(s *xorm.Session) error {
 		pn := node{Inode: parent}
 		ok, err := s.Get(&pn)
@@ -2662,19 +2658,30 @@ func (m *dbMeta) doBatchUnlink(ctx Context, parent Ino, entries []*Entry, length
 		if (pn.Flags&FlagAppend != 0) || (pn.Flags&FlagImmutable) != 0 {
 			return syscall.EPERM
 		}
-		entryInfos = make([]*entryInfo, 0, len(entries))
 		now := time.Now().UnixNano()
+		entryInfos = make([]*entryInfo, 0, len(entries))
+		names := make([][]byte, 0, len(entries))
+		for _, entry := range entries {
+			names = append(names, entry.Name)
+		}
+		var foundEdges []edge
+		if err := s.Where("parent=?", parent).In("name", names).Find(&foundEdges); err != nil {
+			return err
+		}
+		entryMap := make(map[string]*edge)
+		for i := range foundEdges {
+			entryMap[string(foundEdges[i].Name)] = &foundEdges[i]
+		}
 
-		// collect unique inode ids from entries (avoid operating N times on same inode for hard links)
 		inodes := make([]Ino, 0, len(entries))
 		inodeM := make(map[Ino]struct{}) // filter hardlinks
 		for _, entry := range entries {
-			e := &edge{Parent: parent, Name: entry.Name, Inode: entry.Inode}
-			if entry.Attr != nil {
-				if entry.Attr.Typ == TypeDirectory {
-					continue
-				}
-				e.Type = entry.Attr.Typ
+			e, ok := entryMap[string(entry.Name)]
+			if !ok {
+				continue
+			}
+			if e.Inode != entry.Inode || e.Type == TypeDirectory || (entry.Attr != nil && e.Type != entry.Attr.Typ) {
+				continue
 			}
 			entryInfos = append(entryInfos, &entryInfo{e: e, trash: trash})
 			if _, exists := inodeM[entry.Inode]; !exists {
@@ -2766,7 +2773,9 @@ func (m *dbMeta) doBatchUnlink(ctx Context, parent Ino, entries []*Entry, length
 		symlinksDel := make([]Ino, 0)
 		xattrsDel := make([]Ino, 0)
 		edgesIns := make([]interface{}, 0)
-
+		if userGroupQuotas != nil {
+			*userGroupQuotas = make([]userGroupQuotaDelta, 0, len(entries))
+		}
 		// walk each edge to decide whether to move to trash, decrement nlink or delete inode & xattrs
 		for _, info := range entryInfos {
 			edgesDel = append(edgesDel, edge{Parent: parent, Name: info.e.Name})

pkg/meta/tkv.go

@@ -1491,7 +1491,6 @@ func (m *kvMeta) doBatchUnlink(ctx Context, parent Ino, entries []*Entry, length
 				batchUserGroupQuotas = make([]userGroupQuotaDelta, 0, len(batch))
 			}
 			delNodes = make(map[Ino]*dNode)
-			// Get parent directory attribute
 			pbuf := tx.get(m.inodeKey(parent))
 			if pbuf == nil {
 				return syscall.ENOENT
@@ -1510,22 +1509,27 @@ func (m *kvMeta) doBatchUnlink(ctx Context, parent Ino, entries []*Entry, length
 
 			entryInfos = make([]*entryInfo, 0, len(batch))
 			now := time.Now()
-
-			// Collect entry info and filter out directories
-			if len(batch) > 0 {
-				for _, entry := range batch {
-					if entry.Attr.Typ == TypeDirectory {
-						continue
-					}
-					info := entryInfo{
-						name:  string(entry.Name),
-						inode: entry.Inode,
-						typ:   entry.Attr.Typ,
-						trash: trash,
-						buf:   m.packEntry(entry.Attr.Typ, entry.Inode),
-					}
-					entryInfos = append(entryInfos, &info)
+			keys := make([][]byte, 0, len(batch))
+			for _, entry := range batch {
+				keys = append(keys, m.entryKey(parent, string(entry.Name)))
+			}
+			vals := tx.gets(keys...)
+			for idx, entry := range batch {
+				if vals[idx] == nil {
+					continue
+				}
+				typ, ino := m.parseEntry(vals[idx])
+				if ino != entry.Inode || typ == TypeDirectory || (entry.Attr != nil && typ != entry.Attr.Typ) {
+					continue
+				}
+				info := entryInfo{
+					name:  string(entry.Name),
+					inode: ino,
+					typ:   typ,
+					trash: trash,
+					buf:   vals[idx],
 				}
+				entryInfos = append(entryInfos, &info)
 			}
 
 			// Collect unique inodes