s3api: make ambiguous routed-PUT recovery atomic and stop transaction replay (#11787)

* filer_pb: map PRECONDITION_FAILED to a sentinel error

A conditional CreateEntry that the owner's path lock refuses comes back as
FilerError_PRECONDITION_FAILED; mapping it to ErrPreconditionFailed lets the
S3 gateway tell a rejected precondition apart from a transport failure and
report 412 instead of retrying the write.

* s3api: make the ambiguous-PUT recovery create conditional under the lock

createAfterAmbiguousRoute verified state and then issued an unconditional
create, so a write landing between the lookup and the create was still
clobbered by the stale re-commit (issue #11631). The recovery create now
carries a WriteCondition — IF_NOT_EXISTS for a proven-absent entry,
IF_ENTRY_EQUAL for a directory it may need to write onto — which the owner
filer evaluates under its per-path lock, making the check and the write
atomic. A refused precondition loops back to re-read the state instead of
overwriting whatever won the race.

* s3api: do not fail over a routed transaction past a possibly-sent request

withFilerClientFailover retried the callback on the next filer on any
transport error, so an ObjectTransaction whose response was lost could be
re-applied over a write that committed in between (issue #11631). A gRPC
error cannot distinguish "never sent" from "committed", but channel
readiness can: a non-replayable callback now runs only on a connection that
just reached READY, and failover continues only over channels that never
came up — the one failure mode that provably sent nothing.

* s3api: run ambiguous-route recovery for empty PUTs too

The recovery was gated on a nonempty chunk list, so a zero-byte PUT whose
routed transaction lost its response still fell back to an unconditional
create that could overwrite a newer write and delete its chunks. Chunk-count
mismatch on an empty upload list is enough to distinguish a superseding
entry from this PUT's — there is nothing to save by skipping the check.

* s3api: keep conditional recovery creates off the replay path

The conditional create in the ambiguous-PUT fallback rode the replayable
WithFilerClient failover, so a committed create whose response was lost
could be re-sent to another filer — where a delete in between would let
IF_NOT_EXISTS pass again and resurrect an entry pointing at reclaimed
chunks. Conditional creates now run through the non-replayable failover;
an ambiguous result falls through to confirmCreateLanded, which resolves
the committed entry rather than resending.

* s3api: do not mark a filer unhealthy for an answered request

A reply that reached the callback — including an application-level failure
like a refused write precondition — proves the filer is alive. Checking
received before the health bookkeeping keeps definitive application errors
from tripping the failure circuit or flagging a live routed owner as
unreachable.

* s3api: do not claim an empty stored entry as an ambiguous PUT's own

Two chunkless entries compare equal under sameFileChunks, so an older
empty object satisfies "the route landed" even though it predates this
write — recovery would return success while leaving the request's
Content-Type and metadata unapplied. A landed PUT stores the entry
exactly as built, so require mime/size/md5/content and the extended
attributes to match before claiming it; anything else stays uncertain
and the client retries.

* s3api: prove ambiguous empty PUTs by write id and pin recovery to the owner

- Conditional recovery creates are sent to the routed owner alone instead of failing over: the condition only serializes with routed writes under the owner's per-path lock, and another filer's forwarding is only as correct as its ring view. An unreachable owner now yields an uncertain result rather than a create under a different serialization boundary.
- A proven-absent path no longer satisfies an empty, chunkless PUT: the routed commit may have landed and been deleted, and there are no chunks to verify, so IF_NOT_EXISTS could resurrect a removed object. Recovery returns uncertain for the client to retry.
- Chunkless identity moves from metadata equality to a per-write id (x-seaweedfs-write-id) stamped into Extended: an identical older object can match mime/md5/content but not the id. Entries without this write's id are not claimed, and confirmCreateLanded can now positively confirm a committed empty object instead of always reporting failure.

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
 	}

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)

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.

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)
 		}

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

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.

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,

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

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/<id> (its chunks are the object's
 // chunks), and the latest-pointer recompute, in that order.