fix(mount): make a concurrent duplicate mkdir fail with EEXIST instead of both succeeding (#11079)

* fix(mount): make Mkdir exclusive so a concurrent duplicate fails with EEXIST Mkdir sent CreateEntryRequest without OExcl, so the filer treated a concurrent duplicate as an update and reported success to both callers; the kernel's pre-mkdir lookup only masks this when the winner's create is already visible. Set OExcl, map the entry-already-exists sentinel to EEXIST instead of EIO, and drop the parent's children cache on the losing side so the next lookup fetches the winner's entry. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(mount): route exclusive creates to the path's owner filer The filer's per-path lock is filer-local and the store insert has upsert semantics, so two mounts streaming to different filers can both create the same path even with OExcl (measured 18/30 both-success on a 3-filer cluster). Hash the path over the sorted filer list so every mount sends the same path's exclusive create to the same owner filer: keep the mutation stream when it already targets the owner, fall back to it when the owner is unreachable. Also let doUnary hand failed creates to CreateEntry so the structured error code survives as EEXIST instead of collapsing into the stream's generic EIO. Same race after the change: 30/30 exactly one winner, every loser fails with EEXIST. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(mount): review fixes — pin exclusive creates to the owner filer An OExcl create now goes only to the path's owner filer: retrying on a different filer would race the owner's possibly still-in-flight create through a separate per-path lock, the very hole this routing closes. A broken mutation stream retries the same owner over unary, and an unreachable owner fails the create instead of degrading. Pick the owner by rendezvous hashing so the choice is independent of the configured filer order, and mounts configured with different but overlapping lists still agree wherever the winning filer appears in both. Reject a stream create wrapper whose nested response is nil instead of handing it to CreateEntry, which would dereference it. Add ownerFilerAddress unit tests: order independence, subset agreement, spread. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * mount: drop the client-side owner ring, the filer routes exclusive creates now Exclusive creates are arbitrated cluster-wide on the server: the filer resolves an OExcl create's ring owner and forwards one hop, so one filer's per-path lock binds every creator — mount, S3, the HTTP surface and the Java client alike, not only the ones that opted into a client-side ring. That makes exclusiveCreateEntry redundant. It hashed the mount's configured -filer list, which names a different owner than the master-maintained ring, and failed the mkdir outright when its chosen owner was unreachable rather than letting the ring reassign. Mkdir goes back to streamCreateEntry; OExcl and the EEXIST mapping stay, and now mean what they say. Claude-Session: https://claude.ai/code/session_01Fx1Hx8RqsJqHpbfbgTf4WJ * mount: cover the create error plumbing that turns a lost race into EEXIST Letting a failed create's structured code survive doUnary is what makes a lost mkdir race report EEXIST instead of EIO, and it had no test. Pull the two steps out so they can be exercised without a live stream: hasCreateResponse decides whether a response still carries a code to unwrap, createEntryFromResponse does the unwrapping. Reading the guard the other way round also says what it means — consume the response only when there is no nested code left to recover — rather than negating a type assertion inline. Claude-Session: https://claude.ai/code/session_01Fx1Hx8RqsJqHpbfbgTf4WJ * mount: do not trust a create reply's shape before reading it createEntryFromResponse read cr.ErrorCode without checking the nested response was there. Nothing our filer sends is shaped that way, but the mount reads this off the wire and a nil there panics the whole mount, so report it instead. A top-level failure whose nested response carries no code was also returned as success, silently losing the error. Fall back to the top-level errno when the nested response explains nothing. Claude-Session: https://claude.ai/code/session_01Fx1Hx8RqsJqHpbfbgTf4WJ --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: Chris Lu <chris.lu@gmail.com>
2026-09-08 15:41:15 +02:00 · 2026-09-02 21:26:04 -07:00
parent 9e34426a56
commit 7465b6a80f
3 changed files with 181 additions and 2 deletions
@@ -2,6 +2,7 @@ package mount

 import (
 	"context"
+	"errors"
 	"os"
 	"syscall"
 	"time"
@@ -75,6 +76,14 @@ func (wfs *WFS) Mkdir(cancel <-chan struct{}, in *fuse.MkdirIn, name string, out
 		Entry:                    newEntry,
 		Signatures:               []int32{wfs.signature},
 		SkipCheckParentDirectory: true,
+		// mkdir(2) is exclusive by contract: creating an existing path must
+		// fail with EEXIST. Without OExcl the filer treats a concurrent
+		// duplicate as an update and reports success to both callers; the
+		// kernel's pre-mkdir lookup only catches the duplicate when the
+		// winner's create is already visible, which a cross-node race defeats.
+		// The filer routes an OExcl create to the entry's ring owner, so its
+		// per-path lock arbitrates every creator in the cluster.
+		OExcl: true,
 	}

 	glog.V(1).Infof("mkdir: %v", request)
@@ -99,6 +108,14 @@ func (wfs *WFS) Mkdir(cancel <-chan struct{}, in *fuse.MkdirIn, name string, out

 	if err != nil {
 		wfs.mapPbIdFromFilerToLocal(newEntry)
+		if errors.Is(err, filer_pb.ErrEntryAlreadyExists) {
+			// Lost a create race: the path exists on the filer but not yet in
+			// this mount's cache. Drop the parent's children cache so the next
+			// lookup fetches the winner's entry instead of waiting for the
+			// metadata subscription to deliver it.
+			wfs.inodeToPath.InvalidateChildrenCache(dirFullPath)
+			return fuse.Status(syscall.EEXIST)
+		}
 		return fuse.EIO
 	}

@@ -29,6 +29,14 @@ type streamMutateError struct {
 func (e *streamMutateError) Error() string        { return e.msg }
 func (e *streamMutateError) Errno() syscall.Errno { return e.errno }

+// hasCreateResponse reports whether resp carries a nested CreateEntryResponse to
+// unwrap. A create wrapper without one has no structured code to recover, and
+// CreateEntry would dereference the nil.
+func hasCreateResponse(resp *filer_pb.StreamMutateEntryResponse) bool {
+	cr, ok := resp.Response.(*filer_pb.StreamMutateEntryResponse_CreateResponse)
+	return ok && cr.CreateResponse != nil
+}
+
 // ErrStreamTransport is a sentinel error type for transport-level stream
 // failures (disconnects, send errors). Callers use errors.Is to decide
 // whether to fall back to unary RPCs.
@@ -82,12 +90,21 @@ func (m *streamMutateMux) CreateEntry(ctx context.Context, req *filer_pb.CreateE
 	if err != nil {
 		return nil, err
 	}
+	return createEntryFromResponse(resp, req)
+}
+
+// createEntryFromResponse unwraps a create's nested response, mapping its
+// structured code back to the sentinel callers match on (same logic as
+// CreateEntryWithResponse).
+func createEntryFromResponse(resp *filer_pb.StreamMutateEntryResponse, req *filer_pb.CreateEntryRequest) (*filer_pb.CreateEntryResponse, error) {
 	r, ok := resp.Response.(*filer_pb.StreamMutateEntryResponse_CreateResponse)
 	if !ok {
 		return nil, fmt.Errorf("unexpected response type %T", resp.Response)
 	}
-	// Check nested error fields (same logic as CreateEntryWithResponse).
 	cr := r.CreateResponse
+	if cr == nil {
+		return nil, &streamMutateError{msg: "create response missing", errno: syscall.EIO}
+	}
 	if cr.ErrorCode != filer_pb.FilerError_OK {
 		if sentinel := filer_pb.FilerErrorToSentinel(cr.ErrorCode); sentinel != nil {
 			return nil, fmt.Errorf("CreateEntry %s/%s: %w", req.Directory, req.Entry.Name, sentinel)
@@ -97,6 +114,11 @@ func (m *streamMutateMux) CreateEntry(ctx context.Context, req *filer_pb.CreateE
 	if cr.Error != "" {
 		return nil, &streamMutateError{msg: cr.Error, errno: syscall.EIO}
 	}
+	if resp.Error != "" {
+		// The nested response explained nothing, so the top-level failure is all
+		// there is: reporting success here would lose the create's error entirely.
+		return nil, &streamMutateError{msg: resp.Error, errno: syscall.Errno(resp.Errno)}
+	}
 	return cr, nil
 }

@@ -235,7 +257,11 @@ func (m *streamMutateMux) doUnary(ctx context.Context, req *filer_pb.StreamMutat
 		if !ok {
 			return nil, fmt.Errorf("%w: stream closed", ErrStreamTransport)
 		}
-		if resp.Error != "" {
+		// A failed create still carries a structured error code in its nested
+		// CreateEntryResponse, so hand that back for CreateEntry to unwrap and
+		// the sentinel (entry-already-exists → EEXIST) survives instead of
+		// collapsing into the top-level generic errno.
+		if resp.Error != "" && !hasCreateResponse(resp) {
 			return nil, &streamMutateError{
 				msg:   resp.Error,
 				errno: syscall.Errno(resp.Errno),
@@ -0,0 +1,136 @@
+package mount
+
+import (
+	"errors"
+	"syscall"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+// The filer reports a failed create twice: a generic errno at the top level and
+// a structured code in the nested CreateEntryResponse. doUnary must not consume
+// the response on the generic one, or the sentinel never reaches Mkdir and a
+// lost create race surfaces as EIO instead of EEXIST.
+func TestHasCreateResponse(t *testing.T) {
+	cases := []struct {
+		name string
+		resp *filer_pb.StreamMutateEntryResponse
+		want bool
+	}{
+		{
+			name: "create carrying a nested response",
+			resp: &filer_pb.StreamMutateEntryResponse{
+				Response: &filer_pb.StreamMutateEntryResponse_CreateResponse{
+					CreateResponse: &filer_pb.CreateEntryResponse{
+						Error:     "entry already exists",
+						ErrorCode: filer_pb.FilerError_ENTRY_ALREADY_EXISTS,
+					},
+				},
+			},
+			want: true,
+		},
+		{
+			name: "create wrapper with no nested response",
+			resp: &filer_pb.StreamMutateEntryResponse{
+				Response: &filer_pb.StreamMutateEntryResponse_CreateResponse{},
+			},
+			want: false,
+		},
+		{
+			name: "a different mutation",
+			resp: &filer_pb.StreamMutateEntryResponse{
+				Response: &filer_pb.StreamMutateEntryResponse_DeleteResponse{
+					DeleteResponse: &filer_pb.DeleteEntryResponse{Error: "boom"},
+				},
+			},
+			want: false,
+		},
+		{
+			name: "no response at all",
+			resp: &filer_pb.StreamMutateEntryResponse{},
+			want: false,
+		},
+	}
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			if got := hasCreateResponse(tc.resp); got != tc.want {
+				t.Fatalf("hasCreateResponse = %v, want %v", got, tc.want)
+			}
+		})
+	}
+}
+
+// CreateEntry turns the nested code into the sentinel Mkdir matches on.
+func TestStreamMutateCreateEntryUnwrapsAlreadyExists(t *testing.T) {
+	resp := &filer_pb.StreamMutateEntryResponse{
+		Response: &filer_pb.StreamMutateEntryResponse_CreateResponse{
+			CreateResponse: &filer_pb.CreateEntryResponse{
+				Error:     "/dir/name: entry already exists",
+				ErrorCode: filer_pb.FilerError_ENTRY_ALREADY_EXISTS,
+			},
+		},
+	}
+	_, err := createEntryFromResponse(resp, &filer_pb.CreateEntryRequest{
+		Directory: "/dir",
+		Entry:     &filer_pb.Entry{Name: "name"},
+	})
+	if !errors.Is(err, filer_pb.ErrEntryAlreadyExists) {
+		t.Fatalf("err = %v, want it to wrap ErrEntryAlreadyExists", err)
+	}
+}
+
+// Any other create failure keeps the generic errno rather than inventing one.
+func TestStreamMutateCreateEntryKeepsGenericFailure(t *testing.T) {
+	resp := &filer_pb.StreamMutateEntryResponse{
+		Response: &filer_pb.StreamMutateEntryResponse_CreateResponse{
+			CreateResponse: &filer_pb.CreateEntryResponse{Error: "store unavailable"},
+		},
+	}
+	_, err := createEntryFromResponse(resp, &filer_pb.CreateEntryRequest{
+		Directory: "/dir",
+		Entry:     &filer_pb.Entry{Name: "name"},
+	})
+	var sme *streamMutateError
+	if !errors.As(err, &sme) || sme.Errno() != syscall.EIO {
+		t.Fatalf("err = %v, want a streamMutateError with EIO", err)
+	}
+}
+
+// A create wrapper whose nested response is missing must be reported, not
+// dereferenced. Nothing our filer sends looks like this, but the mount is
+// reading off the wire and a panic here takes the whole mount down.
+func TestStreamMutateCreateEntryRejectsMissingNestedResponse(t *testing.T) {
+	_, err := createEntryFromResponse(&filer_pb.StreamMutateEntryResponse{
+		Response: &filer_pb.StreamMutateEntryResponse_CreateResponse{},
+	}, &filer_pb.CreateEntryRequest{
+		Directory: "/dir",
+		Entry:     &filer_pb.Entry{Name: "name"},
+	})
+	var sme *streamMutateError
+	if !errors.As(err, &sme) || sme.Errno() != syscall.EIO {
+		t.Fatalf("err = %v, want a streamMutateError with EIO", err)
+	}
+}
+
+// A top-level failure the nested response does not explain must not read as
+// success just because the nested fields happen to be empty.
+func TestStreamMutateCreateEntryKeepsUnexplainedTopLevelError(t *testing.T) {
+	_, err := createEntryFromResponse(&filer_pb.StreamMutateEntryResponse{
+		Error: "filer went away mid-create",
+		Errno: int32(syscall.EAGAIN),
+		Response: &filer_pb.StreamMutateEntryResponse_CreateResponse{
+			CreateResponse: &filer_pb.CreateEntryResponse{},
+		},
+	}, &filer_pb.CreateEntryRequest{
+		Directory: "/dir",
+		Entry:     &filer_pb.Entry{Name: "name"},
+	})
+	var sme *streamMutateError
+	if !errors.As(err, &sme) {
+		t.Fatalf("err = %v, want a streamMutateError", err)
+	}
+	if sme.Errno() != syscall.EAGAIN {
+		t.Fatalf("errno = %v, want the top-level EAGAIN", sme.Errno())
+	}
+}