Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 32 additions & 9 deletions core/backend_s3.go
Original file line number Diff line number Diff line change
Expand Up @@ -335,7 +335,7 @@ func (s *S3Backend) detectBucketLocationByHEAD() (err error, isAws bool) {
switch resp.StatusCode {
case 200:
// note that this only happen if the bucket is in us-east-1
if len(s.config.Profile) == 0 && os.Getenv("AWS_ACCESS_KEY_ID") == "" {
if len(s.config.Profile) == 0 && os.Getenv("AWS_ACCESS_KEY_ID") == "" && !s.config.UseIAM {
s.awsConfig.Credentials = credentials.AnonymousCredentials
s3Log.Info().Msg("anonymous bucket detected")
}
Expand Down Expand Up @@ -933,12 +933,24 @@ func (s *S3Backend) copyObjectMultipart(size int64, from string, to string, mpuI
params.ACL = &s.config.ACL
}

resp, err := s.CreateMultipartUpload(params)
if err != nil {
return "", err
resp, createErr := s.CreateMultipartUpload(params)
if createErr != nil {
return "", createErr
}

mpuId = *resp.UploadId
// This helper owns uploads it creates; callers cannot resume them because
// their IDs are not returned on failure. Leave caller-supplied uploads alone.
defer func() {
if err != nil {
_, abortErr := s.AbortMultipartUpload(&s3.AbortMultipartUploadInput{
Bucket: &s.bucket, Key: &to, UploadId: &mpuId,
})
if abortErr != nil {
s3Log.Warnf("Abort multipart copy %s: %v", to, abortErr)
}
}
}()
}

partSizes := s.defaultCopyPartSizes(size)
Expand Down Expand Up @@ -975,19 +987,23 @@ func (s *S3Backend) copyObjectMultipart(size int64, from string, to string, mpuI
}

func (s *S3Backend) CopyBlob(param *CopyBlobInput) (*CopyBlobOutput, error) {
return s.copyBlob(param, false)
}

func (s *S3Backend) copyBlob(param *CopyBlobInput, forceMultipart bool) (*CopyBlobOutput, error) {
metadataDirective := s3.MetadataDirectiveCopy
if param.Metadata != nil {
metadataDirective = s3.MetadataDirectiveReplace
}

from := s.bucket + "/" + param.Source

// Copy into the same object is used to just update metadata
// and should be very quick regardless of parameters
if param.Source != param.Destination {
// Prefer the optimized metadata-only self-copy when the backend supports it.
// Some S3 implementations enforce the single-copy size limit on self-copies too.
if forceMultipart || param.Source != param.Destination {

// FIXME Remove additional HEAD query
if param.Size == nil || param.ETag == nil || (*param.Size > s.config.MultipartCopyThreshold &&
if param.Size == nil || param.ETag == nil || ((forceMultipart || *param.Size > s.config.MultipartCopyThreshold) &&
(param.Metadata == nil || param.StorageClass == nil)) {

params := &HeadBlobInput{Key: param.Source}
Expand All @@ -1008,7 +1024,7 @@ func (s *S3Backend) CopyBlob(param *CopyBlobInput) (*CopyBlobOutput, error) {
param.StorageClass = s.selectStorageClass(param.Size)
}

if !s.gcs && *param.Size > s.config.MultipartCopyThreshold {

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fallback copy drops caller ETag

Medium Severity

The EntityTooLarge self-copy fallback always HEADs when StorageClass is unset and then replaces the caller-supplied ETag. Metadata flushes pass knownETag so the multipart copy can If-Match that version; after the overwrite, part copies pin whatever is live now. A remote resize or replace is applied with local metadata instead of returning ERANGE and discarding the stale update.

Additional Locations (1)
Fix in Cursor Fix in Web

Reviewed by Cursor Bugbot for commit 271f945. Configure here.

if !s.gcs && (forceMultipart || *param.Size > s.config.MultipartCopyThreshold) {
reqId, err := s.copyObjectMultipart(int64(*param.Size), from, param.Destination, "", param.ETag, param.Metadata, param.StorageClass)
if err != nil {
return nil, err
Expand Down Expand Up @@ -1056,6 +1072,13 @@ func (s *S3Backend) CopyBlob(param *CopyBlobInput) (*CopyBlobOutput, error) {
req.Config.HTTPClient.Timeout = 15 * time.Minute
err := req.Send()
if err != nil {
// Some S3 implementations enforce the single-copy size limit even on a
// self-copy, which is only used to rewrite metadata. Fall back to a
// multipart copy rather than failing the metadata update.
var awsErr awserr.Error
if !s.gcs && param.Source == param.Destination && errors.As(err, &awsErr) && awsErr.Code() == "EntityTooLarge" {
return s.copyBlob(param, true)
}
s3Log.Warn().Interface("params", params).Err(err).Msg("CopyObject failed")
return nil, err
}
Expand Down
159 changes: 159 additions & 0 deletions core/backend_s3_copy_test.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,159 @@
package core

import (
"fmt"
"net/http"
"net/http/httptest"
"reflect"
"sync"
"testing"

"github.com/tigrisdata/tigrisfs/core/cfg"
)

func TestS3CopyBlobMetadataSelfCopy(t *testing.T) {
const size = uint64(100 * 1024 * 1024)
for _, tc := range []struct {
name string
copyError string
failPart bool
failCommit bool
withoutHint bool
wantErr bool
wantMPU bool
}{
{name: "supported self-copy remains a single request"},
{name: "oversized self-copy uses multipart", copyError: "EntityTooLarge", wantMPU: true},
{name: "missing hints preserve source metadata", copyError: "EntityTooLarge", withoutHint: true, wantMPU: true},
{name: "access denial is not retried as multipart", copyError: "AccessDenied", wantErr: true},
{name: "part failure aborts owned upload", copyError: "EntityTooLarge", failPart: true, wantMPU: true, wantErr: true},
{name: "completion failure aborts owned upload", copyError: "EntityTooLarge", failCommit: true, wantMPU: true, wantErr: true},
} {
t.Run(tc.name, func(t *testing.T) {
var mu sync.Mutex
var copies, heads, starts, parts, completes, aborts int
var pending map[string]string
stored := map[string]string{"mtime": "100", "custom": "retained"}
wanted := map[string]string{"mtime": "200", "custom": "retained"}
if tc.withoutHint {
wanted = map[string]string{"mtime": "100", "custom": "retained"}
}
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
defer mu.Unlock()
failure := func(status int, code string) {
w.WriteHeader(status)
fmt.Fprintf(w, "<Error><Code>%s</Code><Message>fixture</Message></Error>", code)
}
if r.URL.Path != "/testbucket/object.iso" {
t.Errorf("unexpected object path: %s", r.URL.Path)
failure(http.StatusBadRequest, "InvalidRequest")
return
}
query := r.URL.Query()
switch {
case r.Method == http.MethodHead:
heads++
w.Header().Set("Content-Length", fmt.Sprint(size))
w.Header().Set("ETag", `"original"`)
w.Header().Set("Last-Modified", "Mon, 01 Jan 2024 00:00:00 GMT")
w.Header().Set("X-Amz-Storage-Class", "STANDARD_IA")
w.Header().Set("X-Amz-Meta-Mtime", "100")
w.Header().Set("X-Amz-Meta-Custom", "retained")
case r.Method == http.MethodPost && query.Has("uploads"):
starts++
pending = map[string]string{"mtime": r.Header.Get("X-Amz-Meta-Mtime"), "custom": r.Header.Get("X-Amz-Meta-Custom")}
if r.Header.Get("X-Amz-Storage-Class") != "STANDARD_IA" {
t.Error("multipart copy lost source storage class")
}
fmt.Fprint(w, "<InitiateMultipartUploadResult><UploadId>upload</UploadId></InitiateMultipartUploadResult>")
case r.Method == http.MethodPut && query.Get("uploadId") == "upload":
parts++
if r.Header.Get("X-Amz-Copy-Source") != "testbucket/object.iso" || r.Header.Get("X-Amz-Copy-Source-If-Match") != `"original"` {
t.Error("multipart copy did not pin the original source/ETag")
}
if tc.failPart {
failure(http.StatusPreconditionFailed, "PreconditionFailed")
return
}
fmt.Fprint(w, `<CopyPartResult><ETag>"part"</ETag></CopyPartResult>`)
case r.Method == http.MethodPost && query.Get("uploadId") == "upload":
completes++
if tc.failCommit {
failure(http.StatusForbidden, "AccessDenied")
return
}
stored = pending
fmt.Fprint(w, `<CompleteMultipartUploadResult><ETag>"copied"</ETag></CompleteMultipartUploadResult>`)
case r.Method == http.MethodDelete && query.Get("uploadId") == "upload":
aborts++
w.WriteHeader(http.StatusNoContent)
case r.Method == http.MethodPut && len(query) == 0:
copies++
if tc.copyError != "" {
status := http.StatusBadRequest
if tc.copyError == "AccessDenied" {
status = http.StatusForbidden
}
failure(status, tc.copyError)
return
}
stored = map[string]string{"mtime": r.Header.Get("X-Amz-Meta-Mtime"), "custom": r.Header.Get("X-Amz-Meta-Custom")}
fmt.Fprint(w, `<CopyObjectResult><ETag>"copied"</ETag></CopyObjectResult>`)
default:
t.Errorf("unexpected request: %s %s", r.Method, r.URL)
failure(http.StatusBadRequest, "InvalidRequest")
}
}))
defer server.Close()
s, err := NewS3("testbucket", &cfg.FlagStorage{Endpoint: server.URL, MaxParallelParts: 2}, (&cfg.S3Config{
Region: "us-east-1", AccessKey: "test", SecretKey: "test",
// The service's single-copy limit can be below the configured threshold.
MultipartCopyThreshold: 256 * 1024 * 1024,
}).Init())
if err != nil {
t.Fatal(err)
}
in := &CopyBlobInput{Source: "object.iso", Destination: "object.iso"}
if !tc.withoutHint {
in.Size = PUInt64(size)
in.ETag = PString(`"original"`)
in.Metadata = map[string]*string{"mtime": PString("200"), "custom": PString("retained")}
}
_, err = s.CopyBlob(in)
if (err != nil) != tc.wantErr {
t.Fatalf("CopyBlob error = %v, want error %v", err, tc.wantErr)
}
mu.Lock()
defer mu.Unlock()
if copies != 1 {
t.Errorf("single-copy attempts = %d, want 1", copies)
}
if tc.wantMPU {
if heads != 1 || starts != 1 || parts == 0 {
t.Errorf("invalid multipart flow: heads=%d starts=%d parts=%d", heads, starts, parts)
}
wantAborts, wantCompletes := 0, 1
if tc.wantErr {
wantAborts = 1
}
if tc.failPart {
wantCompletes = 0
}
if aborts != wantAborts || completes != wantCompletes {
t.Errorf("aborts/completes = %d/%d, want %d/%d", aborts, completes, wantAborts, wantCompletes)
}
} else {
if heads != 0 || starts != 0 || parts != 0 || completes != 0 || aborts != 0 {
t.Errorf("unexpected multipart activity: heads=%d starts=%d parts=%d completes=%d aborts=%d", heads, starts, parts, completes, aborts)
}
}
if tc.wantErr {
wanted = map[string]string{"mtime": "100", "custom": "retained"}
}
if !reflect.DeepEqual(stored, wanted) {
t.Errorf("stored metadata = %v, want %v", stored, wanted)
}
})
}
}
14 changes: 10 additions & 4 deletions core/cfg/flags.go
Original file line number Diff line number Diff line change
Expand Up @@ -277,7 +277,7 @@ MISC OPTIONS:

cli.BoolFlag{
Name: "no-detect",
Usage: "Turn off bucket location and signature algorithm autodetection on start",
Usage: "Turn off autodetection of anonymous access, bucket location and signature algorithm on start",
},

cli.BoolFlag{
Expand Down Expand Up @@ -621,8 +621,8 @@ MISC OPTIONS:

cli.IntFlag{
Name: "read-retry-attempts",
Value: 0,
Usage: "Maximum read retry attempts (0 means unlimited)",
Value: 10,
Usage: "Maximum read retry attempts (minimum: 1)",
},

cli.IntFlag{
Expand Down Expand Up @@ -874,6 +874,11 @@ func PopulateFlags(c *cli.Context) (ret *FlagStorage) {
singlePart = 5
}

readRetryAttempts := c.Int("read-retry-attempts")
if readRetryAttempts < 1 {
panic("--read-retry-attempts must be at least 1")
}

flags := &FlagStorage{
// File system
MountOptions: c.StringSlice("o"),
Expand Down Expand Up @@ -901,7 +906,7 @@ func PopulateFlags(c *cli.Context) (ret *FlagStorage) {
ReadRetryInterval: c.Duration("read-retry-interval"),
ReadRetryMultiplier: c.Float64("read-retry-mul"),
ReadRetryMax: c.Duration("read-retry-max-interval"),
ReadRetryAttempts: c.Int("read-retry-attempts"),
ReadRetryAttempts: readRetryAttempts,
ReadAheadKB: uint64(c.Int("read-ahead")),
SmallReadCount: uint64(c.Int("small-read-count")),
SmallReadCutoffKB: uint64(c.Int("small-read-cutoff")),
Expand Down Expand Up @@ -1130,6 +1135,7 @@ func DefaultFlags() *FlagStorage {
StatCacheTTL: 30 * time.Second,
HTTPTimeout: 30 * time.Second,
RetryInterval: 30 * time.Second,
ReadRetryAttempts: 10,
MaxDiskCacheFD: 512,
RefreshFilename: ".invalidate",
FlushFilename: ".fsyncdir",
Expand Down
Loading
Loading