From 2dbf9c910839e243a4d882602bff7f19a7cf2efe Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Thu, 29 Oct 2026 11:49:48 -0700 Subject: [PATCH] s3: evaluate all conditional write headers under the object lock (#12211) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * filer_pb,filer: per-mutation preconditions on ObjectTransaction A transaction's request-level condition evaluates one entry before any mutation runs, so a commit guarding two paths — the object it publishes and the upload directory it assembles — could not be expressed. Give each ObjectMutation an optional condition (evaluated against condition_key, else the mutation's target, under the same lock just before the mutation applies) and report which mutation failed through failed_mutation on the response, so callers can tell a mutation guard's rejection apart from the request-level precondition's. * s3: evaluate multipart completion conditions under the object lock CompleteMultipartUpload checked conditional headers gateway-side before committing, and the routed commit transaction guarded only the upload directory — nothing made If-None-Match: * atomic with the publish, so concurrent conditional completions of the same absent key could all report success. The request's reducible write condition (If-None-Match/If-Match, star or single strong ETag) now rides the commit as the transaction's request condition on the owner filer: on the object path for unversioned buckets, falling back to the .versions directory for suspended and enabled versioning. The upload-exists guard moves onto the PUT mutation, keeping its check-before-publish ordering. Conditions the reducer cannot express stay on the distributed write lock, where check and commit already serialize. * filer,s3: resolve the latest version first for versioned conditional writes * filer: keep directory entries holding content out of the absent-slot rule * s3: keep a gateway-side upload-exists check for filers without mutation conditions * filer: honor null-version-is-latest resolving prefer-alt conditions * drop scratch multipart-race repro programs * s3: note the write lock does not cover routed writes on non-reducible conditions * filer: mirror the gateway's remote-etag precedence in condition evaluation storedEntryETag only knew the extended attribute and the chunk fallback, so a synced remote entry compared chunk-derived ETags while the gateway compares the remote ETag. Follow the same precedence: extended attribute, remote ETag while the entry still holds the remote bytes, Md5, then the chunk hash — and no ETag for an entry holding none. * s3: reduce all conditional write headers into owner-lock clauses Time-based headers, combined conditions, ETag lists, and weak ETags stayed on the distributed write lock because buildWriteCondition could not express them, leaving their gateway check racing owner-routed writes. Reduce every parseable header form to WriteCondition clauses — multiple clauses compose AND, ETag candidates carry the raw list for the filer's canonicalization, and RFC 7232 precedence keeps If-Unmodified-Since hidden by If-Match and If-Modified-Since hidden by If-None-Match — so the owner filer evaluates them under the same object-path lock as the commit. Only unparseable headers still fall back, where the lock path reports the same parse error. --- weed/s3api/filer_multipart.go | 7 +- weed/s3api/s3api_object_handlers_put.go | 2 +- ...3api_object_handlers_put_ambiguous_test.go | 19 +++- weed/s3api/s3api_object_routed_write.go | 59 ++++++------ weed/s3api/s3api_object_routed_write_test.go | 91 +++++++++++++++---- weed/server/filer_grpc_server_condition.go | 23 ++++- .../filer_grpc_server_condition_test.go | 17 ++++ 7 files changed, 162 insertions(+), 56 deletions(-) diff --git a/weed/s3api/filer_multipart.go b/weed/s3api/filer_multipart.go index 753246698a..6be70b8adf 100644 --- a/weed/s3api/filer_multipart.go +++ b/weed/s3api/filer_multipart.go @@ -691,10 +691,9 @@ func (s3a *S3ApiServer) completeMultipartUpload(r *http.Request, input *s3.Compl routeKey := s3a.objectRouteKey(*input.Bucket, *input.Key) // The write condition rides the commit transaction so the owner filer // evaluates it under the object's path lock — the gateway's earlier - // checkConditionalHeaders is only a fast-path. Conditions it cannot reduce - // (time-based, combined, ETag lists, weak ETags) keep the pre-change - // semantics: check and commit run under the distributed write lock, which - // routed writes still bypass. + // checkConditionalHeaders is only a fast-path. Every header form reduces + // to clauses, so reducible is false only when the headers cannot parse, + // where the lock path reports the same parse error. cond, reducible := buildWriteCondition(r) completionBody := func() s3err.ErrorCode { var prepCode s3err.ErrorCode diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 502c73e7a5..8e1f3b720c 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -1067,7 +1067,7 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader } // Route the create to the object's owner filer (its per-path lock serializes - // it); conditional/object-lock/non-reducible cases fall back to the lock. + // it); unparseable or unroutable conditions fall back to the lock. var createCode s3err.ErrorCode routed := false routedAmbiguous := false diff --git a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go index 0d08dcdb7c..796b6e7ef5 100644 --- a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go +++ b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go @@ -287,9 +287,9 @@ func (f *ambiguousPutFiler) ObjectTransaction(_ context.Context, req *filer_pb.O case filer_pb.WriteCondition_IF_EXISTS: ok = current != nil case filer_pb.WriteCondition_IF_ETAG_MATCH: - ok = current != nil && slices.Contains(clause.Etags, md5Hex) + ok = current != nil && etagCandidateMatch(clause.Etags, md5Hex) case filer_pb.WriteCondition_IF_ETAG_NOT_MATCH: - ok = current == nil || !slices.Contains(clause.Etags, md5Hex) + ok = current == nil || !etagCandidateMatch(clause.Etags, md5Hex) } if !ok { return &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil @@ -317,6 +317,21 @@ func (f *ambiguousPutFiler) ObjectTransaction(_ context.Context, req *filer_pb.O return &filer_pb.ObjectTransactionResponse{}, f.txnErr } +// etagCandidateMatch mirrors the filer's strong-comparison canonicalization: +// quotes are stripped on both sides and a weak (W/) candidate never matches. +func etagCandidateMatch(candidates []string, stored string) bool { + for _, c := range candidates { + c = strings.TrimSpace(c) + if strings.HasPrefix(c, "W/") || strings.HasPrefix(c, "w/") { + continue + } + if strings.Trim(c, `"`) == stored { + return true + } + } + return false +} + // ensureParents mirrors the filer's InsertEntry, which creates missing parent // directory entries for a stored object (e.g. the .versions dir a version file // lands under). diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index 5c5b5bee7d..4b89d01239 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/weed/s3api/s3api_object_routed_write.go @@ -78,9 +78,13 @@ func routeWriteCondition(r *http.Request, uniqueWritePath bool) (*filer_pb.Write } // buildWriteCondition reduces the request's conditional headers to a -// WriteCondition. ok=false (combined headers, time conditions, ETag lists, weak -// ETags) keeps gateway-side evaluation under the lock; a nil condition with -// ok=true means unconditional. +// WriteCondition evaluated under the owner's object-path lock. Every header +// form reduces to clauses (ETag lists and weak ETags compare through the same +// canonicalization the gateway applies; time conditions become unix bounds), +// composed per RFC 7232 precedence: If-Unmodified-Since is ignored when +// If-Match is present, If-Modified-Since when If-None-Match is. ok=false only +// when the headers cannot even parse; a nil condition with ok=true means +// unconditional. func buildWriteCondition(r *http.Request) (*filer_pb.WriteCondition, bool) { headers, errCode := parseConditionalHeaders(r) if errCode != s3err.ErrNone { @@ -89,31 +93,24 @@ func buildWriteCondition(r *http.Request) (*filer_pb.WriteCondition, bool) { if !headers.isSet { return nil, true } - if !headers.ifModifiedSince.IsZero() || !headers.ifUnmodifiedSince.IsZero() { - return nil, false + var clauses []*filer_pb.WriteCondition_Clause + if headers.ifMatch == "*" { + clauses = append(clauses, &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_EXISTS}) + } else if headers.ifMatch != "" { + clauses = append(clauses, &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ETAG_MATCH, Etags: etagList(headers.ifMatch)}) } - hasMatch := headers.ifMatch != "" - hasNoneMatch := headers.ifNoneMatch != "" - switch { - case hasMatch && !hasNoneMatch: - if headers.ifMatch == "*" { - return clause(filer_pb.WriteCondition_IF_EXISTS), true - } - if etag, single := singleStrongETag(headers.ifMatch); single { - return etagClause(filer_pb.WriteCondition_IF_ETAG_MATCH, etag), true - } - return nil, false - case hasNoneMatch && !hasMatch: - if headers.ifNoneMatch == "*" { - return clause(filer_pb.WriteCondition_IF_NOT_EXISTS), true - } - if etag, single := singleStrongETag(headers.ifNoneMatch); single { - return etagClause(filer_pb.WriteCondition_IF_ETAG_NOT_MATCH, etag), true - } - return nil, false - default: - return nil, false + if !headers.ifUnmodifiedSince.IsZero() && headers.ifMatch == "" { + clauses = append(clauses, &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_UNMODIFIED_SINCE, UnixTime: headers.ifUnmodifiedSince.Unix()}) } + if headers.ifNoneMatch == "*" { + clauses = append(clauses, &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_NOT_EXISTS}) + } else if headers.ifNoneMatch != "" { + clauses = append(clauses, &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_ETAG_NOT_MATCH, Etags: etagList(headers.ifNoneMatch)}) + } + if !headers.ifModifiedSince.IsZero() && headers.ifNoneMatch == "" { + clauses = append(clauses, &filer_pb.WriteCondition_Clause{Kind: filer_pb.WriteCondition_IF_MODIFIED_SINCE, UnixTime: headers.ifModifiedSince.Unix()}) + } + return &filer_pb.WriteCondition{Clauses: clauses}, true } // buildDeleteCondition reduces a DeleteObject's If-Match header to a condition; @@ -154,6 +151,16 @@ func singleStrongETag(v string) (string, bool) { return strings.Trim(v, `"`), true } +// etagList splits a conditional ETag header into its candidates, each kept +// verbatim — the evaluator's canonicalization handles quotes and weak markers. +func etagList(v string) []string { + parts := strings.Split(v, ",") + for i := range parts { + parts[i] = strings.TrimSpace(parts[i]) + } + return parts +} + // objectTxnOnFiler applies a routed transaction, preferring the object's owner // filer but failing over to a live filer when the owner is unreachable, so a // restarted filer's stale ring address (e.g. a rolled K8s pod's old IP) can't hang diff --git a/weed/s3api/s3api_object_routed_write_test.go b/weed/s3api/s3api_object_routed_write_test.go index 19f202c865..ebd0602115 100644 --- a/weed/s3api/s3api_object_routed_write_test.go +++ b/weed/s3api/s3api_object_routed_write_test.go @@ -72,7 +72,7 @@ func TestBuildWriteCondition(t *testing.T) { t.Fatal("want ok") } c := oneClause(t, cond) - if c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || len(c.Etags) != 1 || c.Etags[0] != "abc123" { + if c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || len(c.Etags) != 1 || c.Etags[0] != `"abc123"` { t.Fatalf("clause = %+v", c) } }) @@ -82,33 +82,84 @@ func TestBuildWriteCondition(t *testing.T) { t.Fatal("want ok") } c := oneClause(t, cond) - if c.Kind != filer_pb.WriteCondition_IF_ETAG_NOT_MATCH || len(c.Etags) != 1 || c.Etags[0] != "abc123" { + if c.Kind != filer_pb.WriteCondition_IF_ETAG_NOT_MATCH || len(c.Etags) != 1 || c.Etags[0] != `"abc123"` { t.Fatalf("clause = %+v", c) } }) - t.Run("weak etag falls back", func(t *testing.T) { - if _, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `W/"abc"`})); ok { - t.Fatal("weak etag must not take the fast path") + t.Run("weak etag reduces verbatim", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `W/"abc"`})) + if !ok { + t.Fatal("want ok") + } + c := oneClause(t, cond) + if c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || len(c.Etags) != 1 || c.Etags[0] != `W/"abc"` || c.AllowWeak { + t.Fatalf("clause = %+v", c) } }) - t.Run("etag list falls back", func(t *testing.T) { - if _, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"a","b"`})); ok { - t.Fatal("etag list must not take the fast path") + t.Run("etag list reduces", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"a", "b"`})) + if !ok { + t.Fatal("want ok") + } + c := oneClause(t, cond) + if c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH || len(c.Etags) != 2 || c.Etags[0] != `"a"` || c.Etags[1] != `"b"` { + t.Fatalf("clause = %+v", c) } }) - t.Run("both match and none-match falls back", func(t *testing.T) { - if _, ok := buildWriteCondition(reqWith(map[string]string{ + t.Run("both match and none-match reduce to clauses", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(map[string]string{ s3_constants.IfMatch: "*", s3_constants.IfNoneMatch: "*", - })); ok { - t.Fatal("ambiguous combination must not take the fast path") + })) + if !ok { + t.Fatal("want ok") + } + if len(cond.Clauses) != 2 || cond.Clauses[0].Kind != filer_pb.WriteCondition_IF_EXISTS || cond.Clauses[1].Kind != filer_pb.WriteCondition_IF_NOT_EXISTS { + t.Fatalf("clauses = %+v", cond.Clauses) } }) - t.Run("time-based falls back", func(t *testing.T) { - if _, ok := buildWriteCondition(reqWith(map[string]string{ + t.Run("time-based reduces to a unix bound", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(map[string]string{ "If-Unmodified-Since": "Wed, 21 Oct 2015 07:28:00 GMT", + })) + if !ok { + t.Fatal("want ok") + } + c := oneClause(t, cond) + want := time.Date(2015, 10, 21, 7, 28, 0, 0, time.UTC).Unix() + if c.Kind != filer_pb.WriteCondition_IF_UNMODIFIED_SINCE || c.UnixTime != want { + t.Fatalf("clause = %+v, want unix %d", c, want) + } + }) + t.Run("If-Match hides If-Unmodified-Since", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(map[string]string{ + s3_constants.IfMatch: `"abc"`, + "If-Unmodified-Since": "Wed, 21 Oct 2015 07:28:00 GMT", + })) + if !ok { + t.Fatal("want ok") + } + if c := oneClause(t, cond); c.Kind != filer_pb.WriteCondition_IF_ETAG_MATCH { + t.Fatalf("clause = %+v", c) + } + }) + t.Run("If-None-Match hides If-Modified-Since", func(t *testing.T) { + cond, ok := buildWriteCondition(reqWith(map[string]string{ + s3_constants.IfNoneMatch: `"abc"`, + "If-Modified-Since": "Wed, 21 Oct 2015 07:28:00 GMT", + })) + if !ok { + t.Fatal("want ok") + } + if c := oneClause(t, cond); c.Kind != filer_pb.WriteCondition_IF_ETAG_NOT_MATCH { + t.Fatalf("clause = %+v", c) + } + }) + t.Run("unparseable date falls back", func(t *testing.T) { + if _, ok := buildWriteCondition(reqWith(map[string]string{ + "If-Modified-Since": "not a date", })); ok { - t.Fatal("time condition must not take the fast path") + t.Fatal("unparseable headers must not reduce") } }) } @@ -637,8 +688,12 @@ func TestRouteWriteCondition(t *testing.T) { if _, ok := routeWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `"e"`}), true); ok { t.Fatal("conditional unique write must not route") } - // A non-reducible condition bails regardless. - if _, ok := routeWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `W/"e"`}), false); ok { - t.Fatal("weak etag must not route") + // A weak ETag still routes — the clause carries it verbatim. + if _, ok := routeWriteCondition(reqWith(map[string]string{s3_constants.IfMatch: `W/"e"`}), false); !ok { + t.Fatal("weak etag should route") + } + // An unparseable condition bails regardless. + if _, ok := routeWriteCondition(reqWith(map[string]string{"If-Modified-Since": "not a date"}), false); ok { + t.Fatal("unparseable headers must not route") } } diff --git a/weed/server/filer_grpc_server_condition.go b/weed/server/filer_grpc_server_condition.go index dfb8039323..0eaca1b316 100644 --- a/weed/server/filer_grpc_server_condition.go +++ b/weed/server/filer_grpc_server_condition.go @@ -1,6 +1,7 @@ package weed_server import ( + "fmt" "strconv" "strings" "time" @@ -159,15 +160,27 @@ func canonicalETag(etag string) (value string, weak bool) { return strings.Trim(etag, `"`), false } -// storedEntryETag mirrors the S3 gateway's ETag precedence (the stored -// Seaweed ETag extended attribute, then the chunk/Md5 fallback) so conditional -// comparisons match what the gateway computes, without coupling the filer to -// S3 request handling. +// storedEntryETag mirrors the S3 gateway's ETag precedence — the stored +// Seaweed ETag extended attribute, then the remote ETag while the entry still +// holds the remote bytes, then Md5, then the chunk hash (an empty object has +// no ETag) — so conditional comparisons match what the gateway computes, +// without coupling the filer to S3 request handling. func storedEntryETag(entry *filer.Entry) string { if v, ok := entry.Extended[s3_constants.ExtETagKey]; ok && len(v) > 0 { return normalizeETag(string(v)) } - return normalizeETag(filer.ETagEntry(entry)) + if remote := entry.Remote; remote != nil && remote.RemoteETag != "" && + entry.FileSize == uint64(remote.RemoteSize) && + (remote.LastLocalSyncTsNs > 0 || (len(entry.GetChunks()) == 0 && len(entry.Content) == 0)) { + return normalizeETag(remote.RemoteETag) + } + if len(entry.Md5) > 0 { + return fmt.Sprintf("%x", entry.Md5) + } + if len(entry.GetChunks()) == 0 { + return "" + } + return normalizeETag(filer.ETagChunks(entry.GetChunks())) } func normalizeETag(etag string) string { diff --git a/weed/server/filer_grpc_server_condition_test.go b/weed/server/filer_grpc_server_condition_test.go index 33f0fc6bc3..c930ec80e1 100644 --- a/weed/server/filer_grpc_server_condition_test.go +++ b/weed/server/filer_grpc_server_condition_test.go @@ -257,6 +257,23 @@ func TestStoredEntryETag(t *testing.T) { if got := storedEntryETag(md5Only); got != "abcd" { t.Errorf("md5 fallback: got %q", got) } + remoteOnly := &filer.Entry{ + Attr: filer.Attr{FileSize: 10}, + Remote: &filer_pb.RemoteEntry{RemoteETag: `"remote-etag"`, RemoteSize: 10}, + } + if got := storedEntryETag(remoteOnly); got != "remote-etag" { + t.Errorf("remote etag: got %q", got) + } + remoteStale := &filer.Entry{ + Attr: filer.Attr{FileSize: 10, Md5: []byte{0xab}}, + Remote: &filer_pb.RemoteEntry{RemoteETag: `"remote-etag"`, RemoteSize: 99}, + } + if got := storedEntryETag(remoteStale); got != "ab" { + t.Errorf("stale remote size falls to md5: got %q", got) + } + if got := storedEntryETag(&filer.Entry{}); got != "" { + t.Errorf("empty entry: got %q", got) + } } // The CreateEntry handler enforces the precondition atomically: a matching