mirror of
https://github.com/pgsty/minio.git
synced 2026-10-28 11:37:19 +03:00
fix: preserve remote rollback options and reject stale read quorums
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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})
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user