fix(decom): tolerate offline drives within erasure quorum

Verify migration progress and final source cleanup on every online drive, allowing disk-not-found failures within the configured read quorum. Preserve strict rejection of online metadata remnants, corruption and directory errors. Cover parent and child migration with an offline drive and reject outages beyond parity.

Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/erasure-server-pool-decom.go

@@ -777,7 +777,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
 }
 
 // decommissionPoolWalk walks bucket bi of pool idx once and returns the number
-// of snapshot versions verified absent from every source drive afterwards.
+// of snapshot versions verified absent from every online source drive afterwards.
 func (z *erasureServerPools) decommissionPoolWalk(ctx context.Context, idx int, pool *erasureSets, bi decomBucketInfo) (int64, error) {
 	if err := ctx.Err(); err != nil {
 		return 0, err
@@ -1271,7 +1271,7 @@ func (z *erasureServerPools) removeFreeVersions(ctx context.Context, bucket, obj
 	return nil
 }
 
-// checkAfterDecomBucket verifies that a source bucket is physically empty
+// checkAfterDecomBucket verifies that a source bucket's online drives are empty
 // before it is removed from the resumable decommission queue. ILM decisions
 // are deliberately not treated as an exemption: an expired version remains
 // until the expiry path has performed its logical delete, replication, and
@@ -1285,14 +1285,17 @@ func (z *erasureServerPools) checkAfterDecomBucket(ctx context.Context, idx int,
 		if len(disks) == 0 {
 			return toObjectErr(errErasureReadQuorum, bi.Name)
 		}
+		readQuorum := set.migrationReadQuorum(len(disks))
+		onlineCount := 0
 		for _, disk := range disks {
 			if disk == nil {
-				return toObjectErr(errDiskNotFound, bi.Name)
+				continue
 			}
 			// Quorum listings omit minority versions, and WalkDir can skip
 			// unreadable metadata. Inspect physical directory entries instead:
 			// even a corrupt or truncated xl.meta must keep this pool queued.
 			pending := []string{bi.Prefix}
+			diskOffline := false
 			for len(pending) > 0 {
 				if err := ctx.Err(); err != nil {
 					return err
@@ -1306,6 +1309,10 @@ func (z *erasureServerPools) checkAfterDecomBucket(ctx context.Context, idx int,
 				if errors.Is(err, errFileNotFound) || errors.Is(err, errVolumeNotFound) {
 					continue // A concurrent deletion can remove the directory.
 				}
+				if errors.Is(err, errDiskNotFound) {
+					diskOffline = true
+					break
+				}
 				if err != nil {
 					return err
 				}
@@ -1322,6 +1329,13 @@ func (z *erasureServerPools) checkAfterDecomBucket(ctx context.Context, idx int,
 					}
 				}
 			}
+			if diskOffline {
+				continue
+			}
+			onlineCount++
+		}
+		if onlineCount < readQuorum {
+			return toObjectErr(errErasureReadQuorum, bi.Name)
 		}
 	}
 	return ctx.Err()

cmd/erasure-server-pool-migration-progress_test.go

@@ -257,6 +257,27 @@ func TestDecommissionConcurrentParentDeleteRescansChildren(t *testing.T) {
 	}
 }
 
+func TestDecommissionDegradedPoolRescansChildren(t *testing.T) {
+	z, bucket := migrationTestPools(t)
+	putConsistencyObject(t, z, bucket, "k", 0, "parent", ObjectOptions{})
+	putConsistencyObject(t, z, bucket, "k/b", 0, "child", ObjectOptions{})
+	var wrapped int
+	t.Cleanup(conditionalPutSwapDisks(z.serverPools[0], "k", func(d StorageAPI) StorageAPI {
+		wrapped++
+		if wrapped == 1 {
+			return nil
+		}
+		return d
+	}))
+	if failed := migrationTestWalk(t, z, bucket, 0, "decommission"); failed != 0 {
+		t.Fatalf("failed moves: %d", failed)
+	}
+	mustIAM(t, z.checkAfterDecomBucket(t.Context(), 0, decomBucketInfo{Name: bucket}))
+	if body := migrationTestBody(t, z.GetObjectNInfo, bucket, "k/b", ""); body != "child" {
+		t.Fatalf("child reads %q", body)
+	}
+}
+
 func TestFreeVersionDecommissionRescansChildren(t *testing.T) {
 	f := newFreeVersionTest(t, "")
 	f.bucket = "free-parent"
@@ -407,6 +428,50 @@ func TestPoolsMigrationPhysicalProgress(t *testing.T) {
 	}
 }
 
+func TestCountAbsentMigrationVersionsToleratesConfiguredOfflineDrives(t *testing.T) {
+	makeDisks := func(offlineCount int, asError bool) []StorageAPI {
+		disks := make([]StorageAPI, 4)
+		for i := range disks {
+			disks[i] = migrationProgressRawDisk{err: errFileNotFound, reads: &atomic.Int32{}}
+		}
+		for i := range offlineCount {
+			if asError {
+				disks[i] = migrationProgressRawDisk{err: errDiskNotFound, reads: &atomic.Int32{}}
+			} else {
+				disks[i] = nil
+			}
+		}
+		return disks
+	}
+	snapshot := []FileInfo{{VersionID: mustGetUUID()}}
+	for _, tc := range []struct {
+		name      string
+		offline   int
+		parity    int
+		asError   bool
+		wantCount int
+		wantErr   bool
+	}{
+		{name: "nil-drive-within-parity", offline: 1, parity: 1, wantCount: 1},
+		{name: "read-error-within-parity", offline: 1, parity: 1, asError: true, wantCount: 1},
+		{name: "half-offline-within-parity", offline: 2, parity: 2, wantCount: 1},
+		{name: "offline-beyond-parity", offline: 2, parity: 1, wantErr: true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			disks := makeDisks(tc.offline, tc.asError)
+			er := erasureObjects{
+				setDriveCount:      len(disks),
+				defaultParityCount: tc.parity,
+				getDisks:           func() []StorageAPI { return disks },
+			}
+			got, err := er.countAbsentMigrationVersions(t.Context(), "bucket", "object", snapshot)
+			if got != tc.wantCount || (err != nil) != tc.wantErr {
+				t.Fatalf("removed=%d err=%v, want %d / error=%t", got, err, tc.wantCount, tc.wantErr)
+			}
+		})
+	}
+}
+
 // A successful quorum deletion can leave one source drive behind. It must not
 // count as physical removal or pass the final check, nor cause a retry loop.
 func TestDecommissionQuorumDeletionIsNotPhysicalProgress(t *testing.T) {
@@ -472,12 +537,20 @@ func TestDecommissionFinalVerificationFailsClosed(t *testing.T) {
 		t.Run(state, func(t *testing.T) {
 			z, bucket := migrationTestPools(t)
 			switch state {
-			case "offline-drive", "directory-error":
+			case "offline-drive":
+				set := z.serverPools[0].getHashedSet("k")
+				maxOffline := set.defaultParityCount + 1
+				wrapped := 0
 				t.Cleanup(conditionalPutSwapDisks(z.serverPools[0], "k", func(d StorageAPI) StorageAPI {
-					if state == "directory-error" {
-						return migrationProgressListFaultDisk{StorageAPI: d}
+					wrapped++
+					if wrapped <= maxOffline {
+						return nil
 					}
-					return nil
+					return d
+				}))
+			case "directory-error":
+				t.Cleanup(conditionalPutSwapDisks(z.serverPools[0], "k", func(d StorageAPI) StorageAPI {
+					return migrationProgressListFaultDisk{StorageAPI: d}
 				}))
 			default:
 				// Real metadata on only one drive, without a readable quorum.

cmd/erasure-server-pool-migration.go

@@ -32,7 +32,17 @@ func migrationVersionKey(versionID string) string {
 	return versionID
 }
 
-// countAbsentMigrationVersions counts snapshot versions that no source drive
+// migrationReadQuorum requires enough online drives to intersect every
+// committed write quorum. All of those drives must be checked: a minority
+// remnant still prevents progress. Unknown set geometry tolerates no outages.
+func (er erasureObjects) migrationReadQuorum(diskCount int) int {
+	if er.setDriveCount <= 0 || er.setDriveCount != diskCount || er.defaultParityCount < 0 || er.defaultParityCount > er.setDriveCount/2 {
+		return diskCount
+	}
+	return er.defaultRQuorum()
+}
+
+// countAbsentMigrationVersions counts snapshot versions that no online source drive
 // holds anymore, including free versions. A pool-level NotFound is insufficient:
 // the ordinary read path can return it for inconsistent metadata. Only readable
 // per-drive metadata (or physical absence) proves progress for another walk.
@@ -52,20 +62,31 @@ func (er erasureObjects) countAbsentMigrationVersions(ctx context.Context, bucke
 	if len(disks) == 0 {
 		return 0, toObjectErr(errErasureReadQuorum, bucket, object)
 	}
+	readQuorum := er.migrationReadQuorum(len(disks))
+	offline := make([]bool, len(disks))
 	stored := make([][]FileInfo, len(disks))
 	g := errgroup.WithNErrs(len(disks))
 	for i := range disks {
 		g.Go(func() error {
 			if disks[i] == nil {
-				return errDiskNotFound
+				offline[i] = true
+				return nil
 			}
 			raw, err := disks[i].ReadXL(ctx, bucket, object, false)
+			if errors.Is(err, errDiskNotFound) {
+				offline[i] = true
+				return nil
+			}
 			if errors.Is(err, errFileNotFound) {
 				// ReadXL also returns NotFound for an empty xl.meta, and may
 				// ignore legacy metadata. Confirm physical absence before using
 				// that result as progress. This does not descend into child keys.
 				// Both reads run in parallel across drives, once per object.
 				entries, err := disks[i].ListDir(ctx, "", bucket, object, -1)
+				if errors.Is(err, errDiskNotFound) {
+					offline[i] = true
+					return nil
+				}
 				if err != nil && !errors.Is(err, errFileNotFound) && !errors.Is(err, errVolumeNotFound) {
 					return err
 				}
@@ -101,6 +122,15 @@ func (er erasureObjects) countAbsentMigrationVersions(ctx context.Context, bucke
 			delete(absent, migrationVersionKey(version.VersionID))
 		}
 	}
+	offlineCount := 0
+	for _, unavailable := range offline {
+		if unavailable {
+			offlineCount++
+		}
+	}
+	if len(disks)-offlineCount < readQuorum {
+		return 0, toObjectErr(errErasureReadQuorum, bucket, object)
+	}
 	if err := ctx.Err(); err != nil {
 		return 0, err
 	}