diff --git a/weed/shell/command_volume_fsck.go b/weed/shell/command_volume_fsck.go index b2f6ed257..c74c6033d 100644 --- a/weed/shell/command_volume_fsck.go +++ b/weed/shell/command_volume_fsck.go @@ -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 {