refactor(shell): run volume.fsck purge once per volume, after all replicas (#9159)

* refactor(shell): run volume.fsck purge once per volume, after all replicas The purge step in findExtraChunksInVolumeServers was nested inside the outer `for dataNodeId` loop, so it fired once per data-node iteration rather than once total. Two consequences: 1. The replica-intersection safety net was broken. The code marks a fid "found in all replicas" only after every replica has reported its orphans, but the purge ran after the first data node already, so fids contributed only by later replicas never got the `true` flag in time. Without `-forcePurging` that meant some legitimate orphans were never purged; with `-forcePurging` the flag was ignored so the bug was hidden. 2. Visible output got noisy: "purging orphan data for volume X..." printed 2-3 times per volume (N_datanodes * N_replicas RPCs to the same locations) since purgeFileIdsForOneVolume already fans out to every replica location via MasterClient.GetLocations. Split the work into two explicit phases: collect orphans from every replica first, then purge each volume once. Drop the per-replica loop around purgeFileIdsForOneVolume since it already handles all replicas internally. Keep the per-replica mark-writable loop (each replica's readonly bit has to be flipped before the purge RPC fans out to it). Also simplify the gating expression — `isSeveralReplicas && foundInAllReplicas` is redundant given the preceding `!isSeveralReplicas` branch — and replace `!(X > 0)` with the more idiomatic `len(X) == 0`. Related to #9116 follow-up on multiple fsck passes needed to fully clean a volume. * address review: per-replica readonly tracking, count-based intersection, defer-per-volume Three issues raised on the v1: 1. The readonly cleanup stored a single isReadOnlyReplicas[volumeId]=bool that flipped true if any replica was read-only, then the defer marked every replica in serverReplicas[volumeId] read-only on exit. If a volume had mixed replica modes (one RO, one RW), the originally-RW replica ended up RO after fsck returned. Track read-only state per replica in readOnlyServerReplicas[volumeId] and revert only those. 2. The defer inside the volumeId loop accumulated for the entire fsck run, so every volume we processed stayed writable until the whole command returned. Split the per-volume logic into purgeOneVolume so the defers unwind between volumes. 3. The intersection logic used a sticky bool that treated "seen on any 2 of 3 replicas" as "seen on all replicas" — a 3+-replica volume would get purged for fids only 2 replicas agreed on, which is what -forcePurging is supposed to opt into. Switch to a count-based map[fid]int compared against volumeReplicaCounts[volumeId], so we only purge without -forcePurging when every replica agrees. Also drop the now-unused serverReplicas map.
2026-08-17 04:36:50 +00:00 · 2026-04-20 15:32:47 -07:00
parent 08a7502b2c
commit e725eb4079
1 changed files with 78 additions and 52 deletions
@@ -386,30 +386,39 @@ func (c *commandVolumeFsck) findFilerChunksMissingInVolumeServers(volumeIdToVInf
 func (c *commandVolumeFsck) findExtraChunksInVolumeServers(dataNodeVolumeIdToVInfo map[string]map[uint32]VInfo, applyPurging bool, modifyFrom, cutoffFrom uint64) error {

 	var totalInUseCount, totalOrphanChunkCount, totalOrphanDataSize uint64
-	volumeIdOrphanFileIds := make(map[uint32]map[string]bool)
-	isSeveralReplicas := make(map[uint32]bool)
+	// map[volumeId]map[fid]replicaCount — counts how many replicas reported
+	// this fid as orphan. A fid is safe to purge without -forcePurging only
+	// when replicaCount == volumeReplicaCounts[volumeId] (i.e. every replica
+	// agrees it's orphan). The previous bool-based tracking treated "seen on
+	// any 2 replicas" as "seen on all replicas", which was wrong for
+	// 3+-replica volumes.
+	volumeIdOrphanFileIds := make(map[uint32]map[string]int)
+	volumeReplicaCounts := make(map[uint32]int)
 	isEcVolumeReplicas := make(map[uint32]bool)
-	isReadOnlyReplicas := make(map[uint32]bool)
-	serverReplicas := make(map[uint32][]pb.ServerAddress)
+	// Track which specific replicas were read-only so we only flip those
+	// back on exit. The old `isReadOnlyReplicas[volumeId] = bool` leaked
+	// read-only state across replicas: if one replica was RO and another RW,
+	// the deferred cleanup would mark the originally-RW replica RO too.
+	readOnlyServerReplicas := make(map[uint32][]pb.ServerAddress)
+	// Phase 1: collect orphan fids from every replica of every volume.
+	// The purge step runs in Phase 2, AFTER every replica has contributed.
+	// Running purge inside this loop (as the original code did) meant the
+	// first replica's orphans were deleted before later replicas could
+	// participate in the intersection — so the "only purge fids seen on
+	// all replicas" safety net only worked by accident, and purge also
+	// fired multiple times per volume (once per data-node iteration).
 	for dataNodeId, volumeIdToVInfo := range dataNodeVolumeIdToVInfo {
 		for volumeId, vinfo := range volumeIdToVInfo {
 			inUseCount, orphanFileIds, orphanDataSize, checkErr := c.oneVolumeFileIdsSubtractFilerFileIds(dataNodeId, volumeId, &vinfo, modifyFrom, cutoffFrom)
 			if checkErr != nil {
 				return fmt.Errorf("failed to collect file ids from volume %d on %s: %v", volumeId, vinfo.server, checkErr)
 			}
-			isSeveralReplicas[volumeId] = false
 			if _, found := volumeIdOrphanFileIds[volumeId]; !found {
-				volumeIdOrphanFileIds[volumeId] = make(map[string]bool)
-			} else {
-				isSeveralReplicas[volumeId] = true
+				volumeIdOrphanFileIds[volumeId] = make(map[string]int)
 			}
+			volumeReplicaCounts[volumeId]++
 			for _, fid := range orphanFileIds {
-				if isSeveralReplicas[volumeId] {
-					if _, found := volumeIdOrphanFileIds[volumeId][fid]; !found {
-						continue
-					}
-				}
-				volumeIdOrphanFileIds[volumeId][fid] = isSeveralReplicas[volumeId]
+				volumeIdOrphanFileIds[volumeId][fid]++
 			}

 			totalInUseCount += inUseCount
@@ -422,54 +431,30 @@ func (c *commandVolumeFsck) findExtraChunksInVolumeServers(dataNodeVolumeIdToVIn
 				}
 			}
 			isEcVolumeReplicas[volumeId] = vinfo.isEcVolume
-			if isReadOnly, found := isReadOnlyReplicas[volumeId]; !(found && isReadOnly) {
-				isReadOnlyReplicas[volumeId] = vinfo.isReadOnly
+			if vinfo.isReadOnly {
+				readOnlyServerReplicas[volumeId] = append(readOnlyServerReplicas[volumeId], vinfo.server)
 			}
-			serverReplicas[volumeId] = append(serverReplicas[volumeId], vinfo.server)
 		}
+	}

+	// Phase 2: purge. At most one call to purgeFileIdsForOneVolume per
+	// volume — that helper already fans out to all replica locations via
+	// MasterClient.GetLocations, so iterating per replica here (as the old
+	// code did) would issue N*N delete RPCs for N replicas.
+	if applyPurging {
 		for volumeId, orphanReplicaFileIds := range volumeIdOrphanFileIds {
-			if !(applyPurging && len(orphanReplicaFileIds) > 0) {
+			if len(orphanReplicaFileIds) == 0 {
 				continue
 			}
-			orphanFileIds := []string{}
-			for fid, foundInAllReplicas := range orphanReplicaFileIds {
-				if !isSeveralReplicas[volumeId] || *c.forcePurging || (isSeveralReplicas[volumeId] && foundInAllReplicas) {
-					orphanFileIds = append(orphanFileIds, fid)
-				}
-			}
-			if !(len(orphanFileIds) > 0) {
-				continue
-			}
-			if *c.verbose {
-				fmt.Fprintf(c.writer, "purging process for volume %d.\n", volumeId)
-			}
-
 			if isEcVolumeReplicas[volumeId] {
 				fmt.Fprintf(c.writer, "skip purging for Erasure Coded volume %d.\n", volumeId)
 				continue
 			}
-			for _, server := range serverReplicas[volumeId] {
-				needleVID := needle.VolumeId(volumeId)
-
-				if isReadOnlyReplicas[volumeId] {
-					err := markVolumeWritable(c.env.option.GrpcDialOption, needleVID, server, true, false)
-					if err != nil {
-						return fmt.Errorf("mark volume %d read/write: %v", volumeId, err)
-					}
-					fmt.Fprintf(c.writer, "temporarily marked %d on server %v writable for forced purge\n", volumeId, server)
-					defer markVolumeWritable(c.env.option.GrpcDialOption, needleVID, server, false, false)
-
-					fmt.Fprintf(c.writer, "marked %d on server %v writable for forced purge\n", volumeId, server)
-				}
-
-				if *c.verbose {
-					fmt.Fprintf(c.writer, "purging files from volume %d\n", volumeId)
-				}
-
-				if err := c.purgeFileIdsForOneVolume(volumeId, orphanFileIds); err != nil {
-					return fmt.Errorf("purging volume %d: %v", volumeId, err)
-				}
+			// Call out to a closure per volume so the deferred "mark
+			// readonly again" fires between volumes instead of piling up
+			// until findExtraChunksInVolumeServers returns.
+			if err := c.purgeOneVolume(volumeId, orphanReplicaFileIds, volumeReplicaCounts[volumeId], readOnlyServerReplicas[volumeId]); err != nil {
+				return err
 			}
 		}
 	}
@@ -494,6 +479,47 @@ func (c *commandVolumeFsck) findExtraChunksInVolumeServers(dataNodeVolumeIdToVIn
 	return nil
 }

+// purgeOneVolume picks the orphan fids to delete for a single volume and
+// fires the delete RPC. It's split out of findExtraChunksInVolumeServers so
+// the `defer markVolumeWritable(..., false, false)` at the bottom fires
+// between volumes — putting that defer inside the caller's for-loop would
+// leave every processed volume writable until the whole fsck run finished.
+func (c *commandVolumeFsck) purgeOneVolume(volumeId uint32, orphanReplicaFileIds map[string]int, replicaCount int, readOnlyReplicas []pb.ServerAddress) error {
+	orphanFileIds := make([]string, 0, len(orphanReplicaFileIds))
+	for fid, foundInReplicaCount := range orphanReplicaFileIds {
+		// Default safety net: only purge fids every replica reported as
+		// orphan. -forcePurging bypasses this for operators who've already
+		// decided they're OK with the single-replica evidence.
+		if foundInReplicaCount == replicaCount || *c.forcePurging {
+			orphanFileIds = append(orphanFileIds, fid)
+		}
+	}
+	if len(orphanFileIds) == 0 {
+		return nil
+	}
+	if *c.verbose {
+		fmt.Fprintf(c.writer, "purging process for volume %d.\n", volumeId)
+	}
+
+	needleVID := needle.VolumeId(volumeId)
+	for _, server := range readOnlyReplicas {
+		if err := markVolumeWritable(c.env.option.GrpcDialOption, needleVID, server, true, false); err != nil {
+			return fmt.Errorf("mark volume %d on %v read/write: %v", volumeId, server, err)
+		}
+		fmt.Fprintf(c.writer, "temporarily marked %d on server %v writable for forced purge\n", volumeId, server)
+		defer markVolumeWritable(c.env.option.GrpcDialOption, needleVID, server, false, false)
+	}
+
+	if *c.verbose {
+		fmt.Fprintf(c.writer, "purging files from volume %d\n", volumeId)
+	}
+
+	if err := c.purgeFileIdsForOneVolume(volumeId, orphanFileIds); err != nil {
+		return fmt.Errorf("purging volume %d: %v", volumeId, err)
+	}
+	return nil
+}
+
 func (c *commandVolumeFsck) collectOneVolumeFileIds(dataNodeId string, volumeId uint32, vinfo VInfo) error {

 	if *c.verbose {