mirrors/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-09-19 14:34:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(s3api/list): cancel ListEntries stream in hasChildren (#9617)

Browse Source 
* fix(s3api/list): cancel ListEntries stream in hasChildren

* fix(s3api): use filer_pb.List in hasChildren

filer_pb.List already wraps the ListEntries stream in a cancellable
context, so the single-entry probe needs no separate helper or manual
context plumbing to avoid the leaked gRPC stream goroutine.

* fix(s3api): propagate request context into hasChildren

Thread r.Context() through listFilerEntries and hasChildren so the
implicit-directory probe cancels when the client disconnects, instead
of running on context.Background().

---------

Co-authored-by: Chris Lu <chris.lu@gmail.com>
This commit is contained in:
Aleksey
2026-05-21 15:48:47 -07:00
committed by GitHub
co-authored by Chris Lu
parent 8fa769f29a
commit 917a87928c
2 changed files with 16 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
weed/s3api/s3api_object_handlers.go
+12 -27
View File
@@ -324,7 +324,7 @@ func removeDuplicateSlashes(object string) string {
// hasChildren("bucket", "empty-dir") where no children exist → false
//
// Performance: ~1-5ms per call (one gRPC LIST request with Limit=1)
func (s3a *S3ApiServer) hasChildren(bucket, prefix string) bool {
func (s3a *S3ApiServer) hasChildren(ctx context.Context, bucket, prefix string) bool {
// Clean up prefix: remove leading slashes
cleanPrefix := strings.TrimPrefix(prefix, "/")
@@ -332,35 +332,20 @@ func (s3a *S3ApiServer) hasChildren(bucket, prefix string) bool {
bucketDir := s3a.bucketDir(bucket)
fullPath := bucketDir + "/" + cleanPrefix
// Try to list one child object in the directory
err := s3a.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
request := &filer_pb.ListEntriesRequest{
Directory: fullPath,
Limit: 1,
InclusiveStartFrom: true,
}
stream, err := client.ListEntries(context.Background(), request)
if err != nil {
return err
}
// Check if we got at least one entry
_, err = stream.Recv()
if err == io.EOF {
return io.EOF // No children
}
if err != nil {
return err
}
// List one child object. filer_pb.List cancels the underlying ListEntries
// stream when it returns, so gRPC's per-stream client goroutine is not leaked.
// The caller's request context is propagated so the probe is cancelled if the
// client disconnects.
found := false
err := filer_pb.List(ctx, s3a, fullPath, "", func(*filer_pb.Entry, bool) error {
found = true
return nil
})
}, "", true, 1)
// If we got an entry (not EOF), then it has children
if err == nil {
return true
return found
}
if errors.Is(err, io.EOF) || errors.Is(err, filer_pb.ErrNotFound) {
if errors.Is(err, filer_pb.ErrNotFound) {
return false
}
glog.V(1).Infof("hasChildren: list entries failed for %s/%s: %v", bucket, cleanPrefix, err)
@@ -2337,7 +2322,7 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
}
if isZeroByteFile {
// Check if it has children (making it an implicit directory)
if s3a.hasChildren(bucket, object) {
if s3a.hasChildren(r.Context(), bucket, object) {
// This is an implicit directory with children
// Return 404 to force clients (like s3fs) to use LIST-based discovery
s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
weed/s3api/s3api_object_handlers_list.go
+4 -4
View File
@@ -109,7 +109,7 @@ func (s3a *S3ApiServer) ListObjectsV2Handler(w http.ResponseWriter, r *http.Requ
// Adjust marker if it ends with delimiter to skip all entries with that prefix
marker = adjustMarkerForDelimiter(marker, delimiter)
response, err := s3a.listFilerEntries(bucket, originalPrefix, maxKeys, marker, delimiter, encodingTypeUrl, fetchOwner)
response, err := s3a.listFilerEntries(r.Context(), bucket, originalPrefix, maxKeys, marker, delimiter, encodingTypeUrl, fetchOwner)
if err != nil {
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
@@ -173,7 +173,7 @@ func (s3a *S3ApiServer) ListObjectsV1Handler(w http.ResponseWriter, r *http.Requ
// Adjust marker if it ends with delimiter to skip all entries with that prefix
marker = adjustMarkerForDelimiter(marker, delimiter)
response, err := s3a.listFilerEntries(bucket, originalPrefix, uint16(maxKeys), marker, delimiter, encodingTypeUrl, true)
response, err := s3a.listFilerEntries(r.Context(), bucket, originalPrefix, uint16(maxKeys), marker, delimiter, encodingTypeUrl, true)
if err != nil {
s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
@@ -232,7 +232,7 @@ func sanitizeV1MarkerEcho(response *ListBucketResult, marker string, encodingTyp
}
}
func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, maxKeys uint16, originalMarker string, delimiter string, encodingTypeUrl bool, fetchOwner bool) (response ListBucketResult, err error) {
func (s3a *S3ApiServer) listFilerEntries(ctx context.Context, bucket string, originalPrefix string, maxKeys uint16, originalMarker string, delimiter string, encodingTypeUrl bool, fetchOwner bool) (response ListBucketResult, err error) {
// convert full path prefix into directory name and prefix for entry name
requestDir, prefix, marker := normalizePrefixMarker(originalPrefix, originalMarker)
bucketPrefix := s3a.bucketPrefix(bucket)
@@ -307,7 +307,7 @@ func (s3a *S3ApiServer) listFilerEntries(bucket string, originalPrefix string, m
if normalizedPrefix != "" {
relativePath := strings.TrimPrefix(fmt.Sprintf("%s/%s", dir, entry.Name), bucketPrefix)
relativePath = strings.TrimPrefix(relativePath, "/")
if normalizedPrefix == relativePath && !s3a.hasChildren(bucket, relativePath) && !entry.IsDirectoryKeyObject() {
if normalizedPrefix == relativePath && !s3a.hasChildren(ctx, bucket, relativePath) && !entry.IsDirectoryKeyObject() {
return
}
}