s3: answer directory-path probes like AWS so Flink savepoints restore (#10225)

* s3: answer application/x-directory for a directory without a stored mime

A real directory reached via a trailing-slash GET/HEAD answered the
octet-stream default when it had no stored mime, so Hadoop-style S3
filesystems (flink-s3-fs-presto and friends) classified the path as a
0-byte file instead of a directory and then failed reading it as one.
Answer application/x-directory, the marker type those clients probe
for. Stored mimes still echo verbatim, and a file promoted to a
directory keeps octet-stream for its data.

* s3: 404 GET and HEAD on a bare directory path consistently

A directory with no object data of its own answered differently per
path: plain GET gave an empty 200, ranged GET and non-versioned HEAD
gave 404, and on versioned buckets the null-version fallback adopted
the filer directory as a 0-byte object and answered 200. Clients that
probe HEAD-then-GET took the 200s at face value, treated the path as
an empty file, and never fell back to LIST-based directory discovery.

Answer 404 for a bare directory path everywhere, which is what AWS
returns for a prefix. A file promoted to a directory keeps its data
and stays retrievable.

weed/s3api/s3_constants/s3_actions.go

@@ -21,4 +21,5 @@ const (
 	MultipartUploadsFolder          = ".uploads"
 	VersionsFolder                  = ".versions"
 	FolderMimeType                  = "httpd/unix-directory"
+	DirectoryMimeType               = "application/x-directory"
 )

weed/s3api/s3api_object_handlers.go

@@ -362,6 +362,13 @@ func (s3a *S3ApiServer) hasChildren(ctx context.Context, bucket, prefix string)
 	return true
 }
 
+// isBareDirectory reports whether entry is a directory carrying no object data of its own.
+// Such a path is not an S3 object: GET/HEAD answer 404 like AWS does for a prefix, and
+// Hadoop-style clients then discover the directory through their LIST fallback.
+func isBareDirectory(entry *filer_pb.Entry) bool {
+	return entry != nil && entry.IsDirectory && filer.FileSize(entry) == 0
+}
+
 // checkDirectoryObject checks if the object is a directory object (ends with "/") and if it exists
 // Returns: (entry, isDirectoryObject, error)
 // - entry: the directory entry if found and is a directory
@@ -427,10 +434,17 @@ func (s3a *S3ApiServer) serveDirectoryContent(w http.ResponseWriter, r *http.Req
 		return
 	}
 
-	// Set content type - use stored MIME type or default
+	// Set content type - use stored MIME type or default. A directory without a stored
+	// mime and without data of its own answers application/x-directory, the marker type
+	// Hadoop-style clients (e.g. flink-s3-fs-presto) require to classify the path as a
+	// directory; defaulting to octet-stream makes them treat it as a 0-byte file.
 	contentType := entry.Attributes.Mime
 	if contentType == "" {
-		contentType = "application/octet-stream"
+		if entry.IsDirectoryKeyObject() {
+			contentType = "application/octet-stream"
+		} else {
+			contentType = s3_constants.DirectoryMimeType
+		}
 	}
 	w.Header().Set("Content-Type", contentType)
 
@@ -684,7 +698,7 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
 			} else if errors.Is(versionsErr, filer_pb.ErrNotFound) {
 				// .versions/ doesn't exist (confirmed not found), check regular path for null version
 				regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject)
-				if regularErr == nil && regularEntry != nil {
+				if regularErr == nil && regularEntry != nil && !isBareDirectory(regularEntry) {
 					// Found object at regular path - this is the null version
 					entry = regularEntry
 					targetVersionId = "null"
@@ -785,6 +799,13 @@ func (s3a *S3ApiServer) GetObjectHandler(w http.ResponseWriter, r *http.Request)
 		return
 	}
 
+	// A bare directory path is not an object: 404 like AWS instead of an empty 200,
+	// so Hadoop-style clients fall back to LIST discovery. HEAD and ranged GET already 404.
+	if !strings.HasSuffix(object, "/") && isBareDirectory(objectEntryForSSE) {
+		s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
+		return
+	}
+
 	// Re-check bucket policy with object entry for tag-based conditions (e.g., s3:ExistingObjectTag)
 	if errCode := s3a.recheckPolicyWithObjectEntry(r, bucket, object, string(s3_constants.ACTION_READ), objectEntryForSSE.Extended, "GetObjectHandler"); errCode != s3err.ErrNone {
 		s3err.WriteErrorResponse(w, r, errCode)
@@ -2209,7 +2230,7 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
 			} else if errors.Is(versionsErr, filer_pb.ErrNotFound) {
 				// .versions/ doesn't exist (confirmed not found), check regular path for null version
 				regularEntry, regularErr := s3a.getEntry(bucketDir, normalizedObject)
-				if regularErr == nil && regularEntry != nil {
+				if regularErr == nil && regularEntry != nil && !isBareDirectory(regularEntry) {
 					// Found object at regular path - this is the null version
 					entry = regularEntry
 					targetVersionId = "null"
@@ -2348,7 +2369,7 @@ func (s3a *S3ApiServer) HeadObjectHandler(w http.ResponseWriter, r *http.Request
 		if objectEntryForSSE.Attributes != nil {
 			isZeroByteFile := objectEntryForSSE.Attributes.FileSize == 0 && !objectEntryForSSE.IsDirectory
 			// A directory with data (a promoted file) is retrievable; empty directories 404 for LIST fallback.
-			if objectEntryForSSE.IsDirectory && filer.FileSize(objectEntryForSSE) == 0 {
+			if isBareDirectory(objectEntryForSSE) {
 				s3err.WriteErrorResponse(w, r, s3err.ErrNoSuchKey)
 				return
 			}

weed/s3api/s3api_object_handlers_dir_test.go

@@ -0,0 +1,49 @@
+package s3api
+
+import (
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
+)
+
+func TestServeDirectoryContentContentType(t *testing.T) {
+	s3a := &S3ApiServer{}
+
+	tests := []struct {
+		name     string
+		entry    *filer_pb.Entry
+		wantType string
+	}{
+		{
+			name:     "bare directory answers the directory marker type",
+			entry:    &filer_pb.Entry{Name: "savepoint", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{}},
+			wantType: "application/x-directory",
+		},
+		{
+			name:     "promoted file with data keeps octet-stream",
+			entry:    &filer_pb.Entry{Name: "promoted", IsDirectory: true, Content: []byte("data"), Attributes: &filer_pb.FuseAttributes{FileSize: 4}},
+			wantType: "application/octet-stream",
+		},
+		{
+			name:     "stored mime is echoed verbatim",
+			entry:    &filer_pb.Entry{Name: "marker", IsDirectory: true, Attributes: &filer_pb.FuseAttributes{Mime: "httpd/unix-directory"}},
+			wantType: "httpd/unix-directory",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			req := httptest.NewRequest(http.MethodHead, "/bucket/dir/", nil)
+			rec := httptest.NewRecorder()
+			s3a.serveDirectoryContent(rec, req, tt.entry)
+			if rec.Code != http.StatusOK {
+				t.Fatalf("status = %d, want 200", rec.Code)
+			}
+			if got := rec.Header().Get("Content-Type"); got != tt.wantType {
+				t.Fatalf("Content-Type = %q, want %q", got, tt.wantType)
+			}
+		})
+	}
+}