diff --git a/weed/mount/weedfs_dir_mkrm.go b/weed/mount/weedfs_dir_mkrm.go index 43aa2cbb5..755fbf444 100644 --- a/weed/mount/weedfs_dir_mkrm.go +++ b/weed/mount/weedfs_dir_mkrm.go @@ -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 } diff --git a/weed/mount/weedfs_stream_mutate.go b/weed/mount/weedfs_stream_mutate.go index b4b3298fa..79406c402 100644 --- a/weed/mount/weedfs_stream_mutate.go +++ b/weed/mount/weedfs_stream_mutate.go @@ -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), diff --git a/weed/mount/weedfs_stream_mutate_error_test.go b/weed/mount/weedfs_stream_mutate_error_test.go new file mode 100644 index 000000000..2db4f7371 --- /dev/null +++ b/weed/mount/weedfs_stream_mutate_error_test.go @@ -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()) + } +}