Merge origin/main: conditional multipart initiation (#292)

Signed-off-by: Feng Ruohang <rh@vonng.com>
2026-11-09 17:21:51 +03:00 · 2026-11-09 13:39:10 +08:00
parent b14dc454fb af3064f2b5
commit 1f579afeed
5 changed files with 462 additions and 55 deletions
@@ -90,8 +90,8 @@ jobs:
      - name: Run S3 Select tests under race detector
        run: go test -race ./internal/s3select/... -count=1

-      - name: Run conditional PUT tests under race detector
-        run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP)' -count=1 -timeout=5m
+      - name: Run conditional PUT and multipart tests under race detector
+        run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP|PoolsMultipartCondition)' -count=1 -timeout=5m

      - name: Run multi-pool delete consistency tests under race detector
        run: go test -race ./cmd -run '^Test(Pools(Delete|BulkDelete|RetiringCopyPreservesSharedTierObject)|BulkDeleteQuorumEvidence)' -count=1 -timeout=5m
@@ -25,7 +25,7 @@ When a proposal was incorporated through a later repair, its adoption is credite
 <a href="https://github.com/AleksaMCode"><img src="https://silo.pgsty.com/images/contributors/AleksaMCode.svg" width="60" height="60" alt="@AleksaMCode" title="@AleksaMCode — Fixed the Lambda target-set emptiness check and added a regression test"></a>
 <a href="https://github.com/yupoet"><img src="https://silo.pgsty.com/images/contributors/yupoet.svg" width="60" height="60" alt="@yupoet" title="@yupoet — Improved checksum mismatch messages, verified erasure-coding documentation, and contributed a single-node migration case"></a>
 <a href="https://github.com/clown15"><img src="https://silo.pgsty.com/images/contributors/clown15.svg" width="60" height="60" alt="@clown15" title="@clown15 — Fixed native multipart-upload timestamp parsing during stale-upload cleanup"></a>
-<a href="https://github.com/botdeniska"><img src="https://silo.pgsty.com/images/contributors/botdeniska.svg" width="60" height="60" alt="@botdeniska" title="@botdeniska — Contributed multi-pool migration and deletion safety, tiered expiry and replication checksum repairs, and faster policy evaluation in silo-pkg"></a>
+<a href="https://github.com/botdeniska"><img src="https://silo.pgsty.com/images/contributors/botdeniska.svg" width="60" height="60" alt="@botdeniska" title="@botdeniska — Contributed multi-pool migration, deletion safety and conditional multipart checks, tiered expiry and replication checksum repairs, and faster policy evaluation in silo-pkg"></a>
 <a href="https://github.com/910patin"><img src="https://silo.pgsty.com/images/contributors/910patin.svg" width="60" height="60" alt="@910patin" title="@910patin — Fixed long-DN LDAP login while preserving IAM revocation-record errors"></a>
 <a href="https://github.com/magicxor"><img src="https://silo.pgsty.com/images/contributors/magicxor.svg" width="60" height="60" alt="@magicxor" title="@magicxor — Reported and proposed conditional DELETE support for If-Match"></a>
 <a href="https://github.com/davinkevin"><img src="https://silo.pgsty.com/images/contributors/davinkevin.svg" width="60" height="60" alt="@davinkevin" title="@davinkevin — Proposed the distroless container image and dependency automation"></a>
