mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-30 08:50:48 +01:00
s3: evaluate all conditional write headers under the object lock (#12211)
* 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.
This commit is contained in:
1 parent
2737315e9d
commit
2dbf9c9108
7 files changed
+162
-56
No files matched your search
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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).
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in new issue
Block a user