diff --git a/weed/pb/filer_pb/filer_pb_helper.go b/weed/pb/filer_pb/filer_pb_helper.go index 390747919e..fe5ed4fd00 100644 --- a/weed/pb/filer_pb/filer_pb_helper.go +++ b/weed/pb/filer_pb/filer_pb_helper.go @@ -13,6 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/util" "github.com/viant/ptrie" + "google.golang.org/grpc/codes" "google.golang.org/protobuf/proto" ) @@ -210,7 +211,18 @@ func UpdateEntryWithResponse(ctx context.Context, client SeaweedFilerClient, req func LookupEntry(ctx context.Context, client SeaweedFilerClient, request *LookupDirectoryEntryRequest) (*LookupDirectoryEntryResponse, error) { resp, err := client.LookupDirectoryEntry(ctx, request) if err != nil { - if err == ErrNotFound || strings.Contains(err.Error(), ErrNotFound.Error()) { + if err == ErrNotFound { + return nil, ErrNotFound + } + // The filer converts its own miss to codes.NotFound; a raw sentinel + // crossing gRPC arrives as codes.Unknown carrying only its text. The + // same text on any other status is a failed read, not a miss. + if st, ok := util.ServerStatus(err); ok { + if st.Code() == codes.NotFound || + (st.Code() == codes.Unknown && strings.Contains(st.Message(), ErrNotFound.Error())) { + return nil, ErrNotFound + } + } else if strings.Contains(err.Error(), ErrNotFound.Error()) { return nil, ErrNotFound } glog.V(3).InfofCtx(ctx, "read %s/%v: %v", request.Directory, request.Name, err) diff --git a/weed/s3api/s3_error_utils.go b/weed/s3api/s3_error_utils.go index e6eed6b56d..a138a5c716 100644 --- a/weed/s3api/s3_error_utils.go +++ b/weed/s3api/s3_error_utils.go @@ -14,8 +14,10 @@ import ( // isFilerNotFound reports whether a filer error is a not-found. // Unlike lookups (normalized in filer_pb.LookupEntry), list and cache errors // cross gRPC as raw status errors, so the sentinel survives as codes.NotFound -// or only as text. The text is matched last and only on what the filer itself -// said, since callers wrap these errors with a path the client chose. +// or as codes.Unknown carrying only its text. The text is matched only on +// miss-shaped statuses and only on what the filer itself said: the same +// message on another status is a failed read, and callers wrap these errors +// with a path the client chose. func isFilerNotFound(err error) bool { if err == nil { return false @@ -24,7 +26,8 @@ func isFilerNotFound(err error) bool { return true } if st, ok := util.ServerStatus(err); ok { - return st.Code() == codes.NotFound || strings.Contains(st.Message(), filer_pb.ErrNotFound.Error()) + return st.Code() == codes.NotFound || + (st.Code() == codes.Unknown && strings.Contains(st.Message(), filer_pb.ErrNotFound.Error())) } return strings.Contains(err.Error(), filer_pb.ErrNotFound.Error()) } diff --git a/weed/s3api/s3api_object_retention.go b/weed/s3api/s3api_object_retention.go index 53f92f3574..b8dddac576 100644 --- a/weed/s3api/s3api_object_retention.go +++ b/weed/s3api/s3api_object_retention.go @@ -13,6 +13,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants" "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "github.com/seaweedfs/seaweedfs/weed/util" + "google.golang.org/grpc/codes" ) // ==================================================================== @@ -208,12 +209,35 @@ func (s3a *S3ApiServer) getObjectEntry(bucket, object, versionId string) (*filer } if err != nil { - return nil, fmt.Errorf("failed to retrieve object %s/%s: %w", bucket, object, ErrObjectNotFound) + // Only a genuine miss becomes ErrObjectNotFound. Transport and + // internal failures must stay distinguishable: callers like + // LifecycleDelete resolve a miss permanently, so collapsing a + // flaky read into "not found" silently drops the object. + if isObjectEntryMiss(err) { + return nil, fmt.Errorf("failed to retrieve object %s/%s: %w", bucket, object, ErrObjectNotFound) + } + return nil, fmt.Errorf("failed to retrieve object %s/%s: %w", bucket, object, err) } return entry, nil } +// isObjectEntryMiss reports whether an object-entry fetch error is a genuine +// miss (no object, version, or latest version) rather than a transient or +// internal failure the caller may want to retry. +func isObjectEntryMiss(err error) bool { + // Unavailable is a failed read, not a miss - a wrapped not-found + // message in its status text must not classify it as one. + if st, ok := util.ServerStatus(err); ok && st.Code() == codes.Unavailable { + return false + } + return isFilerNotFound(err) || + errors.Is(err, ErrObjectNotFound) || + errors.Is(err, ErrVersionNotFound) || + errors.Is(err, ErrLatestVersionNotFound) || + errors.Is(err, errInvalidVersionID) +} + // getObjectEntryForUpdate resolves the entry a metadata update targets and the // filer path it is written under. It applies the same version resolution as // getSpecificObjectVersion - "null" names the plain pre-versioning entry, not a diff --git a/weed/s3api/s3api_object_retention_test.go b/weed/s3api/s3api_object_retention_test.go index 201d471658..2b19c7c124 100644 --- a/weed/s3api/s3api_object_retention_test.go +++ b/weed/s3api/s3api_object_retention_test.go @@ -2,6 +2,7 @@ package s3api import ( "context" + "errors" "fmt" "io" "net/http" @@ -11,7 +12,9 @@ import ( "time" "google.golang.org/grpc" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/status" "google.golang.org/protobuf/proto" "github.com/seaweedfs/seaweedfs/weed/pb" @@ -871,14 +874,18 @@ func timePtr(t time.Time) *time.Time { type retentionTestFiler struct { filer_pb.UnimplementedSeaweedFilerServer - mu sync.Mutex - entries map[string]*filer_pb.Entry - updates []string + mu sync.Mutex + entries map[string]*filer_pb.Entry + updates []string + lookupErrs map[string]error } func (f *retentionTestFiler) LookupDirectoryEntry(_ context.Context, req *filer_pb.LookupDirectoryEntryRequest) (*filer_pb.LookupDirectoryEntryResponse, error) { f.mu.Lock() defer f.mu.Unlock() + if err, ok := f.lookupErrs[req.Directory+"/"+req.Name]; ok { + return nil, err + } if entry, ok := f.entries[req.Directory+"/"+req.Name]; ok { return &filer_pb.LookupDirectoryEntryResponse{Entry: proto.Clone(entry).(*filer_pb.Entry)}, nil } @@ -903,7 +910,8 @@ func newRetentionTestServer(t *testing.T, f *retentionTestFiler) *S3ApiServer { GrpcDialOption: dialOption, BucketsPath: "/buckets", }, - filerClient: wdclient.NewFilerClient([]pb.ServerAddress{addr}, dialOption, ""), + filerClient: wdclient.NewFilerClient([]pb.ServerAddress{addr}, dialOption, ""), + bucketConfigCache: NewBucketConfigCache(time.Minute), } } @@ -957,3 +965,140 @@ func TestNullVersionLockWritesToPlainEntry(t *testing.T) { t.Fatalf("plain entry legal hold = %q, want ON", got) } } + +// getObjectEntry must only report ErrObjectNotFound for a genuine miss. A +// transient filer error used to be collapsed into ErrObjectNotFound, which +// LifecycleDelete resolves permanently as NOOP_RESOLVED - one flaky read at +// dispatch time would silently skip the object's expiration. +func TestGetObjectEntryPreservesTransientErrors(t *testing.T) { + f := &retentionTestFiler{ + entries: map[string]*filer_pb.Entry{ + "/buckets/b": {Name: "b", IsDirectory: true}, + }, + lookupErrs: map[string]error{ + "/buckets/b/key": status.Error(codes.Unavailable, "filer overloaded"), + }, + } + s3a := newRetentionTestServer(t, f) + + _, err := s3a.getObjectEntry("b", "key", "") + if err == nil { + t.Fatal("expected error for transient lookup failure") + } + if errors.Is(err, ErrObjectNotFound) || errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("transient lookup error must not collapse to not-found: %v", err) + } +} + +// A transport failure whose status text happens to carry the filer not-found +// sentinel (e.g. a proxy re-wrapping a miss as Unavailable) is a failed read, +// not a miss - LookupEntry must not normalize it to ErrNotFound. +func TestGetObjectEntryPreservesUnavailableWithNotFoundMessage(t *testing.T) { + f := &retentionTestFiler{ + entries: map[string]*filer_pb.Entry{ + "/buckets/b": {Name: "b", IsDirectory: true}, + }, + lookupErrs: map[string]error{ + "/buckets/b/key": status.Error(codes.Unavailable, filer_pb.ErrNotFound.Error()), + }, + } + s3a := newRetentionTestServer(t, f) + + _, err := s3a.getObjectEntry("b", "key", "") + if err == nil { + t.Fatal("expected error for unavailable lookup") + } + if errors.Is(err, ErrObjectNotFound) || errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("unavailable lookup must not collapse to not-found: %v", err) + } +} + +// Only miss-shaped statuses may carry the not-found text into ErrObjectNotFound: +// a raw sentinel crossing gRPC arrives as codes.Unknown, while the same text on +// codes.Internal is a failed read. +func TestGetObjectEntryNotFoundTextStatusGate(t *testing.T) { + for _, tc := range []struct { + name string + err error + wantMiss bool + }{ + {"unknown keeps sentinel text", status.Error(codes.Unknown, filer_pb.ErrNotFound.Error()), true}, + {"internal with sentinel text", status.Error(codes.Internal, filer_pb.ErrNotFound.Error()), false}, + {"not-found status", status.Error(codes.NotFound, "entry gone"), true}, + } { + t.Run(tc.name, func(t *testing.T) { + f := &retentionTestFiler{ + entries: map[string]*filer_pb.Entry{ + "/buckets/b": {Name: "b", IsDirectory: true}, + }, + lookupErrs: map[string]error{ + "/buckets/b/key": tc.err, + }, + } + s3a := newRetentionTestServer(t, f) + + _, err := s3a.getObjectEntry("b", "key", "") + if err == nil { + t.Fatal("expected error") + } + if got := errors.Is(err, ErrObjectNotFound); got != tc.wantMiss { + t.Fatalf("isObjectEntryMiss = %v, want %v: %v", got, tc.wantMiss, err) + } + }) + } +} + +// A pre-versioning object under a now-versioned bucket has no .versions +// directory; when the plain-object fallback read fails transiently, that +// failure must surface instead of the directory's not-found error. +func TestGetObjectEntryPreservesTransientFallbackError(t *testing.T) { + f := &retentionTestFiler{ + entries: map[string]*filer_pb.Entry{ + "/buckets/b": { + Name: "b", + IsDirectory: true, + Extended: map[string][]byte{s3_constants.ExtVersioningKey: []byte(s3_constants.VersioningEnabled)}, + }, + }, + lookupErrs: map[string]error{ + "/buckets/b/key": status.Error(codes.Unavailable, "filer overloaded"), + }, + } + s3a := newRetentionTestServer(t, f) + + _, err := s3a.getObjectEntry("b", "key", "") + if err == nil { + t.Fatal("expected error for transient fallback lookup failure") + } + if errors.Is(err, ErrObjectNotFound) || errors.Is(err, filer_pb.ErrNotFound) { + t.Fatalf("transient fallback error must not collapse to not-found: %v", err) + } +} + +func TestGetObjectEntryInvalidVersionIdStillNotFound(t *testing.T) { + f := &retentionTestFiler{ + entries: map[string]*filer_pb.Entry{ + "/buckets/b": {Name: "b", IsDirectory: true}, + }, + } + s3a := newRetentionTestServer(t, f) + + _, err := s3a.getObjectEntry("b", "key", "not/../a-version") + if !errors.Is(err, ErrObjectNotFound) { + t.Fatalf("invalid version id must report ErrObjectNotFound, got: %v", err) + } +} + +func TestGetObjectEntryMissStillNotFound(t *testing.T) { + f := &retentionTestFiler{ + entries: map[string]*filer_pb.Entry{ + "/buckets/b": {Name: "b", IsDirectory: true}, + }, + } + s3a := newRetentionTestServer(t, f) + + _, err := s3a.getObjectEntry("b", "missing", "") + if !errors.Is(err, ErrObjectNotFound) { + t.Fatalf("missing object must report ErrObjectNotFound, got: %v", err) + } +} diff --git a/weed/s3api/s3api_object_versioning.go b/weed/s3api/s3api_object_versioning.go index 8b9c4209cf..2cb5dba999 100644 --- a/weed/s3api/s3api_object_versioning.go +++ b/weed/s3api/s3api_object_versioning.go @@ -1993,6 +1993,11 @@ func (s3a *S3ApiServer) doGetLatestObjectVersion(bucket, object string, maxRetri }, maxRetries) if err != nil { + // Only a genuine miss means "no .versions directory" and earns the + // pre-versioning fallback; a failed lookup says nothing about the dir. + if !isObjectEntryMiss(err) { + return nil, fmt.Errorf("failed to get %s/%s .versions directory: %w", bucket, normalizedObject, err) + } // .versions directory doesn't exist - this can happen for objects that existed // before versioning was enabled on the bucket. Fall back to checking for a // regular (non-versioned) object file. @@ -2000,6 +2005,9 @@ func (s3a *S3ApiServer) doGetLatestObjectVersion(bucket, object string, maxRetri regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject) if regularErr != nil { + if !isObjectEntryMiss(regularErr) { + return nil, fmt.Errorf("failed to get regular object %s/%s: %w", bucket, normalizedObject, regularErr) + } glog.V(1).Infof("getLatestObjectVersion: no pre-versioning object found for %s/%s (error: %v)", bucket, normalizedObject, regularErr) return nil, fmt.Errorf("failed to get %s/%s .versions directory and no regular object found: %w", bucket, normalizedObject, err) }