diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index 43f9c54b2..c093bdfab 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -935,6 +935,16 @@ func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object s if err != nil { return FileInfo{}, nil, nil, time.Time{}, "", err } + for i, meta := range metaArr { + if errs[i] == nil && meta.IsValid() && meta.VersionID == fi.VersionID && meta.ModTime.After(fi.ModTime) { + // A quorum of an older generation does not resolve a later + // overwrite of the same version. Keep the uncertainty visible + // instead of silently rolling the object back on reads. + return FileInfo{}, nil, nil, time.Time{}, "", InsufficientReadQuorum{ + Err: errErasureReadQuorum, Type: RQInconsistentMeta, + } + } + } onlineMeta := make([]FileInfo, len(metaArr)) for i, disk := range onlineDisks { diff --git a/cmd/erasure-read-generation_test.go b/cmd/erasure-read-generation_test.go index bdb9c2c08..6ba917b6b 100644 --- a/cmd/erasure-read-generation_test.go +++ b/cmd/erasure-read-generation_test.go @@ -58,9 +58,20 @@ func TestPoolsReadGenerationConflictDoesNotExposeOldValue(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() router := initTestAPIEndPoints(z, []string{"GetObject", "HeadObject"}) - for broken := range 2 { - t.Run(fmt.Sprintf("broken-pool-%d", broken), func(t *testing.T) { - object := fmt.Sprintf("conflict-%d", broken) + for _, tc := range []struct { + pool int + mode string + }{ + {0, "split"}, + {1, "split"}, + {0, "older-quorum"}, + {1, "older-quorum"}, + {0, "newer-quorum"}, + {1, "newer-quorum"}, + } { + t.Run(fmt.Sprintf("%s/pool-%d", tc.mode, tc.pool), func(t *testing.T) { + broken := tc.pool + object := fmt.Sprintf("conflict-%s-%d", tc.mode, broken) base := UTCNow().Add(-time.Hour) putConsistencyObject(t, z, bucket, object, 1-broken, "old-acknowledged", ObjectOptions{MTime: base}) set := z.serverPools[broken].getHashedSet(object) @@ -70,7 +81,20 @@ func TestPoolsReadGenerationConflictDoesNotExposeOldValue(t *testing.T) { putConsistencyObject(t, z, bucket, object, broken, fmt.Sprintf("newer-%d", generation), ObjectOptions{MTime: base.Add(time.Duration(generation+1) * time.Minute)}) for i, disk := range disks { - if i%4 != generation { + g := i % 4 + switch tc.mode { + case "older-quorum": + g = 0 + if i >= len(disks)-3 { + g = i - (len(disks) - 3) + 1 + } + case "newer-quorum": + g = 3 + if i < 3 { + g = i + } + } + if g != generation { continue } buf, err := disk.ReadAll(ctx, bucket, pathJoin(object, xlStorageFormatFile)) @@ -88,8 +112,15 @@ func TestPoolsReadGenerationConflictDoesNotExposeOldValue(t *testing.T) { for _, version := range []string{"", nullVersionID} { for _, method := range []string{http.MethodHead, http.MethodGet} { rec := consistencyRequest(t, router, method, bucket, object, version) - if rec.Code != http.StatusServiceUnavailable { - t.Errorf("%s version=%q exposed an old value or false absence: status=%d body=%q", method, version, rec.Code, rec.Body.String()) + want := http.StatusServiceUnavailable + if tc.mode == "newer-quorum" { + want = http.StatusOK + } + if rec.Code != want { + t.Errorf("%s version=%q: want status=%d, got=%d body=%q", method, version, want, rec.Code, rec.Body.String()) + } + if want == http.StatusOK && method == http.MethodGet && rec.Body.String() != "newer-3" { + t.Errorf("latest quorum did not return its payload: %q", rec.Body.String()) } } } diff --git a/cmd/object-retention-overwrite_test.go b/cmd/object-retention-overwrite_test.go index b6a910cba..83ce43ebe 100644 --- a/cmd/object-retention-overwrite_test.go +++ b/cmd/object-retention-overwrite_test.go @@ -244,6 +244,26 @@ func TestPoolsObjectLockOverwrite(t *testing.T) { } } defer conditionalPutPool(t, z, key, 0)() + var source string + var copyInfo ObjectInfo + if op == "copy" || op == "self-copy" { + source = key + if op == "copy" { + source = "source-" + state + putConsistencyObject(t, z, bucket, source, 0, "replacement", ObjectOptions{Versioned: true}) + } + // Resolve the source before the pool becomes unreadable. + // The destination must still reject a previously read source. + gr, gerr := z.GetObjectNInfo(ctx, bucket, source, nil, nil, ObjectOptions{}) + if gerr != nil { + t.Fatal(gerr) + } + defer gr.Close() + copyInfo = gr.ObjInfo + copyInfo.metadataOnly = op == "self-copy" + copyInfo.PutObjReader = mustGetPutObjReader(t, gr, copyInfo.Size, "", "") + copyInfo.Reader = copyInfo.PutObjReader.Reader + } var restore func() if state == "unreadable" { restore = conditionalPutSwapDisks(z.serverPools[1], key, func(d StorageAPI) StorageAPI { @@ -257,22 +277,8 @@ func TestPoolsObjectLockOverwrite(t *testing.T) { case "put": _, writeErr = z.PutObject(ctx, bucket, key, mustGetPutObjReader(t, bytes.NewBufferString("replacement"), 11, "", ""), opts) case "copy", "self-copy": - source := key - if op == "copy" { - source = "source-" + state - putConsistencyObject(t, z, bucket, source, 0, "replacement", ObjectOptions{Versioned: true}) - } - gr, gerr := z.GetObjectNInfo(ctx, bucket, source, nil, nil, ObjectOptions{}) - if gerr != nil { - t.Fatal(gerr) - } - defer gr.Close() - info := gr.ObjInfo - info.UserDefined = opts.UserDefined - info.metadataOnly = op == "self-copy" - info.PutObjReader = mustGetPutObjReader(t, gr, info.Size, "", "") - info.Reader = info.PutObjReader.Reader - _, writeErr = z.CopyObject(ctx, bucket, source, bucket, key, info, ObjectOptions{VersionID: nullVersionID}, opts) + copyInfo.UserDefined = opts.UserDefined + _, writeErr = z.CopyObject(ctx, bucket, source, bucket, key, copyInfo, ObjectOptions{VersionID: nullVersionID}, opts) case "multipart": id, parts := multipartConditionUpload(t, z, bucket, key, 0, opts) _, writeErr = z.CompleteMultipartUpload(ctx, bucket, key, id, parts, ObjectOptions{Versioned: true}) diff --git a/cmd/rename-cancellation_test.go b/cmd/rename-cancellation_test.go index 1f78f8d87..989f95205 100644 --- a/cmd/rename-cancellation_test.go +++ b/cmd/rename-cancellation_test.go @@ -234,6 +234,10 @@ func (d gatedRenameDisk) WriteMetadata(ctx context.Context, ov, volume, path str return d.client.WriteMetadata(ctx, ov, volume, path, fi) } +func (d gatedRenameDisk) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, force bool, opts DeleteOptions) error { + return d.client.DeleteVersion(ctx, volume, path, fi, force, opts) +} + // Use real disk commits behind the real RPC client. The barrier fixes the // order: every commit has started, then cancellation, then remote completion. func TestRenameCancellationKeepsLockUntilRollback(t *testing.T) { @@ -267,6 +271,23 @@ func TestRenameCancellationKeepsLockUntilRollback(t *testing.T) { } copyInfo.metadataOnly = true copyInfo.UserDefined["copy-test"] = "canceled" + // Exercise the production rollback handler and wire options as + // well as the client. A local DeleteVersion bypasses that boundary. + globalLocalDrivesMu.Lock() + previousSets, previousMap := globalLocalSetDrives, globalLocalDrivesMap + globalLocalSetDrives = nil + globalLocalDrivesMap = make(map[string]StorageAPI) + for _, pool := range z.serverPools { + for _, disk := range pool.getHashedSet(object).getDisks() { + globalLocalDrivesMap[disk.Endpoint().String()] = disk + } + } + globalLocalDrivesMu.Unlock() + t.Cleanup(func() { + globalLocalDrivesMu.Lock() + globalLocalSetDrives, globalLocalDrivesMap = previousSets, previousMap + globalLocalDrivesMu.Unlock() + }) tg, err := grid.SetupTestGrid(2) if err != nil { t.Fatal(err) @@ -285,6 +306,10 @@ func TestRenameCancellationKeepsLockUntilRollback(t *testing.T) { wrapped := append([]StorageAPI(nil), disks...) for i, disk := range disks { route := fmt.Sprintf("/cancel-rollback/%d/%d", pi, i) + server := storageRESTServer{endpoint: disk.Endpoint()} + if err := storageDeleteVersionRPC.Register(tg.Managers[1], server.DeleteVersionHandler, route); err != nil { + t.Fatal(err) + } handle := func(p *RenameDataHandlerParams) (*RenameDataResp, *grid.RemoteErr) { entered <- struct{}{} <-release diff --git a/cmd/storage-rest-server-guard_test.go b/cmd/storage-rest-server-guard_test.go index 0216a09d6..e7c5e65bf 100644 --- a/cmd/storage-rest-server-guard_test.go +++ b/cmd/storage-rest-server-guard_test.go @@ -102,14 +102,9 @@ func TestGuardedStorageCoversEveryPathMethod(t *testing.T) { } } -// TestGuardChecksUnreachableFields pins the guard on path fields that exist on -// the wire but that no handler currently reads, so an end-to-end test cannot -// observe them. DeleteVersionHandler hardcodes `opts := DeleteOptions{}` and -// discards p.Opts, which means DeleteOptions.OldDataDir - a value that reaches -// renameAll() with no containment of its own - is unreachable by accident -// rather than by design. If someone later plumbs p.Opts through, the guard is -// already there; this test is what stops it from being removed as "dead". -func TestGuardChecksUnreachableFields(t *testing.T) { +// OldDataDir can reach renameAll during rollback. Guard it independently of +// which handlers currently forward deletion options. +func TestGuardChecksDeleteOptionPaths(t *testing.T) { g := guardedStorage{poisonStorage{}} ctx := context.Background() diff --git a/cmd/storage-rest-server.go b/cmd/storage-rest-server.go index 17d274bca..04941510d 100644 --- a/cmd/storage-rest-server.go +++ b/cmd/storage-rest-server.go @@ -415,8 +415,7 @@ func (s *storageRESTServer) DeleteVersionHandler(p *DeleteVersionHandlerParams) filePath := p.FilePath forceDelMarker := p.ForceDelMarker - opts := DeleteOptions{} - err := s.getStorage().DeleteVersion(context.Background(), volume, filePath, p.FI, forceDelMarker, opts) + err := s.getStorage().DeleteVersion(context.Background(), volume, filePath, p.FI, forceDelMarker, p.Opts) return np, grid.NewRemoteErr(err) } diff --git a/cmd/storage-rest-traversal_test.go b/cmd/storage-rest-traversal_test.go index d1bd4358f..07d11483e 100644 --- a/cmd/storage-rest-traversal_test.go +++ b/cmd/storage-rest-traversal_test.go @@ -231,12 +231,9 @@ func TestStorageRESTTraversalRejected(t *testing.T) { {"DeleteVersion-DataDir", func() error { return restClient.DeleteVersion(ctx, "foo", "obj", badFI(p), false, DeleteOptions{}) }}, - // DeleteOptions.OldDataDir is guarded too, but cannot be exercised - // from here: DeleteVersionHandler hardcodes `opts := DeleteOptions{}` - // and never reads the wire value, so the field is unreachable today. - // That is an accidental mitigation, not a control - see - // TestGuardChecksUnreachableFields for the unit-level assertion that - // the guard is ready if anyone ever plumbs it through. + {"DeleteVersion-OldDataDir", func() error { + return restClient.DeleteVersion(ctx, "foo", "obj", badFI(""), false, DeleteOptions{UndoWrite: true, OldDataDir: p}) + }}, {"RenameData-srcPath", func() error { _, err := restClient.RenameData(ctx, "foo", p, badFI(""), "bar", "dst", RenameOptions{}) return err