diff --git a/weed/pb/filer_pb/filer_pb_helper.go b/weed/pb/filer_pb/filer_pb_helper.go index 15dc020cd..390747919 100644 --- a/weed/pb/filer_pb/filer_pb_helper.go +++ b/weed/pb/filer_pb/filer_pb_helper.go @@ -235,6 +235,7 @@ var ( ErrExistingIsFile = errors.New("existing entry is a file") ErrEntryAlreadyExists = errors.New("entry already exists") ErrReadOnly = errors.New("read only") + ErrPreconditionFailed = errors.New("precondition failed") ) // FilerErrorToSentinel maps a proto FilerError code to its sentinel error. @@ -253,6 +254,8 @@ func FilerErrorToSentinel(code FilerError) error { return ErrEntryAlreadyExists case FilerError_READ_ONLY: return ErrReadOnly + case FilerError_PRECONDITION_FAILED: + return ErrPreconditionFailed default: return nil } diff --git a/weed/s3api/s3_constants/header.go b/weed/s3api/s3_constants/header.go index 24a487151..28a2d1baf 100644 --- a/weed/s3api/s3_constants/header.go +++ b/weed/s3api/s3_constants/header.go @@ -170,6 +170,12 @@ const ( // rename, and it is dropped whenever the key is written again. SeaweedFSRenameToken = "x-seaweedfs-rename-token" + // SeaweedFSWriteId identifies the single write that stored an entry: putToFiler + // stamps a fresh id into Extended on every call. Recovery for an ambiguously + // committed write compares it to tell its own entry from an identical older + // object — chunk lists cannot, since an empty object carries none. + SeaweedFSWriteId = "x-seaweedfs-write-id" + // Multipart upload metadata keys for SSE-S3 SeaweedFSSSES3Encryption = "x-seaweedfs-sse-s3-encryption" // Encryption type for multipart upload SSE-S3 inheritance SeaweedFSSSES3BaseIV = "x-seaweedfs-sse-s3-base-iv" // Base IV for multipart upload SSE-S3 (for IV offset calculation) diff --git a/weed/s3api/s3api_handlers.go b/weed/s3api/s3api_handlers.go index 4c3dd7f57..baaf58435 100644 --- a/weed/s3api/s3api_handlers.go +++ b/weed/s3api/s3api_handlers.go @@ -6,10 +6,12 @@ import ( "errors" "fmt" "net/http" + "time" "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "google.golang.org/grpc" + "google.golang.org/grpc/connectivity" "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" @@ -30,7 +32,7 @@ func (s3a *S3ApiServer) WithFilerClient(streamingMode bool, fn func(filer_pb.Sea func (s3a *S3ApiServer) withFilerClient(ctx context.Context, streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { // Use filerClient for proper connection management and failover if s3a.filerClient != nil { - return s3a.withFilerClientFailover(ctx, "", streamingMode, fn) + return s3a.withFilerClientFailover(ctx, "", streamingMode, true, fn) } // Fallback to direct connection if filerClient not initialized @@ -56,7 +58,14 @@ func (s3a *S3ApiServer) withFilerClient(ctx context.Context, streamingMode bool, // caller decides whether a clean-slate retry is safe. ctx is the budget bounding // fn's own calls; it is only read, to tell a filer's failure apart from the // caller running out of time. -func (s3a *S3ApiServer) withFilerClientFailover(ctx context.Context, preferred pb.ServerAddress, streamingMode bool, fn func(filer_pb.SeaweedFilerClient) error) error { +// +// replayable=false is for requests that must not be re-sent once possibly +// applied (a committed transaction whose response was lost is +// indistinguishable from one that never arrived): such fn runs only on a +// channel that just proved ready, and any error it returns ends the walk — +// moving to the next filer is reserved for channels that never got ready, the +// one failure mode that provably sent nothing. +func (s3a *S3ApiServer) withFilerClientFailover(ctx context.Context, preferred pb.ServerAddress, streamingMode bool, replayable bool, fn func(filer_pb.SeaweedFilerClient) error) error { currentFiler := s3a.filerClient.GetCurrentFiler() candidates := make([]pb.ServerAddress, 0, 2+len(s3a.option.Filers)) @@ -99,10 +108,15 @@ func (s3a *S3ApiServer) withFilerClientFailover(ctx context.Context, preferred p var lastErr error for _, filer := range ordered { received := false + fnRan := false // Background, not ctx: WithGrpcClient's context only decides whether an // error invalidates the shared connection, and that call is fn's to make, // with the context fn's own RPC ran on. err := pb.WithGrpcClient(context.Background(), streamingMode, s3a.randomClientId, func(grpcConnection *grpc.ClientConn) error { + if !replayable && !filerConnReady(ctx, grpcConnection) { + return errFilerConnNotReady + } + fnRan = true return fn(filer_pb.NewSeaweedFilerClient(receiveTrackingConn{ClientConnInterface: grpcConnection, received: &received})) }, filer.ToGrpcAddress(), false, s3a.option.GrpcDialOption) @@ -127,6 +141,12 @@ func (s3a *S3ApiServer) withFilerClientFailover(ctx context.Context, preferred p return err } + // fn consumed part of a response; a replay would stack a second copy + // onto whatever it accumulated, so surface the error unwrapped. The filer + // answered, so this also says nothing about its health. + if received { + return err + } s3a.filerClient.RecordFilerFailure(filer) // A preferred owner is often outside the static filer list, where the health // tracking above no-ops; flag it so route-by-key reads skip it briefly. @@ -134,9 +154,9 @@ func (s3a *S3ApiServer) withFilerClientFailover(ctx context.Context, preferred p s3a.markOwnerUnreachable(filer) } glog.V(2).Infof("WithFilerClient: filer %s failed: %v", filer, err) - // fn consumed part of a response; a replay would stack a second copy - // onto whatever it accumulated, so surface the error unwrapped. - if received { + // A non-replayable fn that ran may have committed server-side; only a + // channel that never became ready proves nothing was sent. + if !replayable && fnRan { return err } lastErr = err @@ -148,6 +168,32 @@ func (s3a *S3ApiServer) withFilerClientFailover(ctx context.Context, preferred p return fmt.Errorf("all filers failed, last error: %w", lastErr) } +// errFilerConnNotReady marks a candidate whose channel never became usable, so +// the request was provably never sent and the walk may continue. +var errFilerConnNotReady = errors.New("filer connection not ready") + +// filerConnReadyWait bounds how long a candidate gets to bring its channel up +// before the walk tries the next filer. +const filerConnReadyWait = 2 * time.Second + +func filerConnReady(ctx context.Context, conn *grpc.ClientConn) bool { + readyCtx, cancel := context.WithTimeout(ctx, filerConnReadyWait) + defer cancel() + conn.Connect() + for { + s := conn.GetState() + if s == connectivity.Ready { + return true + } + if s == connectivity.TransientFailure || s == connectivity.Shutdown { + return false + } + if !conn.WaitForStateChange(readyCtx, s) { + return false + } + } +} + // receiveTrackingConn flags *received once a unary reply or a streamed message // has been handed to the callback, the point past which a failover replay is // no longer transparent. diff --git a/weed/s3api/s3api_object_ambiguous_route_test.go b/weed/s3api/s3api_object_ambiguous_route_test.go index 1aafe247a..cc7d5c88a 100644 --- a/weed/s3api/s3api_object_ambiguous_route_test.go +++ b/weed/s3api/s3api_object_ambiguous_route_test.go @@ -41,7 +41,7 @@ func TestCreateAfterAmbiguousRoute(t *testing.T) { s3a, owner := newS3a(t, f) entryCreated := false ran := false - code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func() s3err.ErrorCode { ran = true; return s3err.ErrNone }) + code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func(*filer_pb.WriteCondition) s3err.ErrorCode { ran = true; return s3err.ErrNone }) if code != s3err.ErrNone || !entryCreated { t.Fatalf("code=%v entryCreated=%v", code, entryCreated) } @@ -55,7 +55,7 @@ func TestCreateAfterAmbiguousRoute(t *testing.T) { s3a, owner := newS3a(t, f) entryCreated := false ran := false - code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func() s3err.ErrorCode { ran = true; return s3err.ErrNone }) + code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func(*filer_pb.WriteCondition) s3err.ErrorCode { ran = true; return s3err.ErrNone }) if code != s3err.ErrServiceUnavailable { t.Fatalf("code = %v, want ServiceUnavailable", code) } @@ -65,14 +65,24 @@ func TestCreateAfterAmbiguousRoute(t *testing.T) { }) t.Run("stored entry is a directory, chunks alive — lock path answers", func(t *testing.T) { - f := &fakeLookupFiler{entry: &filer_pb.Entry{Name: "o", IsDirectory: true}} + dir := &filer_pb.Entry{Name: "o", IsDirectory: true} + f := &fakeLookupFiler{entry: dir} s3a, owner := newS3a(t, f) entryCreated := false ran := false - code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func() s3err.ErrorCode { ran = true; return s3err.ErrExistingObjectIsDirectory }) + var gotCond *filer_pb.WriteCondition + code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func(cond *filer_pb.WriteCondition) s3err.ErrorCode { + ran = true + gotCond = cond + return s3err.ErrExistingObjectIsDirectory + }) if code != s3err.ErrExistingObjectIsDirectory || !ran { t.Fatalf("code=%v ran=%v — a directory conflict with live chunks must reach createUnderLock, which maps it", code, ran) } + c := oneClause(t, gotCond) + if c.Kind != filer_pb.WriteCondition_IF_ENTRY_EQUAL || !c.ExpectedEntry.IsDirectory { + t.Fatalf("directory retry carried %+v, want an IF_ENTRY_EQUAL pin on the directory", c) + } }) // The routed PUT may have committed before a delete removed the entry and @@ -83,7 +93,7 @@ func TestCreateAfterAmbiguousRoute(t *testing.T) { s3a, owner := newS3a(t, f) entryCreated := false ran := false - code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksDead, func() s3err.ErrorCode { ran = true; return s3err.ErrNone }) + code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksDead, func(*filer_pb.WriteCondition) s3err.ErrorCode { ran = true; return s3err.ErrNone }) if code != s3err.ErrServiceUnavailable || ran { t.Fatalf("code=%v ran=%v — re-committed over a directory with possibly-deleted chunks", code, ran) } @@ -95,21 +105,25 @@ func TestCreateAfterAmbiguousRoute(t *testing.T) { s3a := ambiguousRouteServer(t, deadAddr, matchAddr) entryCreated := false ran := false - code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", deadAddr, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func() s3err.ErrorCode { ran = true; return s3err.ErrNone }) + code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", deadAddr, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func(*filer_pb.WriteCondition) s3err.ErrorCode { ran = true; return s3err.ErrNone }) if code != s3err.ErrServiceUnavailable || entryCreated || ran { t.Fatalf("code=%v entryCreated=%v ran=%v — a match behind an unresolved lookup must not finalize", code, entryCreated, ran) } }) - t.Run("entry proven absent, chunks alive — normal create proceeds", func(t *testing.T) { + t.Run("entry proven absent, chunks alive — conditional create proceeds", func(t *testing.T) { f := &fakeLookupFiler{} s3a, owner := newS3a(t, f) entryCreated := false ran := false - code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func() s3err.ErrorCode { ran = true; return s3err.ErrNone }) + var gotCond *filer_pb.WriteCondition + code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func(cond *filer_pb.WriteCondition) s3err.ErrorCode { ran = true; gotCond = cond; return s3err.ErrNone }) if code != s3err.ErrNone || !ran { t.Fatalf("code=%v ran=%v — proven-absent entry did not reach createUnderLock", code, ran) } + if c := oneClause(t, gotCond); c.Kind != filer_pb.WriteCondition_IF_NOT_EXISTS { + t.Fatalf("absent-entry create carried %+v, want IF_NOT_EXISTS", c) + } }) t.Run("entry absent but chunks unverifiable — refuse", func(t *testing.T) { @@ -117,7 +131,7 @@ func TestCreateAfterAmbiguousRoute(t *testing.T) { s3a, owner := newS3a(t, f) entryCreated := false ran := false - code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksDead, func() s3err.ErrorCode { ran = true; return s3err.ErrNone }) + code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksDead, func(*filer_pb.WriteCondition) s3err.ErrorCode { ran = true; return s3err.ErrNone }) if code != s3err.ErrServiceUnavailable || ran { t.Fatalf("code=%v ran=%v — re-committed an entry over possibly-deleted chunks", code, ran) } @@ -128,7 +142,7 @@ func TestCreateAfterAmbiguousRoute(t *testing.T) { s3a, owner := newS3a(t, f) entryCreated := false ran := false - code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func() s3err.ErrorCode { ran = true; return s3err.ErrNone }) + code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, nil, &entryCreated, chunksAlive, func(*filer_pb.WriteCondition) s3err.ErrorCode { ran = true; return s3err.ErrNone }) if code != s3err.ErrServiceUnavailable || ran { t.Fatalf("code=%v ran=%v — uncertain state still committed", code, ran) } @@ -140,7 +154,7 @@ func TestCreateAfterAmbiguousRoute(t *testing.T) { entryCreated := false ran := false finalize := &putFinalize{afterCreate: func(*filer_pb.Entry) s3err.ErrorCode { return s3err.ErrInternalError }} - code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, finalize, &entryCreated, chunksAlive, func() s3err.ErrorCode { ran = true; return s3err.ErrNone }) + code := s3a.createAfterAmbiguousRoute(filePath, "b", "o", owner, &filer_pb.Entry{Name: "o"}, uploaded, finalize, &entryCreated, chunksAlive, func(*filer_pb.WriteCondition) s3err.ErrorCode { ran = true; return s3err.ErrNone }) if code != s3err.ErrInternalError || ran { t.Fatalf("code=%v ran=%v", code, ran) } diff --git a/weed/s3api/s3api_object_handlers_put.go b/weed/s3api/s3api_object_handlers_put.go index 9b2af5a53..527f46a80 100644 --- a/weed/s3api/s3api_object_handlers_put.go +++ b/weed/s3api/s3api_object_handlers_put.go @@ -18,6 +18,7 @@ import ( "strings" "time" + "github.com/google/uuid" "github.com/pquerna/cachecontrol/cacheobject" "github.com/seaweedfs/seaweedfs/weed/filer" "github.com/seaweedfs/seaweedfs/weed/glog" @@ -850,6 +851,11 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader // Store ETag in Extended attribute for future retrieval (e.g. multipart parts) entry.Extended[s3_constants.ExtETagKey] = []byte(etag) + // Stamp this write's identity: an ambiguously committed PUT is told apart + // from an identical older object by it during recovery, which chunk identity + // cannot do for a chunkless (empty) object. + entry.Extended[s3_constants.SeaweedFSWriteId] = []byte(uuid.NewString()) + // Store additional checksum if one was computed checksumBase64 := "" if checksumHash != nil && checksumHeaderName != "" { @@ -969,17 +975,19 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader var createErr error var rollbackErr error entryCreated := false + var owner pb.ServerAddress preconditionFn := func() s3err.ErrorCode { if object == "" { return s3err.ErrNone } return s3a.checkConditionalHeaders(r, bucket, object) } - createUnderLock := func() s3err.ErrorCode { - createErr = s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error { + createUnderLock := func(cond *filer_pb.WriteCondition) s3err.ErrorCode { + create := func(client filer_pb.SeaweedFilerClient) error { req := &filer_pb.CreateEntryRequest{ Directory: path.Dir(filePath), Entry: entry, + Condition: cond, } glog.V(3).Infof("putToFiler: Calling CreateEntry for %s", filePath) err := filer_pb.CreateEntry(context.Background(), client, req) @@ -994,7 +1002,27 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader return err } return nil - }) + } + // A conditional create must not be replayed once possibly sent: a lost + // commit followed by a delete lets the replayed IF_NOT_EXISTS pass and + // recreates an entry whose chunks the delete already reclaimed. + if cond != nil && owner != "" { + // The condition and the routed write serialize on the owner's + // per-path lock. Failover cannot preserve that: another filer's lock + // is a different boundary, and its forwarding is only correct when + // its ring view agrees — exactly what is not guaranteed after an + // ambiguous write. The create goes to the owner alone; an + // unreachable owner leaves the outcome uncertain. + createErr = pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, create) + switch status.Code(createErr) { + case codes.Unavailable, codes.Canceled, codes.DeadlineExceeded: + s3a.markOwnerUnreachable(owner) + } + } else if s3a.filerClient != nil { + createErr = s3a.withFilerClientFailover(context.Background(), "", false, cond == nil, create) + } else { + createErr = s3a.withFilerClient(context.Background(), false, create) + } if createErr != nil { return filerErrorToS3Error(createErr) } @@ -1018,7 +1046,6 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader var createCode s3err.ErrorCode routed := false routedAmbiguous := false - var owner pb.ServerAddress condAltKey := "" if finalize != nil { condAltKey = finalize.conditionAltKey @@ -1056,8 +1083,8 @@ func (s3a *S3ApiServer) putToFiler(r *http.Request, filePath string, dataReader } } if !routed { - createFn := createUnderLock - if routedAmbiguous && len(chunkResult.FileChunks) > 0 { + createFn := func() s3err.ErrorCode { return createUnderLock(nil) } + if routedAmbiguous { createFn = func() s3err.ErrorCode { return s3a.createAfterAmbiguousRoute(filePath, bucket, object, owner, entry, chunkResult.FileChunks, finalize, &entryCreated, s3a.uploadedChunksExist, createUnderLock) } @@ -1131,62 +1158,96 @@ const createLookupTimeout = 10 * time.Second // superseded the commit, and its old-chunk cleanup may have deleted this // entry's chunks — committing would write back a stale entry pointing at dead // needles, so the request fails for the client to retry with a fresh upload. -// Unresolvable lookups fail the same way; only a proven-absent entry or a -// directory conflict with verifiably-live chunks falls through to the normal -// create. -func (s3a *S3ApiServer) createAfterAmbiguousRoute(filePath, bucket, object string, owner pb.ServerAddress, entry *filer_pb.Entry, uploaded []*filer_pb.FileChunk, finalize *putFinalize, entryCreated *bool, chunksExist func(ctx context.Context, chunks []*filer_pb.FileChunk) bool, createUnderLock func() s3err.ErrorCode) s3err.ErrorCode { +// Unresolvable lookups fail the same way; only a proven-absent entry with +// verifiably-live chunks or a directory conflict with them falls through to +// the normal create — an empty upload has nothing to verify, so neither the +// absent nor the lookalike-entry outcome is provable for it. +func (s3a *S3ApiServer) createAfterAmbiguousRoute(filePath, bucket, object string, owner pb.ServerAddress, entry *filer_pb.Entry, uploaded []*filer_pb.FileChunk, finalize *putFinalize, entryCreated *bool, chunksExist func(ctx context.Context, chunks []*filer_pb.FileChunk) bool, createUnderLock func(cond *filer_pb.WriteCondition) s3err.ErrorCode) s3err.ErrorCode { dir, name := path.Dir(filePath), path.Base(filePath) lookupCtx, cancel := context.WithTimeout(context.Background(), createLookupTimeout) defer cancel() - existing, uncertain := s3a.lookupCommittedEntry(lookupCtx, s3a.createTargetFilers(owner, bucket, object), dir, name) - if existing != nil { - if uncertain { - // A filer earlier in the list could not answer and may hold a - // newer entry; this match is not authoritative. - glog.Warningf("putToFiler: ambiguous routed PUT for %s matches an entry but a filer could not be checked; not recovering", filePath) - return s3err.ErrServiceUnavailable - } - if existing.IsDirectory { - // The directory can postdate this PUT's commit: the route may - // have stored the file entry, a delete removed it, and a nested - // write recreated the directory before recovery took the lock. - // Re-committing is only safe while the uploaded chunks verifiably - // survive — that delete may already have reclaimed the needles, - // and a create retrying over the directory would store an entry - // pointing at dead chunks. - if !chunksExist(lookupCtx, uploaded) { - glog.Warningf("putToFiler: ambiguous routed PUT for %s resolved to a directory and its chunks cannot be verified; not re-applying entry", filePath) + // The verified state and the create are made atomic by a WriteCondition on + // the create itself, which the filer evaluates under its per-path lock — a + // routed write committing in between fails the precondition instead of + // being clobbered, and the next pass re-derives the verdict from what the + // racer left. + for attempt := 0; attempt < 2; attempt++ { + existing, uncertain := s3a.lookupCommittedEntry(lookupCtx, s3a.createTargetFilers(owner, bucket, object), dir, name) + if existing != nil { + if uncertain { + // A filer earlier in the list could not answer and may hold a + // newer entry; this match is not authoritative. + glog.Warningf("putToFiler: ambiguous routed PUT for %s matches an entry but a filer could not be checked; not recovering", filePath) return s3err.ErrServiceUnavailable } - return createUnderLock() + if existing.IsDirectory { + // The directory can postdate this PUT's commit: the route may + // have stored the file entry, a delete removed it, and a nested + // write recreated the directory before recovery took the lock. + // Re-committing is only safe while the uploaded chunks verifiably + // survive — that delete may already have reclaimed the needles, + // and a create retrying over the directory would store an entry + // pointing at dead chunks. + if !chunksExist(lookupCtx, uploaded) { + glog.Warningf("putToFiler: ambiguous routed PUT for %s resolved to a directory and its chunks cannot be verified; not re-applying entry", filePath) + return s3err.ErrServiceUnavailable + } + code := createUnderLock(&filer_pb.WriteCondition{Clauses: []*filer_pb.WriteCondition_Clause{{ + Kind: filer_pb.WriteCondition_IF_ENTRY_EQUAL, + ExpectedEntry: existing, + }}}) + if code != s3err.ErrPreconditionFailed { + return code + } + continue + } + resolved, _, resolveErr := filer.ResolveChunkManifest(lookupCtx, s3a.createLookupFileIdFunction(), existing.GetChunks(), 0, math.MaxInt64, s3a.filerClient) + if resolveErr != nil || !sameFileChunks(resolved, uploaded) { + glog.Warningf("putToFiler: ambiguous routed PUT for %s was superseded by another write; not committing stale entry", filePath) + return s3err.ErrServiceUnavailable + } + // An empty PUT carries no fids, so two chunkless entries compare + // equal even when the stored one predates this write; only this + // write's stamped id proves the entry is its own commit. + if len(uploaded) == 0 && !sameStoredEntry(existing, entry) { + glog.Warningf("putToFiler: ambiguous routed PUT for %s cannot prove an empty stored entry is its own; not claiming it", filePath) + return s3err.ErrServiceUnavailable + } + *entryCreated = true + if finalize != nil && finalize.afterCreate != nil { + if code := finalize.afterCreate(entry); code != s3err.ErrNone { + // Same undo the create path applies when post-create + // finalization fails. + if rbErr := s3a.rmObject(context.Background(), dir, name, true, false); rbErr != nil { + glog.Errorf("putToFiler: failed to rollback recovered entry for %s: %v", filePath, rbErr) + } + return code + } + } + return s3err.ErrNone } - resolved, _, resolveErr := filer.ResolveChunkManifest(lookupCtx, s3a.createLookupFileIdFunction(), existing.GetChunks(), 0, math.MaxInt64, s3a.filerClient) - if resolveErr != nil || !sameFileChunks(resolved, uploaded) { - glog.Warningf("putToFiler: ambiguous routed PUT for %s was superseded by another write; not committing stale entry", filePath) + if uncertain { + glog.Warningf("putToFiler: cannot confirm what ambiguous routed PUT for %s committed; not re-applying entry", filePath) return s3err.ErrServiceUnavailable } - *entryCreated = true - if finalize != nil && finalize.afterCreate != nil { - if code := finalize.afterCreate(entry); code != s3err.ErrNone { - // Same undo the create path applies when post-create - // finalization fails. - if rbErr := s3a.rmObject(context.Background(), dir, name, true, false); rbErr != nil { - glog.Errorf("putToFiler: failed to rollback recovered entry for %s: %v", filePath, rbErr) - } - return code - } + // An absent entry cannot prove an empty PUT never committed — the routed + // create may have landed and a delete already reclaimed it — and there + // are no chunks to verify either way. IF_NOT_EXISTS would recreate an + // object a delete already removed, so the outcome stays uncertain. + if len(uploaded) == 0 { + glog.Warningf("putToFiler: ambiguous routed PUT for %s left no entry and no chunks to verify; not re-applying entry", filePath) + return s3err.ErrServiceUnavailable + } + if !chunksExist(lookupCtx, uploaded) { + glog.Warningf("putToFiler: ambiguous routed PUT for %s left no entry and its chunks cannot be verified; not re-applying entry", filePath) + return s3err.ErrServiceUnavailable + } + code := createUnderLock(clause(filer_pb.WriteCondition_IF_NOT_EXISTS)) + if code != s3err.ErrPreconditionFailed { + return code } - return s3err.ErrNone } - if uncertain { - glog.Warningf("putToFiler: cannot confirm what ambiguous routed PUT for %s committed; not re-applying entry", filePath) - return s3err.ErrServiceUnavailable - } - if !chunksExist(lookupCtx, uploaded) { - glog.Warningf("putToFiler: ambiguous routed PUT for %s left no entry and its chunks cannot be verified; not re-applying entry", filePath) - return s3err.ErrServiceUnavailable - } - return createUnderLock() + return s3err.ErrServiceUnavailable } // lookupCommittedEntry returns the entry any of the candidate filers stores @@ -1275,11 +1336,16 @@ func (s3a *S3ApiServer) confirmCreateLanded(filePath, bucket, object string, ent return s3err.ErrNone } if len(uploaded) == 0 { - return s3err.ErrNone - } - resolved, _, resolveErr := filer.ResolveChunkManifest(lookupCtx, s3a.createLookupFileIdFunction(), existing.GetChunks(), 0, math.MaxInt64, s3a.filerClient) - if resolveErr != nil || !sameFileChunks(resolved, uploaded) { - return s3err.ErrNone + // No chunks to compare; only this write's stamped id proves the + // entry is its own commit rather than an identical older object. + if !sameStoredEntry(existing, entry) { + return s3err.ErrNone + } + } else { + resolved, _, resolveErr := filer.ResolveChunkManifest(lookupCtx, s3a.createLookupFileIdFunction(), existing.GetChunks(), 0, math.MaxInt64, s3a.filerClient) + if resolveErr != nil || !sameFileChunks(resolved, uploaded) { + return s3err.ErrNone + } } glog.Warningf("putToFiler: create entry for %s failed but the entry exists, treating the write as successful", filePath) if finalize == nil || finalize.afterCreate == nil { @@ -1356,6 +1422,16 @@ func sameFileChunks(a, b []*filer_pb.FileChunk) bool { return true } +// sameStoredEntry reports whether the stored entry is this write's own +// commit. Matching metadata cannot say so — an identical older object carries +// the same fields — but the write id putToFiler stamps into every entry does: +// the filer stores Extended verbatim, so the id rides the commit and a lookup +// reads it back. An entry without this write's id cannot be claimed. +func sameStoredEntry(existing, entry *filer_pb.Entry) bool { + id := entry.GetExtended()[s3_constants.SeaweedFSWriteId] + return len(id) > 0 && bytes.Equal(existing.GetExtended()[s3_constants.SeaweedFSWriteId], id) +} + // checksumAlgorithmMapping maps algorithm name strings to their enum and header name. var checksumAlgorithmMapping = map[string]struct { alg ChecksumAlgorithm @@ -1642,6 +1718,8 @@ func filerErrorToS3Error(err error) s3err.ErrorCode { return s3err.ErrExistingObjectIsDirectory case errors.Is(err, weed_server.ErrReadOnly): return s3err.ErrAccessDenied + case errors.Is(err, filer_pb.ErrPreconditionFailed): + return s3err.ErrPreconditionFailed } // A transport failure leaves the outcome ambiguous — the write may have diff --git a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go index b2c236f9a..7c2b9ae7c 100644 --- a/weed/s3api/s3api_object_handlers_put_ambiguous_test.go +++ b/weed/s3api/s3api_object_handlers_put_ambiguous_test.go @@ -3,6 +3,7 @@ package s3api import ( "bytes" "context" + "crypto/md5" "fmt" "io" "net" @@ -88,7 +89,21 @@ func startFakeVolumeServer(t *testing.T) *fakeVolumeServer { w.Write(data) return } - io.Copy(io.Discard, r.Body) + if r.Method == http.MethodHead { + v.mu.Lock() + _, ok := v.stored[fid] + v.mu.Unlock() + if !ok { + w.WriteHeader(http.StatusNotFound) + return + } + w.WriteHeader(http.StatusOK) + return + } + data, _ := io.ReadAll(r.Body) + v.mu.Lock() + v.stored[fid] = data + v.mu.Unlock() w.Header().Set("Content-MD5", r.Header.Get("Content-MD5")) w.WriteHeader(http.StatusCreated) io.WriteString(w, `{"size":1}`) @@ -125,6 +140,20 @@ type ambiguousPutFiler struct { lookupErr error lookupFailKey string nextKey uint64 + // supersede, once, swaps in a concurrent write's entry as the routed + // transaction commits — issue 11631's client retry. txnNoApply models a + // transaction that failed before applying (e.g. its forward was refused). + supersede *filer_pb.Entry + txnErr error + txnNoApply bool + // beforeCreate runs inside CreateEntry's lock, before the condition is + // evaluated, letting a test land a racing write at the worst moment. + // beforeLookup runs inside LookupDirectoryEntry, letting a test delete a + // just-committed entry before recovery reads it. + beforeCreate func() + beforeLookup func() + stores int + conds []*filer_pb.WriteCondition } func (f *ambiguousPutFiler) AssignVolume(context.Context, *filer_pb.AssignVolumeRequest) (*filer_pb.AssignVolumeResponse, error) { @@ -145,10 +174,19 @@ func (f *ambiguousPutFiler) AssignVolume(context.Context, *filer_pb.AssignVolume func (f *ambiguousPutFiler) CreateEntry(_ context.Context, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) { f.mu.Lock() defer f.mu.Unlock() + if f.beforeCreate != nil { + f.beforeCreate() + } + f.conds = append(f.conds, req.Condition) + key := req.Directory + "/" + req.Entry.Name + if !putConditionSatisfied(req.Condition, f.entries[key]) { + return &filer_pb.CreateEntryResponse{Error: "precondition failed", ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil + } if f.apply { entry := proto.Clone(req.Entry).(*filer_pb.Entry) filer_pb.BeforeEntrySerialization(entry.Chunks) - f.entries[req.Directory+"/"+req.Entry.Name] = entry + f.entries[key] = entry + f.stores++ } if f.createErr != nil { return nil, f.createErr @@ -156,9 +194,42 @@ func (f *ambiguousPutFiler) CreateEntry(_ context.Context, req *filer_pb.CreateE return &filer_pb.CreateEntryResponse{Error: f.respError}, nil } +// putConditionSatisfied mirrors the filer's clauseSatisfied for the condition +// kinds the ambiguous-route recovery emits. +func putConditionSatisfied(cond *filer_pb.WriteCondition, current *filer_pb.Entry) bool { + for _, c := range cond.GetClauses() { + switch c.Kind { + case filer_pb.WriteCondition_IF_NOT_EXISTS: + if current != nil { + return false + } + case filer_pb.WriteCondition_IF_ENTRY_EQUAL: + if current == nil || c.ExpectedEntry == nil { + if current != nil || c.ExpectedEntry != nil { + return false + } + continue + } + expected := proto.Clone(c.ExpectedEntry).(*filer_pb.Entry) + filer_pb.BeforeEntrySerialization(expected.Chunks) + actual := proto.Clone(current).(*filer_pb.Entry) + filer_pb.BeforeEntrySerialization(actual.Chunks) + if !proto.Equal(filer.FromPbEntry("", actual).ToProtoEntry(), filer.FromPbEntry("", expected).ToProtoEntry()) { + return false + } + default: + return false + } + } + return true +} + func (f *ambiguousPutFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { f.mu.Lock() defer f.mu.Unlock() + if f.beforeLookup != nil { + f.beforeLookup() + } if f.lookupErr != nil && req.Directory+"/"+req.Name == f.lookupFailKey { return nil, f.lookupErr } @@ -220,12 +291,20 @@ func (f *ambiguousPutFiler) ObjectTransaction(_ context.Context, req *filer_pb.O return &filer_pb.ObjectTransactionResponse{ErrorCode: filer_pb.FilerError_PRECONDITION_FAILED}, nil } } - for _, m := range req.Mutations { - if m.Type == filer_pb.ObjectMutation_PUT { - f.entries[m.Directory+"/"+m.Entry.Name] = proto.Clone(m.Entry).(*filer_pb.Entry) + if !f.txnNoApply { + for _, m := range req.Mutations { + if m.Type == filer_pb.ObjectMutation_PUT { + entry := proto.Clone(m.Entry).(*filer_pb.Entry) + filer_pb.BeforeEntrySerialization(entry.Chunks) + f.entries[m.Directory+"/"+m.Entry.Name] = entry + } } } - return &filer_pb.ObjectTransactionResponse{}, nil + if f.supersede != nil { + f.entries[req.LockKey] = f.supersede + f.supersede = nil + } + return &filer_pb.ObjectTransactionResponse{}, f.txnErr } func (f *ambiguousPutFiler) ListEntries(req *filer_pb.ListEntriesRequest, stream filer_pb.SeaweedFiler_ListEntriesServer) error { @@ -442,6 +521,372 @@ func TestPutToFilerUnverifiableCreateKeepsChunks(t *testing.T) { } } +// Issue 11631: a routed PUT committed on the owner but its response was lost; +// a client retry then superseded the entry and the retry's old-chunk cleanup +// deleted this PUT's chunks. Re-committing the stale entry would resurrect a +// pointer to dead needles — the recovery must leave the newer write alone. +func TestPutToFilerRoutedAmbiguousKeepsNewerWrite(t *testing.T) { + volume := startFakeVolumeServer(t) + newer := &filer_pb.Entry{ + Name: "o", + Attributes: &filer_pb.FuseAttributes{FileSize: 11}, + Chunks: []*filer_pb.FileChunk{{FileId: "3,00000000000000bb0000000b", Size: 11}}, + } + filerImpl := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + supersede: newer, + txnErr: status.Error(codes.Unavailable, "owner restarted mid-write"), + } + filerAddr := startFakeFiler(t, filerImpl) + s3a := newPutTestServer(t, filerAddr) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + _, code := putTestObject(t, s3a) + if code != s3err.ErrServiceUnavailable { + t.Fatalf("putToFiler returned %v, want refusal once a newer write is found", code) + } + if filerImpl.stores != 0 { + t.Fatalf("recovery re-committed the stale entry %d times", filerImpl.stores) + } + if got := filerImpl.entries["/buckets/b/o"]; got == nil || len(got.Chunks) == 0 || got.Chunks[0].FileId != newer.Chunks[0].FileId { + t.Fatalf("the newer write's entry was clobbered: %+v", got) + } + if deleted := volume.deleted(); len(deleted) != 0 { + t.Fatalf("chunks were deleted during recovery: %v", deleted) + } +} + +// Issue 11631, raced variant: the routed PUT never committed and the recovery +// lookup finds the entry absent, but a concurrent write lands before the +// fallback create. The create's precondition — evaluated under the owner's +// path lock — refuses instead of overwriting the racer. +func TestPutToFilerRoutedRecoveryLosesRace(t *testing.T) { + volume := startFakeVolumeServer(t) + newer := &filer_pb.Entry{ + Name: "o", + Attributes: &filer_pb.FuseAttributes{FileSize: 11}, + Chunks: []*filer_pb.FileChunk{{FileId: "3,00000000000000bb0000000b", Size: 11}}, + } + filerImpl := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + apply: true, + txnNoApply: true, + txnErr: status.Error(codes.Unavailable, "response lost"), + } + filerImpl.beforeCreate = func() { + filerImpl.entries["/buckets/b/o"] = newer + filerImpl.beforeCreate = nil + } + filerAddr := startFakeFiler(t, filerImpl) + s3a := newPutTestServer(t, filerAddr) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + _, code := putTestObject(t, s3a) + if code != s3err.ErrServiceUnavailable { + t.Fatalf("putToFiler returned %v, want refusal after losing the create race", code) + } + if filerImpl.stores != 0 { + t.Fatalf("the raced recovery create stored %d entries", filerImpl.stores) + } + if len(filerImpl.conds) == 0 || filerImpl.conds[0].GetClauses()[0].Kind != filer_pb.WriteCondition_IF_NOT_EXISTS { + t.Fatalf("recovery create carried no IF_NOT_EXISTS precondition: %+v", filerImpl.conds) + } + if got := filerImpl.entries["/buckets/b/o"]; got == nil || len(got.Chunks) == 0 || got.Chunks[0].FileId != newer.Chunks[0].FileId { + t.Fatalf("the racing write's entry was clobbered: %+v", got) + } +} + +// Same incident, but the ambiguous PUT wrote a zero-byte object: recovery must +// still run — an unconditional fallback would overwrite the newer write even +// though there were no chunks to verify. +func TestPutToFilerRoutedAmbiguousEmptyPutKeepsNewerWrite(t *testing.T) { + volume := startFakeVolumeServer(t) + newer := &filer_pb.Entry{ + Name: "o", + Attributes: &filer_pb.FuseAttributes{FileSize: 11}, + Chunks: []*filer_pb.FileChunk{{FileId: "3,00000000000000bb0000000b", Size: 11}}, + } + filerImpl := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + supersede: newer, + txnErr: status.Error(codes.Unavailable, "owner restarted mid-write"), + } + filerAddr := startFakeFiler(t, filerImpl) + s3a := newPutTestServer(t, filerAddr) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + r := httptest.NewRequest(http.MethodPut, "/b/o", nil) + _, code, _ := s3a.putToFiler(r, "/buckets/b/o", strings.NewReader(""), "b", "o", 1, 0, nil, false, "") + if code != s3err.ErrServiceUnavailable { + t.Fatalf("putToFiler returned %v, want refusal once a newer write is found", code) + } + if filerImpl.stores != 0 { + t.Fatalf("recovery re-committed the stale entry %d times", filerImpl.stores) + } + if got := filerImpl.entries["/buckets/b/o"]; got == nil || len(got.Chunks) == 0 || got.Chunks[0].FileId != newer.Chunks[0].FileId { + t.Fatalf("the newer write's entry was clobbered: %+v", got) + } +} + +// Chunk lists cannot prove a zero-byte PUT landed — an older empty object at +// the same path compares equal too. A stored entry whose metadata differs from +// this write's must not be claimed as its commit: it belongs to a different +// request, so the outcome stays uncertain for the client to retry. +func TestPutToFilerRoutedAmbiguousEmptyPutStaleMetadata(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{ + volume: volume, + txnNoApply: true, + txnErr: status.Error(codes.Unavailable, "response lost before commit"), + entries: map[string]*filer_pb.Entry{ + "/buckets/b/o": { + Name: "o", + Attributes: &filer_pb.FuseAttributes{Mime: "text/plain"}, + Extended: map[string][]byte{s3_constants.ExtETagKey: []byte("\"d41d8cd98f00b204e9800998ecf8427e\"")}, + }, + }, + } + filerAddr := startFakeFiler(t, filerImpl) + s3a := newPutTestServer(t, filerAddr) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + r := httptest.NewRequest(http.MethodPut, "/b/o", nil) + _, code, _ := s3a.putToFiler(r, "/buckets/b/o", strings.NewReader(""), "b", "o", 1, 0, nil, false, "") + if code != s3err.ErrServiceUnavailable { + t.Fatalf("putToFiler returned %v, want refusal when the stored empty entry cannot be this write's", code) + } + if filerImpl.stores != 0 { + t.Fatalf("recovery committed over an unverifiable empty entry %d times", filerImpl.stores) + } + if got := filerImpl.entries["/buckets/b/o"]; got.GetAttributes().GetMime() != "text/plain" { + t.Fatalf("the older object's entry was replaced: %+v", got) + } +} + +// The counterpart: when the routed zero-byte PUT did commit and only its +// response was lost, the stored entry is this write's own — it carries this +// write's stamped id — and recovery must still claim it rather than fail a +// successful write. +func TestPutToFilerRoutedAmbiguousEmptyPutLanded(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + txnErr: status.Error(codes.Unavailable, "response lost after commit"), + } + filerAddr := startFakeFiler(t, filerImpl) + s3a := newPutTestServer(t, filerAddr) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + r := httptest.NewRequest(http.MethodPut, "/b/o", nil) + _, code, _ := s3a.putToFiler(r, "/buckets/b/o", strings.NewReader(""), "b", "o", 1, 0, nil, false, "") + if code != s3err.ErrNone { + t.Fatalf("putToFiler returned %v, want success — the stored empty entry is this write's own commit", code) + } + if filerImpl.stores != 0 { + t.Fatalf("recovery re-committed an entry that was already stored %d times", filerImpl.stores) + } +} + +// A conditional recovery create is itself ambiguous once sent: if it commits +// and the response is lost, replaying it on another filer can resurrect an +// entry whose chunks a delete already reclaimed. It also serializes with +// routed writes only under the owner's per-path lock, so it is sent to the +// owner alone — never failed over — and a possibly-committed attempt is +// confirmed by lookup instead. +func TestPutToFilerRecoveryCreateNotReplayed(t *testing.T) { + volume := startFakeVolumeServer(t) + // The owner commits the conditional create and loses the response. The + // healthy failover filer must never see it: the committed copy on the + // owner is confirmed instead of re-sent elsewhere. + filerA := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + apply: true, + createErr: status.Error(codes.Unavailable, "response lost after commit"), + txnNoApply: true, + txnErr: status.Error(codes.Unavailable, "response lost"), + } + filerB := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + apply: true, + } + addrA := startFakeFiler(t, filerA) + addrB := startFakeFiler(t, filerB) + s3a := newPutTestServer(t, addrA, addrB) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, addrA) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{addrA}, 1) + + _, code := putTestObject(t, s3a) + if code != s3err.ErrNone { + t.Fatalf("putToFiler returned %v — the committed create should be confirmed, not replayed", code) + } + if filerA.stores != 1 { + t.Fatalf("the owner stored the entry %d times, want once", filerA.stores) + } + if filerB.stores != 0 || len(filerB.conds) != 0 { + t.Fatalf("the recovery create reached a failover filer: stores=%d conds=%d", filerB.stores, len(filerB.conds)) + } + if got := filerA.entries["/buckets/b/o"]; got == nil || len(got.Chunks) != 1 { + t.Fatalf("the committed entry is missing on the owner: %+v", got) + } + if deleted := volume.deleted(); len(deleted) != 0 { + t.Fatalf("chunks under a live entry were deleted: %v", deleted) + } +} + +// The conditional recovery create serializes with routed writes only under the +// owner's per-path lock, so it goes to the owner alone — a failover filer's +// lock is a different serialization boundary. +func TestPutToFilerRoutedRecoveryCreateStaysOnOwner(t *testing.T) { + volume := startFakeVolumeServer(t) + filerA := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + apply: true, + txnNoApply: true, + txnErr: status.Error(codes.Unavailable, "response lost"), + } + filerB := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + apply: true, + } + addrA := startFakeFiler(t, filerA) + addrB := startFakeFiler(t, filerB) + s3a := newPutTestServer(t, addrA, addrB) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, addrA) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{addrA}, 1) + + _, code := putTestObject(t, s3a) + if code != s3err.ErrNone { + t.Fatalf("putToFiler returned %v — the conditional create should commit on the owner", code) + } + if filerA.stores != 1 { + t.Fatalf("the owner stored the entry %d times, want once", filerA.stores) + } + if len(filerA.conds) != 1 || filerA.conds[0].GetClauses()[0].GetKind() != filer_pb.WriteCondition_IF_NOT_EXISTS { + t.Fatalf("recovery create carried %+v, want one IF_NOT_EXISTS", filerA.conds) + } + if filerB.stores != 0 || len(filerB.conds) != 0 { + t.Fatalf("the conditional create failed over to another filer: stores=%d conds=%d", filerB.stores, len(filerB.conds)) + } +} + +// When the owner cannot accept the recovery create, the outcome is uncertain — +// not a failover create under a different filer's lock. +func TestPutToFilerRoutedRecoveryOwnerDownIsUncertain(t *testing.T) { + volume := startFakeVolumeServer(t) + filerA := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + createErr: status.Error(codes.Unavailable, "connection refused"), + txnNoApply: true, + txnErr: status.Error(codes.Unavailable, "response lost"), + } + filerB := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + apply: true, + } + addrA := startFakeFiler(t, filerA) + addrB := startFakeFiler(t, filerB) + s3a := newPutTestServer(t, addrA, addrB) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, addrA) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{addrA}, 1) + + _, code := putTestObject(t, s3a) + if code != s3err.ErrServiceUnavailable { + t.Fatalf("putToFiler returned %v, want an uncertain result when the owner cannot take the create", code) + } + if filerB.stores != 0 || len(filerB.conds) != 0 { + t.Fatalf("the conditional create failed over to another filer: stores=%d conds=%d", filerB.stores, len(filerB.conds)) + } + if deleted := volume.deleted(); len(deleted) == 0 { + t.Fatal("orphaned chunks were not deleted") + } +} + +// An absent entry cannot prove an empty PUT never committed — the routed +// create may have landed and a delete already removed it. Recovery must not +// recreate the object: the outcome is uncertain, not an IF_NOT_EXISTS create. +func TestPutToFilerRoutedAmbiguousEmptyPutAbsentNotRecreated(t *testing.T) { + volume := startFakeVolumeServer(t) + filerImpl := &ambiguousPutFiler{ + volume: volume, + entries: map[string]*filer_pb.Entry{}, + txnErr: status.Error(codes.Unavailable, "response lost after commit"), + } + // The committed empty object is deleted before recovery looks it up. + filerImpl.beforeLookup = func() { + delete(filerImpl.entries, "/buckets/b/o") + } + filerAddr := startFakeFiler(t, filerImpl) + s3a := newPutTestServer(t, filerAddr) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + r := httptest.NewRequest(http.MethodPut, "/b/o", nil) + _, code, _ := s3a.putToFiler(r, "/buckets/b/o", strings.NewReader(""), "b", "o", 1, 0, nil, false, "") + if code != s3err.ErrServiceUnavailable { + t.Fatalf("putToFiler returned %v, want an uncertain result for an unverifiable empty PUT", code) + } + if filerImpl.stores != 0 { + t.Fatalf("recovery recreated an object a delete may have removed %d times", filerImpl.stores) + } + if got := filerImpl.entries["/buckets/b/o"]; got != nil { + t.Fatalf("the deleted object was resurrected: %+v", got) + } +} + +// Matching metadata alone cannot pass an older empty object off as this PUT's +// commit: a foreign write id marks it as a different write, so the outcome +// stays uncertain for the client to retry. +func TestPutToFilerRoutedAmbiguousEmptyPutForeignWriteId(t *testing.T) { + volume := startFakeVolumeServer(t) + emptyMd5 := md5.Sum(nil) + filerImpl := &ambiguousPutFiler{ + volume: volume, + txnNoApply: true, + txnErr: status.Error(codes.Unavailable, "response lost before commit"), + entries: map[string]*filer_pb.Entry{ + "/buckets/b/o": { + Name: "o", + Attributes: &filer_pb.FuseAttributes{Mime: "application/octet-stream", Md5: emptyMd5[:]}, + Extended: map[string][]byte{ + s3_constants.ExtETagKey: []byte("\"d41d8cd98f00b204e9800998ecf8427e\""), + s3_constants.SeaweedFSWriteId: []byte("another-write"), + }, + }, + }, + } + filerAddr := startFakeFiler(t, filerImpl) + s3a := newPutTestServer(t, filerAddr) + s3a.objectWriteLockClient = cluster.NewLockClient(s3a.option.GrpcDialOption, filerAddr) + s3a.objectWriteLockClient.SetRing([]pb.ServerAddress{filerAddr}, 1) + + r := httptest.NewRequest(http.MethodPut, "/b/o", nil) + _, code, _ := s3a.putToFiler(r, "/buckets/b/o", strings.NewReader(""), "b", "o", 1, 0, nil, false, "") + if code != s3err.ErrServiceUnavailable { + t.Fatalf("putToFiler returned %v, want refusal when the stored empty entry carries another write's id", code) + } + if filerImpl.stores != 0 { + t.Fatalf("recovery committed over an unverifiable empty entry %d times", filerImpl.stores) + } + if got := filerImpl.entries["/buckets/b/o"]; string(got.GetExtended()[s3_constants.SeaweedFSWriteId]) != "another-write" { + t.Fatalf("the older object's entry was replaced: %+v", got) + } +} + // A volume-encrypted chunk must be decrypted before SSE decryption sees it: // fetchChunkData feeds ciphered chunks through the cipher-aware read path and // slices plaintext space, while plain chunks keep streaming range reads. diff --git a/weed/s3api/s3api_object_routed_read.go b/weed/s3api/s3api_object_routed_read.go index 1ff8f7423..f26747879 100644 --- a/weed/s3api/s3api_object_routed_read.go +++ b/weed/s3api/s3api_object_routed_read.go @@ -38,7 +38,7 @@ func (s3a *S3ApiServer) getObjectEntryRoutedByKey(bucket, object string) (*filer dir, name := fullPath.DirAndName() var entry *filer_pb.Entry - err := s3a.withFilerClientFailover(context.Background(), preferred, false, func(client filer_pb.SeaweedFilerClient) error { + err := s3a.withFilerClientFailover(context.Background(), preferred, false, true, func(client filer_pb.SeaweedFilerClient) error { resp, lookupErr := filer_pb.LookupEntry(context.Background(), client, &filer_pb.LookupDirectoryEntryRequest{ Directory: dir, Name: name, diff --git a/weed/s3api/s3api_object_routed_write.go b/weed/s3api/s3api_object_routed_write.go index 5e315aa87..27c518842 100644 --- a/weed/s3api/s3api_object_routed_write.go +++ b/weed/s3api/s3api_object_routed_write.go @@ -157,6 +157,10 @@ func singleStrongETag(v string) (string, bool) { // restarted filer's stale ring address (e.g. a rolled K8s pod's old IP) can't hang // the write. The reached filer forwards to the real owner by route_key, keeping its // per-path lock authoritative. Mirrors getObjectEntryRoutedByKey on the read path. +// The transaction is not replayable: it fails over only over a channel that +// never came up, never past a possibly-committed attempt — a lost response is +// ambiguous and re-sending could re-apply mutations over a write that landed in +// between. The caller resolves that ambiguity (e.g. createAfterAmbiguousRoute). func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.ObjectTransactionRequest) (*filer_pb.ObjectTransactionResponse, error) { // One id per logical transaction: failover retries reuse it so the owner // replays the recorded outcome instead of re-applying the mutations. @@ -179,7 +183,7 @@ func (s3a *S3ApiServer) objectTxnOnFiler(owner pb.ServerAddress, req *filer_pb.O if s3a.ownerRecentlyUnreachable(owner) { preferred = "" } - err := s3a.withFilerClientFailover(context.Background(), preferred, false, txn) + err := s3a.withFilerClientFailover(context.Background(), preferred, false, false, txn) return resp, err } @@ -344,7 +348,7 @@ func (s3a *S3ApiServer) routedSelfCopy(ctx context.Context, owner pb.ServerAddre if s3a.filerClient == nil { err = pb.WithFilerClient(false, 0, owner, s3a.option.GrpcDialOption, update) } else { - err = s3a.withFilerClientFailover(ctx, owner, false, update) + err = s3a.withFilerClientFailover(ctx, owner, false, false, update) } if err != nil { return err diff --git a/weed/s3api/s3api_object_routed_write_test.go b/weed/s3api/s3api_object_routed_write_test.go index 9ebaa4686..470a63e0f 100644 --- a/weed/s3api/s3api_object_routed_write_test.go +++ b/weed/s3api/s3api_object_routed_write_test.go @@ -15,7 +15,9 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "github.com/seaweedfs/seaweedfs/weed/wdclient" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" ) func reqWith(headers map[string]string) *http.Request { @@ -421,6 +423,30 @@ func TestObjectTxnFailsOverStaleOwner(t *testing.T) { } } +// Issue 11631: a transaction the owner may have applied must not be re-sent to +// another filer — a lost response is indistinguishable from a commit. Failover +// is only for channels that never came up, which provably sent nothing. +func TestObjectTxnNoReplayAfterSend(t *testing.T) { + owner := &fakeTxnFiler{err: status.Error(codes.Unavailable, "response lost")} + ownerAddr := startFakeFiler(t, owner) + backup := &fakeTxnFiler{} + backupAddr := startFakeFiler(t, backup) + + dialOption := grpc.WithTransportCredentials(insecure.NewCredentials()) + s3a := &S3ApiServer{ + option: &S3ApiServerOption{GrpcDialOption: dialOption}, + filerClient: wdclient.NewFilerClient([]pb.ServerAddress{ownerAddr, backupAddr}, dialOption, ""), + } + req := &filer_pb.ObjectTransactionRequest{LockKey: "/buckets/b/o", RouteKey: "s3.object.write:/buckets/b/o"} + + if _, err := s3a.objectTxnOnFiler(ownerAddr, req); err == nil { + t.Fatal("expected the lost response to surface as an error") + } + if got := atomic.LoadInt32(&backup.calls); got != 0 { + t.Fatalf("backup filer calls = %d — replayed a possibly-committed transaction", got) + } +} + // A completed multipart upload commits in one transaction: the version file's // PUT, the metadata-only removal of .uploads/ (its chunks are the object's // chunks), and the latest-pointer recompute, in that order.