Skip to content
Merged
Changes from all commits
Commits
File filter

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
44 changes: 44 additions & 0 deletions pkg/storage/bucket/client.go
Original file line number Diff line number Diff line change
@@ -1,10 +1,12 @@
package bucket

import (
"bytes"
"context"
"errors"
"flag"
"fmt"
"io"
"net/http"
"regexp"
"strconv"
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
})
}
Loading