diff --git a/cmd/erasure-server-pool-recovery_test.go b/cmd/erasure-server-pool-recovery_test.go index d1f1fc6ca..8e58dee29 100644 --- a/cmd/erasure-server-pool-recovery_test.go +++ b/cmd/erasure-server-pool-recovery_test.go @@ -5,8 +5,11 @@ package cmd import ( "bytes" + "context" + "crypto/sha256" "net/http" "net/http/httptest" + "sync/atomic" "testing" "time" ) @@ -83,3 +86,130 @@ func TestPoolsMinorityOverwriteRecovery(t *testing.T) { }) } } + +// Read failures activated by a completed disk mutation verify that a recovery +// ACK is not issued from preflight information alone. +type recoveryReadFaultDisk struct { + StorageAPI + bucket, object string + fail *atomic.Bool + afterRename bool +} + +func (d recoveryReadFaultDisk) ReadXL(ctx context.Context, bucket, object string, data bool) (RawFileInfo, error) { + if bucket == d.bucket && object == d.object && d.fail.Load() { + return RawFileInfo{}, errDiskNotFound + } + return d.StorageAPI.ReadXL(ctx, bucket, object, data) +} +func (d recoveryReadFaultDisk) ReadVersion(ctx context.Context, orig, bucket, object, version string, opts ReadOptions) (FileInfo, error) { + if bucket == d.bucket && object == d.object && d.fail.Load() { + return FileInfo{}, errDiskNotFound + } + return d.StorageAPI.ReadVersion(ctx, orig, bucket, object, version, opts) +} +func (d recoveryReadFaultDisk) RenameData(ctx context.Context, sv, sp string, fi FileInfo, dv, dp string, opts RenameOptions) (RenameDataResp, error) { + res, err := d.StorageAPI.RenameData(ctx, sv, sp, fi, dv, dp, opts) + if err == nil && d.afterRename && dv == d.bucket && dp == d.object { + d.fail.Store(true) + } + return res, err +} + +func TestPoolsRecoveryRefusesUnresolvedPools(t *testing.T) { + for _, mode := range []string{"two-inconsistent", "suspended", "rebalancing", "offline", "read-fails-after-commit"} { + t.Run(mode, func(t *testing.T) { + z, _ := consistencyPools(t) + const object = "refused-recovery" + bucket, router := conditionalPutBucket(t, z, "unversioned") + for _, pool := range z.serverPools { + restore := conditionalPutSwapDisks(pool, object, func(d StorageAPI) StorageAPI { return conditionalPutCapacityDisk{StorageAPI: d} }) + defer restore() + } + base := time.Now().UTC().Add(-time.Hour) + old := bytes.Repeat([]byte{'a'}, 8192) + pending := bytes.Repeat([]byte{'b'}, 8192) + replacement := bytes.Repeat([]byte{'c'}, 8192) + inconsistent := func(pool int) { + t.Helper() + putConsistencyObject(t, z, bucket, object, pool, string(old), ObjectOptions{MTime: base}) + disks := z.serverPools[pool].getHashedSet(object).getDisks() + saved := make([][]byte, len(disks)) + for i, d := range disks { + saved[i] = mustReadNullQuorumMeta(t, d, bucket, object) + } + putConsistencyObject(t, z, bucket, object, pool, string(pending), ObjectOptions{MTime: base.Add(time.Minute)}) + for i := 0; i < len(disks)-1; i++ { + if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, saved[i]); err != nil { + t.Fatal(err) + } + } + } + inconsistent(0) + putConsistencyObject(t, z, bucket, object, 1, string(old), ObjectOptions{MTime: base.Add(-time.Minute)}) + if mode == "two-inconsistent" { + inconsistent(1) + } + // Select a healthy destination if the resolver incorrectly ignores pool0. + if mode != "read-fails-after-commit" { + restore := conditionalPutPool(t, z, object, 1) + defer restore() + } + if mode == "suspended" { + z.poolMetaMutex.Lock() + z.poolMeta.Pools[0].Decommission = &PoolDecommissionInfo{} + z.poolMetaMutex.Unlock() + } + if mode == "rebalancing" { + z.rebalMu.Lock() + z.rebalMeta = &rebalanceMeta{PoolStats: []*rebalanceStats{{Participating: true, Info: rebalanceInfo{Status: rebalStarted}}, {}}} + z.rebalMu.Unlock() + } + var fail atomic.Bool + if mode == "offline" || mode == "read-fails-after-commit" { + fail.Store(mode == "offline") + restore := conditionalPutSwapDisks(z.serverPools[0], object, func(d StorageAPI) StorageAPI { + return recoveryReadFaultDisk{StorageAPI: d, bucket: bucket, object: object, fail: &fail, afterRename: mode == "read-fails-after-commit"} + }) + defer restore() + } + snapshot := func() [][32]byte { + var result [][32]byte + for _, pool := range z.serverPools { + for _, disk := range pool.getHashedSet(object).getDisks() { + b, err := disk.ReadAll(t.Context(), bucket, object+"/"+xlStorageFormatFile) + if err != nil { + t.Fatal(err) + } + result = append(result, sha256.Sum256(b)) + } + } + return result + } + before := snapshot() + req, err := newTestSignedRequestV4(http.MethodPut, getGetObjectURL("", bucket, object), int64(len(replacement)), bytes.NewReader(replacement), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + result := httptest.NewRecorder() + router.ServeHTTP(result, req.WithContext(t.Context())) + if result.Code != http.StatusServiceUnavailable { + t.Errorf("unresolved recovery returned HTTP=%d: %s", result.Code, result.Body.String()) + } + if mode != "read-fails-after-commit" { + after := snapshot() + for i := range before { + if before[i] != after[i] { + t.Errorf("refused preflight modified shard %d", i) + } + } + } else { + fail.Store(false) + got := consistencyRequest(t, router, http.MethodGet, bucket, object, "") + if got.Code != http.StatusOK || !bytes.Equal(got.Body.Bytes(), replacement) { + t.Errorf("read validation failure must not undo committed recovery: HTTP=%d", got.Code) + } + } + }) + } +}