@@ -98,7 +98,7 @@ When a proposal was incorporated through a later repair, its adoption is credite
 | [@AleksaMCode](https://github.com/AleksaMCode) | Fixed the Lambda target-set emptiness check and added a regression test | Merged PRs: [pgsty/silo#220](https://github.com/pgsty/silo/pull/220) |
 | [@yupoet](https://github.com/yupoet) | Improved checksum mismatch messages, verified erasure-coding documentation, and contributed a single-node migration case | Merged PRs: [pgsty/silo#228](https://github.com/pgsty/silo/pull/228), [pgsty/silo#229](https://github.com/pgsty/silo/pull/229)<br>PRs closed without merging: [pgsty/silo#230](https://github.com/pgsty/silo/pull/230)<br>Also [cross-checked the two-drive EC:1 documentation](https://github.com/pgsty/silo/issues/197#issuecomment-5846341939). The single-node migration report in #228 was merged on 2026-10-23; the registry-mirror proposal in #230 was closed without merging. |
 | [@clown15](https://github.com/clown15) | Fixed native multipart-upload timestamp parsing during stale-upload cleanup | Merged PRs: [pgsty/silo#236](https://github.com/pgsty/silo/pull/236) |
-| [@botdeniska](https://github.com/botdeniska) | Contributed multi-pool migration and deletion safety, tiered expiry and replication checksum repairs, and faster policy evaluation in silo-pkg | Merged PRs: [pgsty/silo#253](https://github.com/pgsty/silo/pull/253), [pgsty/silo#255](https://github.com/pgsty/silo/pull/255), [pgsty/silo#257](https://github.com/pgsty/silo/pull/257)<br>PRs closed without merging: [pgsty/silo#245](https://github.com/pgsty/silo/pull/245), [pgsty/silo#248](https://github.com/pgsty/silo/pull/248)<br>Closed issues: [pgsty/silo#252](https://github.com/pgsty/silo/issues/252), [pgsty/silo#254](https://github.com/pgsty/silo/issues/254), [pgsty/silo#256](https://github.com/pgsty/silo/issues/256)<br>Merged PRs: [pgsty/silo-pkg#11](https://github.com/pgsty/silo-pkg/pull/11)<br>The migration repair from [#248](https://github.com/pgsty/silo/pull/248) was incorporated in [60c7179dd](https://github.com/pgsty/silo/commit/60c7179ddbfcbd41a1fe67707230f5b360a43020), retaining the original authorship and DCO. |
+| [@botdeniska](https://github.com/botdeniska) | Contributed multi-pool migration, deletion safety and conditional multipart checks, tiered expiry and replication checksum repairs, and faster policy evaluation in silo-pkg | Merged PRs: [pgsty/silo#253](https://github.com/pgsty/silo/pull/253), [pgsty/silo#255](https://github.com/pgsty/silo/pull/255), [pgsty/silo#257](https://github.com/pgsty/silo/pull/257), [pgsty/silo#292](https://github.com/pgsty/silo/pull/292)<br>PRs closed without merging: [pgsty/silo#245](https://github.com/pgsty/silo/pull/245), [pgsty/silo#248](https://github.com/pgsty/silo/pull/248)<br>Closed issues: [pgsty/silo#252](https://github.com/pgsty/silo/issues/252), [pgsty/silo#254](https://github.com/pgsty/silo/issues/254), [pgsty/silo#256](https://github.com/pgsty/silo/issues/256), [pgsty/silo#291](https://github.com/pgsty/silo/issues/291)<br>Merged PRs: [pgsty/silo-pkg#11](https://github.com/pgsty/silo-pkg/pull/11)<br>The migration repair from [#248](https://github.com/pgsty/silo/pull/248) was incorporated in [60c7179dd](https://github.com/pgsty/silo/commit/60c7179ddbfcbd41a1fe67707230f5b360a43020), retaining the original authorship and DCO. |
 | [@910patin](https://github.com/910patin) | Fixed long-DN LDAP login while preserving IAM revocation-record errors | Merged PRs: [pgsty/silo#250](https://github.com/pgsty/silo/pull/250) |

 ## Other pull-request authors
@@ -64,6 +64,41 @@ func (z *erasureServerPools) objectPoolInfos(ctx context.Context, bucket, object
 	return copies, nil
 }

+// checkCurrentPreconditions evaluates the public write conditions in
+// opts.CheckPrecondFn against the logical current object across pools,
+// including draining pools. The pool that receives a write may be empty or
+// hold a stale copy, so its local state cannot decide the condition. A
+// current delete marker reads as an absent key: If-Match then fails with
+// ObjectNotFound and If-None-Match may proceed. An unreadable pool may hold
+// the newest object; it is not absence, and its error is returned without
+// evaluating the condition. The caller must hold the pools-layer object lock
+// and, once the condition is accepted, clear the callback so the destination
+// does not evaluate it again against its local copy.
+func (z *erasureServerPools) checkCurrentPreconditions(ctx context.Context, bucket, object string, opts ObjectOptions) error {
+	copies, err := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{
+		VersionID:        "", // Compare the current object, not the write's version.
+		Versioned:        opts.Versioned,
+		VersionSuspended: opts.VersionSuspended,
+		NoAuditLog:       true,
+	})
+	if err == nil && copies[0].ObjInfo.DeleteMarker {
+		err = toObjectErr(errFileNotFound, bucket, decodeDirObject(object))
+	}
+	switch {
+	case err == nil:
+		if opts.CheckPrecondFn(copies[0].ObjInfo) {
+			return PreConditionFailed{}
+		}
+		return nil
+	case !isErrObjectNotFound(err) && !isErrVersionNotFound(err):
+		return err
+	case opts.HasIfMatch:
+		return err
+	default:
+		return nil
+	}
+}
+
 // checkDataMovementCopies rejects moving a version when another pool holds a
 // newer copy of it, which reads return instead. A cross-pool overwrite can
 // leave such a copy, and a client write can add one while the move copies.
@@ -0,0 +1,389 @@
+// Copyright (c) 2026 Denis Babchuk
+// Copyright (c) 2026 Feng Ruohang
+//
+// This file is part of Silo Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"context"
+	"encoding/xml"
+	"fmt"
+	"net/http"
+	"slices"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	xhttp "github.com/minio/minio/internal/http"
+)
+
+// multipartInitiateUploads returns the number of pending uploads of object in
+// each pool.
+func multipartInitiateUploads(t *testing.T, z *erasureServerPools, bucket, object string) []int {
+	t.Helper()
+	counts := make([]int, len(z.serverPools))
+	for i, pool := range z.serverPools {
+		res, err := pool.listMultipartUploadsExact(t.Context(), bucket, object)
+		if err != nil {
+			t.Fatal(err)
+		}
+		counts[i] = len(res.Uploads)
+	}
+	return counts
+}
+
+// A multi-pool NewMultipartUpload goes to the pool of a pending upload of the
+// key or, without one, to the pool chosen by capacity. Neither has to hold the
+// current object, so a condition sent with the initiate must be evaluated
+// against the current object across pools, as for PUT and completion.
+// Routing itself does not change: an accepted upload lands in the same pool.
+// Allocation follows a current delete marker to its pool, so a stale version
+// under a marker in another pool is reached only through a pending upload.
+func TestPoolsMultipartConditionalInitiateHTTP(t *testing.T) {
+	const wrongETag = "ffffffffffffffffffffffffffffffff"
+	for _, mode := range []string{"unversioned", "versioned"} {
+		t.Run(mode, func(t *testing.T) {
+			z, _ := consistencyPools(t)
+			bucket, router := conditionalPutBucket(t, z, mode)
+			versioned := mode == "versioned"
+			for _, route := range []string{"pending-upload", "allocation"} {
+				for target := range 2 {
+					for _, layout := range []string{"current-in-other-pool", "stale-copy-in-target", "marker-in-other-pool", "control-current-in-target"} {
+						if layout == "marker-in-other-pool" && (!versioned || route == "allocation") {
+							continue
+						}
+						for _, cond := range []string{"if-match-current", "if-match-any", "if-match-stale", "if-none-match-any"} {
+							t.Run(fmt.Sprintf("%s/target=%d/%s/%s", route, target, layout, cond), func(t *testing.T) {
+								object := fmt.Sprintf("%s-%d-%s-%s", route, target, layout, cond)
+								other := 1 - target
+								opts := func(age time.Duration) ObjectOptions {
+									return ObjectOptions{Versioned: versioned, MTime: UTCNow().Add(-age)}
+								}
+								present, staleETag := true, wrongETag
+								var current ObjectInfo
+								switch layout {
+								case "current-in-other-pool":
+									current = putConsistencyObject(t, z, bucket, object, other, "current", opts(time.Minute))
+								case "stale-copy-in-target":
+									staleETag = putConsistencyObject(t, z, bucket, object, target, "stale", opts(time.Hour)).ETag
+									current = putConsistencyObject(t, z, bucket, object, other, "current", opts(time.Minute))
+								case "marker-in-other-pool":
+									// The client knows the version under the marker.
+									current = putConsistencyObject(t, z, bucket, object, target, "deleted", opts(time.Hour))
+									if _, err := z.serverPools[other].DeleteObject(t.Context(), bucket, object, ObjectOptions{Versioned: true, VersionID: mustGetUUID(), DeleteMarker: true, MTime: UTCNow().Add(-time.Minute)}); err != nil {
+										t.Fatal(err)
+									}
+									present = false
+								case "control-current-in-target":
+									staleETag = putConsistencyObject(t, z, bucket, object, other, "stale", opts(time.Hour)).ETag
+									current = putConsistencyObject(t, z, bucket, object, target, "current", opts(time.Minute))
+								}
+
+								before := make([]int, len(z.serverPools))
+								if route == "pending-upload" {
+									if _, err := z.serverPools[target].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{Versioned: versioned}); err != nil {
+										t.Fatal(err)
+									}
+									before[target] = 1
+								} else {
+									defer conditionalPutPool(t, z, object, target)()
+									if idx, err := z.getWritePoolIdx(t.Context(), bucket, object, -1, false); err != nil || idx != target {
+										t.Fatalf("allocation target=%d: idx=%d err=%v", target, idx, err)
+									}
+								}
+
+								var headers map[string]string
+								switch cond {
+								case "if-match-current":
+									headers = map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", current.ETag)}
+								case "if-match-any":
+									headers = map[string]string{xhttp.IfMatch: "*"}
+								case "if-match-stale":
+									headers = map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", staleETag)}
+								case "if-none-match-any":
+									headers = map[string]string{xhttp.IfNoneMatch: "*"}
+								}
+								want := http.StatusOK
+								switch {
+								case !present && strings.HasPrefix(cond, "if-match"):
+									want = http.StatusNotFound
+								case !present:
+								case cond == "if-match-stale", cond == "if-none-match-any":
+									want = http.StatusPreconditionFailed
+								}
+
+								rec := multipartConditionRequest(t, router, http.MethodPost, getNewMultipartURL("", bucket, object), "", headers)
+								after := multipartInitiateUploads(t, z, bucket, object)
+								t.Logf("initiate %v: HTTP %d, uploads per pool %v -> %v", headers, rec.Code, before, after)
+								if rec.Code != want {
+									t.Errorf("initiate HTTP %d, want %d: %s", rec.Code, want, rec.Body.String())
+								}
+								wantAfter := slices.Clone(before)
+								if want == http.StatusOK {
+									wantAfter[target]++
+								}
+								if !slices.Equal(after, wantAfter) {
+									t.Errorf("uploads per pool %v, want %v", after, wantAfter)
+								}
+								if rec.Code != want {
+									return
+								}
+								switch want {
+								case http.StatusPreconditionFailed:
+									multipartConditionError(t, rec, "PreconditionFailed")
+									if etag := multipartConditionResponseETag(rec); etag != current.ETag {
+										t.Errorf("412 describes ETag %s, want the current %s", etag, current.ETag)
+									}
+								case http.StatusNotFound:
+									multipartConditionError(t, rec, "NoSuchKey")
+								}
+							})
+						}
+					}
+				}
+			}
+		})
+	}
+}
+
+// A pool that cannot be read may hold the newest object, so the condition
+// cannot be evaluated even when the upload goes to a readable pool.
+func TestPoolsMultipartConditionalInitiateUnreadable(t *testing.T) {
+	z, _ := consistencyPools(t)
+	bucket, router := conditionalPutBucket(t, z, "unversioned")
+	for faultPool := range 2 {
+		for _, match := range []bool{false, true} {
+			t.Run(fmt.Sprintf("fault-pool=%d/match=%t", faultPool, match), func(t *testing.T) {
+				object := fmt.Sprintf("unreadable-%d-%t", faultPool, match)
+				putConsistencyObject(t, z, bucket, object, 0, "stale", ObjectOptions{MTime: UTCNow().Add(-time.Hour)})
+				putConsistencyObject(t, z, bucket, object, 1, "current", ObjectOptions{MTime: UTCNow().Add(-time.Minute)})
+				defer conditionalPutPool(t, z, object, 0)()
+				defer conditionalPutSwapDisks(z.serverPools[faultPool], object, func(disk StorageAPI) StorageAPI {
+					return consistencyReadFaultDisk{StorageAPI: disk, bucket: bucket, object: object}
+				})()
+				headers := map[string]string{xhttp.IfNoneMatch: "*"}
+				if match {
+					headers = map[string]string{xhttp.IfMatch: "*"}
+				}
+				rec := multipartConditionRequest(t, router, http.MethodPost, getNewMultipartURL("", bucket, object), "", headers)
+				if rec.Code != http.StatusServiceUnavailable {
+					t.Errorf("unverified initiate must fail: HTTP %d: %s", rec.Code, rec.Body.String())
+				}
+				called := 0
+				_, err := z.NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{HasIfMatch: match, CheckPrecondFn: func(ObjectInfo) bool { called++; return false }})
+				if !isErrReadQuorum(err) || called != 0 {
+					t.Errorf("initiate error=%v, callback calls=%d", err, called)
+				}
+				if n := multipartInitiateUploads(t, z, bucket, object); !slices.Equal(n, []int{0, 0}) {
+					t.Errorf("uploads per pool %v after failed initiates", n)
+				}
+			})
+		}
+	}
+}
+
+type multipartInitiateMetadataGate struct {
+	entered, resume chan struct{}
+	once            sync.Once
+}
+
+type multipartInitiateMetadataDisk struct {
+	StorageAPI
+	gate *multipartInitiateMetadataGate
+}
+
+func (d multipartInitiateMetadataDisk) WriteMetadata(ctx context.Context, orig, volume, path string, fi FileInfo) error {
+	if volume == minioMetaMultipartBucket {
+		d.gate.once.Do(func() { close(d.gate.entered) })
+		select {
+		case <-d.gate.resume:
+		case <-ctx.Done():
+			return ctx.Err()
+		}
+	}
+	return d.StorageAPI.WriteMetadata(ctx, orig, volume, path, fi)
+}
+
+// The pools-layer lock must cover placement of the upload, not just reading
+// the current object. Pause the metadata write after the condition passed and
+// require a concurrent public overwrite or delete to wait for that placement.
+func TestPoolsMultipartConditionalInitiateHoldsLockThroughPersistence(t *testing.T) {
+	for _, mutation := range []string{"overwrite", "delete"} {
+		t.Run(mutation, func(t *testing.T) {
+			z, bucket := consistencyPools(t)
+			object := "locked-initiate"
+			current := putConsistencyObject(t, z, bucket, object, 1, "current", ObjectOptions{})
+			if _, err := z.serverPools[0].NewMultipartUpload(t.Context(), bucket, object, ObjectOptions{}); err != nil {
+				t.Fatal(err)
+			}
+			gate := &multipartInitiateMetadataGate{entered: make(chan struct{}), resume: make(chan struct{})}
+			restoreDisks := conditionalPutSwapDisks(z.serverPools[0], object, func(d StorageAPI) StorageAPI {
+				return multipartInitiateMetadataDisk{StorageAPI: d, gate: gate}
+			})
+			ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+			var workers sync.WaitGroup
+			var releaseOnce sync.Once
+			release := func() { releaseOnce.Do(func() { close(gate.resume) }) }
+			defer func() {
+				release()
+				cancel()
+				workers.Wait()
+				restoreDisks()
+			}()
+
+			initiated := make(chan error, 1)
+			workers.Add(1)
+			go func() {
+				defer workers.Done()
+				_, err := z.NewMultipartUpload(ctx, bucket, object, ObjectOptions{
+					HasIfMatch: true,
+					CheckPrecondFn: func(oi ObjectInfo) bool {
+						return oi.ETag != current.ETag
+					},
+				})
+				initiated <- err
+			}()
+			select {
+			case <-gate.entered:
+			case err := <-initiated:
+				t.Fatalf("initiate returned before persistence: %v", err)
+			case <-ctx.Done():
+				t.Fatal(ctx.Err())
+			}
+
+			mutated := make(chan error, 1)
+			started := make(chan struct{})
+			reader := mustGetPutObjReader(t, strings.NewReader("new"), 3, "", "")
+			workers.Add(1)
+			go func() {
+				defer workers.Done()
+				close(started)
+				var err error
+				if mutation == "delete" {
+					_, err = z.DeleteObject(ctx, bucket, object, ObjectOptions{})
+				} else {
+					_, err = z.PutObject(ctx, bucket, object, reader, ObjectOptions{})
+				}
+				mutated <- err
+			}()
+			<-started
+			select {
+			case err := <-mutated:
+				t.Fatalf("mutation escaped initiate lock: %v", err)
+			case <-time.After(100 * time.Millisecond):
+			}
+			release()
+			for _, result := range []<-chan error{initiated, mutated} {
+				select {
+				case err := <-result:
+					if err != nil {
+						t.Fatal(err)
+					}
+				case <-ctx.Done():
+					t.Fatal(ctx.Err())
+				}
+			}
+		})
+	}
+}
+
+// Initiate only checks the current state; it does not reserve the object or
+// persist a condition for completion. Completion must carry its own condition,
+// compares ETags rather than object generations, and leaves parts available
+// for retry when a condition fails.
+func TestPoolsMultipartConditionalInitiateLifecycle(t *testing.T) {
+	z, _ := consistencyPools(t)
+	bucket, router := conditionalPutBucket(t, z, "unversioned")
+	for _, change := range []string{"overwrite", "delete", "recreate-same-etag", "omit-complete-condition"} {
+		t.Run(change, func(t *testing.T) {
+			object := "lifecycle-" + change
+			current := putConsistencyObject(t, z, bucket, object, 1, "original", ObjectOptions{})
+			defer conditionalPutPool(t, z, object, 0)()
+			headers := map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", current.ETag)}
+			initiate := multipartConditionRequest(t, router, http.MethodPost, getNewMultipartURL("", bucket, object), "", headers)
+			if initiate.Code != http.StatusOK {
+				t.Fatalf("initiate: %d %s", initiate.Code, initiate.Body.String())
+			}
+			var upload InitiateMultipartUploadResponse
+			if err := xml.Unmarshal(initiate.Body.Bytes(), &upload); err != nil || upload.UploadID == "" {
+				t.Fatalf("upload ID: %v %s", err, initiate.Body.String())
+			}
+			const finalBody = "multipart-final"
+			part, err := z.PutObjectPart(t.Context(), bucket, object, upload.UploadID, 1,
+				mustGetPutObjReader(t, strings.NewReader(finalBody), int64(len(finalBody)), "", ""), ObjectOptions{})
+			if err != nil {
+				t.Fatal(err)
+			}
+
+			if change == "delete" || change == "recreate-same-etag" {
+				if _, err := z.DeleteObject(t.Context(), bucket, object, ObjectOptions{}); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if change != "delete" {
+				body := "concurrent"
+				if change == "recreate-same-etag" {
+					body = "original"
+				}
+				if _, err := z.PutObject(t.Context(), bucket, object,
+					mustGetPutObjReader(t, strings.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil {
+					t.Fatal(err)
+				}
+			}
+			if change == "omit-complete-condition" {
+				headers = nil
+			}
+			url := getCompleteMultipartUploadURL("", bucket, object, upload.UploadID)
+			body := multipartConditionCompleteBody([]CompletePart{{PartNumber: 1, ETag: part.ETag}})
+			complete := multipartConditionRequest(t, router, http.MethodPost, url, body, headers)
+			want := http.StatusOK
+			if change == "overwrite" {
+				want = http.StatusPreconditionFailed
+			}
+			if change == "delete" {
+				want = http.StatusNotFound
+			}
+			if complete.Code != want {
+				t.Fatalf("complete: %d, want %d: %s", complete.Code, want, complete.Body.String())
+			}
+			if want != http.StatusOK {
+				parts, err := z.ListObjectParts(t.Context(), bucket, object, upload.UploadID, 0, 100, ObjectOptions{})
+				if err != nil || len(parts.Parts) != 1 || parts.Parts[0].ETag != part.ETag {
+					t.Fatalf("failed condition consumed upload: %v %+v", err, parts)
+				}
+				if change == "overwrite" {
+					oi, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{})
+					if err != nil {
+						t.Fatal(err)
+					}
+					headers = map[string]string{xhttp.IfMatch: fmt.Sprintf("%q", oi.ETag)}
+				} else {
+					headers = map[string]string{xhttp.IfNoneMatch: "*"}
+				}
+				complete = multipartConditionRequest(t, router, http.MethodPost, url, body, headers)
+				if complete.Code != http.StatusOK {
+					t.Fatalf("retry with new condition: %d %s", complete.Code, complete.Body.String())
+				}
+			}
+			get := multipartConditionRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object), "", nil)
+			if get.Code != http.StatusOK || get.Body.String() != finalBody {
+				t.Fatalf("committed body: %d %q", get.Code, get.Body.String())
+			}
+		})
+	}
+}
@@ -1157,28 +1157,8 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec
 	// metadata reconciliation at the set layer.
 	if opts.CheckPrecondFn != nil && !opts.ReplicationRequest &&
 		!opts.ReplicaLockReconcile && !opts.DataMovement {
-		copies, lerr := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{
-			VersionID:        "", // Compare the current object, not the write's version.
-			Versioned:        opts.Versioned,
-			VersionSuspended: opts.VersionSuspended,
-			NoAuditLog:       true,
-		})
-		var latest ObjectInfo
-		if lerr == nil {
-			latest = copies[0].ObjInfo
-			if latest.DeleteMarker {
-				lerr = toObjectErr(errFileNotFound, bucket, object)
-			}
-		}
-		// An unreadable pool may hold the newest object; it is not absence.
-		if lerr != nil && !isErrObjectNotFound(lerr) && !isErrVersionNotFound(lerr) {
-			return ObjectInfo{}, lerr
-		}
-		if lerr == nil && opts.CheckPrecondFn(latest) {
-			return ObjectInfo{}, PreConditionFailed{}
-		}
-		if lerr != nil && opts.HasIfMatch {
-			return ObjectInfo{}, lerr
+		if err := z.checkCurrentPreconditions(ctx, bucket, object, opts); err != nil {
+			return ObjectInfo{}, err
 		}
 		// Do not repeat an accepted condition against the destination's copy.
 		opts.CheckPrecondFn = nil
@@ -2211,6 +2191,33 @@ func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, obj
 		return z.serverPools[0].NewMultipartUpload(ctx, bucket, object, opts)
 	}

+	// The upload goes to the pool of a pending upload of the key or to the
+	// pool selected by capacity; neither has to hold the current object. As
+	// for PutObject, public write conditions compare the logical current
+	// object under the pools-layer write lock, which is kept while the upload
+	// is placed. Replica and data-movement callbacks keep their set-layer
+	// semantics.
+	locked := false
+	if opts.CheckPrecondFn != nil && !opts.ReplicationRequest &&
+		!opts.ReplicaLockReconcile && !opts.DataMovement {
+		if !opts.NoLock {
+			lk := z.NewNSLock(bucket, encodeDirObject(object))
+			lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
+			if err != nil {
+				return nil, err
+			}
+			ctx = lkctx.Context()
+			defer lk.Unlock(lkctx)
+			opts.NoLock = true
+		}
+		locked = true
+		if err := z.checkCurrentPreconditions(ctx, bucket, encodeDirObject(object), opts); err != nil {
+			return nil, err
+		}
+		// Do not repeat an accepted condition against the destination's copy.
+		opts.CheckPrecondFn = nil
+	}
+
 	for idx, pool := range z.serverPools {
 		if z.IsSuspended(idx) || z.IsPoolRebalancing(idx) {
 			continue
@@ -2229,8 +2236,9 @@ func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, obj
 	}

 	// any parallel writes on the object will block for this poolIdx
-	// to return since this holds a read lock on the namespace.
-	idx, err := z.getPoolIdx(ctx, bucket, object, -1)
+	// to return since this holds a read lock on the namespace, or the
+	// write lock taken above for a conditional upload.
+	idx, err := z.getWritePoolIdx(ctx, bucket, object, -1, locked)
 	if err != nil {
 		return nil, err
 	}
@@ -2454,33 +2462,8 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
 		// Once satisfied, the callback is cleared so the set layer does not
 		// re-evaluate it against its local copy.
 		if opts.CheckPrecondFn != nil {
-			copies, lerr := z.objectPoolInfos(ctx, bucket, encodeDirObject(object), ObjectOptions{
-				// Conditions always compare the logical current object,
-				// independently of the completion's destination version.
-				VersionID:        "",
-				Versioned:        opts.Versioned,
-				VersionSuspended: opts.VersionSuspended,
-				NoAuditLog:       true,
-			})
-			var latest ObjectInfo
-			if lerr == nil {
-				latest = copies[0].ObjInfo
-				if latest.DeleteMarker {
-					// A delete-marker latest reads as an absent key, matching
-					// the set layer's getObjectInfo.
-					lerr = toObjectErr(errFileNotFound, bucket, object)
-				}
-			}
-			if lerr == nil && opts.CheckPrecondFn(latest) {
-				return ObjectInfo{}, PreConditionFailed{}
-			}
-			if lerr != nil && !isErrVersionNotFound(lerr) && !isErrObjectNotFound(lerr) {
-				return ObjectInfo{}, lerr
-			}
-			// if object doesn't exist return error for If-Match conditional requests
-			// If-None-Match should be allowed to proceed for non-existent objects
-			if lerr != nil && opts.HasIfMatch && (isErrObjectNotFound(lerr) || isErrVersionNotFound(lerr)) {
-				return ObjectInfo{}, lerr
+			if err := z.checkCurrentPreconditions(ctx, bucket, encodeDirObject(object), opts); err != nil {
+				return ObjectInfo{}, err
 			}
 			opts.CheckPrecondFn = nil
 		}