fix: broken get and replace for s3 client (#21163)

Author: sandeepsukhani

pkg/storage/bucket/client.go

@@ -1,10 +1,12 @@
 package bucket
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"flag"
 	"fmt"
+	"io"
 	"net/http"
 	"regexp"
 	"strconv"
@@ -236,6 +238,8 @@ func NewClient(ctx context.Context, backend string, cfg Config, name string, log
 		client = NewPrefixedBucketClient(client, cfg.StoragePrefix)
 	}
 
+	client = newSizedGetAndReplaceBucket(client)
+
 	instrumentedClient := objstoretracing.WrapWithTraces(objstore.WrapWith(client, metrics))
 
 	// Wrap the client with any provided middleware
@@ -274,3 +278,43 @@ func (i *instrumentedRoundTripper) RoundTrip(req *http.Request) (*http.Response,
 
 	return resp, nil
 }
+
+// sizedGetAndReplaceBucket wraps a Bucket and guarantees that the io.ReadCloser
+// returned by GetAndReplace callbacks always has a size known to
+// objstore.TryToGetSize. Without this, callers that return opaque ReadClosers
+// (such as io.NopCloser-wrapped readers) cause the S3 backend to fall into the
+// putObjectMultipartStreamNoLength code path in minio-go, which reconstructs
+// PutObjectOptions before CompleteMultipartUpload and silently drops
+// customHeaders — including the If-Match conditional write header set by
+// GetAndReplace. The result is that stale writes are accepted by S3 even when
+// the object's ETag has already changed, breaking the optimistic-concurrency
+// guarantee that GetAndReplace is supposed to provide.
+//
+// The fix buffers the reader when its size cannot be determined upfront.
+// GetAndReplace is only used for small metadata objects (e.g. ToC files), so
+// the buffering cost is negligible.
+type sizedGetAndReplaceBucket struct {
+	objstore.Bucket
+}
+
+func newSizedGetAndReplaceBucket(bkt objstore.Bucket) objstore.Bucket {
+	return &sizedGetAndReplaceBucket{Bucket: bkt}
+}
+
+func (b *sizedGetAndReplaceBucket) GetAndReplace(ctx context.Context, name string, fn func(io.ReadCloser) (io.ReadCloser, error)) error {
+	return b.Bucket.GetAndReplace(ctx, name, func(existing io.ReadCloser) (io.ReadCloser, error) {
+		rc, err := fn(existing)
+		if err != nil || rc == nil {
+			return rc, err
+		}
+		if _, sizeErr := objstore.TryToGetSize(rc); sizeErr == nil {
+			return rc, nil
+		}
+		data, readErr := io.ReadAll(rc)
+		_ = rc.Close()
+		if readErr != nil {
+			return nil, readErr
+		}
+		return objstore.NopCloserWithSize(bytes.NewReader(data)), nil
+	})
+}