fix(s3api): handle directory markers with custom Content-Type headers

Fixes #8712: S3 API directory markers created via PutObject with an
explicit Content-Type header (e.g. application/octet-stream) were not
appearing in ListObjects results.

Root cause: The filer strips "application/octet-stream" from the Mime
field during InsertEntry/UpdateEntry (filerstore_wrapper.go). This
causes IsDirectoryKeyObject() to return false, so the directory markers
are excluded from listings.

Additionally, when AWS CLI includes a Content-Type header, it may use
AWS chunked encoding. r.ContentLength then includes chunked overhead,
which can exceed 1024 bytes even for zero-byte payloads, preventing the
directory marker fast-path from triggering.

Changes:
- Use x-amz-decoded-content-length header for accurate payload size
- Require actualContentLength >= 0 to reject unknown/negative lengths
- Wrap body reader with io.LimitReader(1024+1) to prevent OOM
- Reject payloads > 1024 bytes with HTTP 413
- Store MIME in ExtMimeType extended attribute for zero-byte markers
  (persists even when the filer strips the Mime field)
- Update IsDirectoryKeyObject() to check ExtMimeType as fallback

weed/pb/filer_pb/filer_pb_helper.go

@@ -22,7 +22,21 @@ func (entry *Entry) IsInRemoteOnly() bool {
 }
 
 func (entry *Entry) IsDirectoryKeyObject() bool {
-	return entry.IsDirectory && entry.Attributes != nil && entry.Attributes.Mime != ""
+	if !entry.IsDirectory || entry.Attributes == nil {
+		return false
+	}
+	// Check if MIME type is set in attributes
+	if entry.Attributes.Mime != "" {
+		return true
+	}
+	// For directory markers with custom MIME types, check extended attributes
+	// (the filer may not persist the Mime field for directories)
+	if entry.Extended != nil {
+		if _, hasMime := entry.Extended[s3_constants.ExtMimeType]; hasMime {
+			return true
+		}
+	}
+	return false
 }
 
 func (entry *Entry) GetExpiryTime() (expiryTime int64) {

weed/s3api/s3_constants/extend_key.go

@@ -19,6 +19,7 @@ const (
 	ExtLatestVersionOwnerKey       = "Seaweed-X-Amz-Latest-Version-Owner"
 	ExtLatestVersionIsDeleteMarker = "Seaweed-X-Amz-Latest-Version-Is-Delete-Marker"
 	ExtMultipartObjectKey          = "key"
+	ExtMimeType                    = "Seaweed-X-Amz-Mime-Type"
 
 	// Bucket Policy
 	ExtBucketPolicyKey = "Seaweed-X-Amz-Bucket-Policy"

weed/s3api/s3api_object_handlers_put.go

@@ -122,7 +122,18 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 	defer dataReader.Close()
 
 	objectContentType := r.Header.Get("Content-Type")
-	if strings.HasSuffix(object, "/") && r.ContentLength <= 1024 {
+
+	// For aws-chunked requests, r.ContentLength includes the chunked encoding
+	// overhead (signatures, trailers). Use x-amz-decoded-content-length when
+	// present to get the actual payload size for the directory-marker decision.
+	actualContentLength := r.ContentLength
+	if decodedStr := r.Header.Get("X-Amz-Decoded-Content-Length"); decodedStr != "" {
+		if decoded, parseErr := strconv.ParseInt(decodedStr, 10, 64); parseErr == nil && decoded >= 0 {
+			actualContentLength = decoded
+		}
+	}
+
+	if strings.HasSuffix(object, "/") && actualContentLength >= 0 && actualContentLength <= 1024 {
 		// Split the object into directory path and name
 		objectWithoutSlash := strings.TrimSuffix(object, "/")
 		dirName := path.Dir(objectWithoutSlash)
@@ -139,16 +150,28 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 			fullDirPath = fullDirPath + "/" + dirName
 		}
 
-		// Read any content through dataReader (handles chunked encoding properly)
+		// Read any content through dataReader (handles chunked encoding properly).
+		// Use actualContentLength (decoded) to decide whether to read, since
+		// r.ContentLength may include aws-chunked overhead for streaming requests.
+		// Limit reader to prevent OOM from unbounded reads.
 		var dirContent []byte
-		if r.ContentLength != 0 {
+		if actualContentLength != 0 {
 			var readErr error
-			dirContent, readErr = io.ReadAll(dataReader)
+			// Limit the read to 1024+1 bytes to prevent OOM and detect oversized payloads
+			limitedReader := io.LimitReader(dataReader, 1024+1)
+			dirContent, readErr = io.ReadAll(limitedReader)
 			if readErr != nil {
 				glog.Errorf("PutObjectHandler: failed to read directory marker content %s/%s: %v", bucket, object, readErr)
 				s3err.WriteErrorResponse(w, r, s3err.ErrInternalError)
 				return
 			}
+
+			// Reject directory marker payloads larger than 1024 bytes
+			if len(dirContent) > 1024 {
+				glog.Warningf("PutObjectHandler: directory marker payload exceeds 1024 bytes: %s/%s (size=%d)", bucket, object, len(dirContent))
+				s3err.WriteErrorResponse(w, r, s3err.ErrEntityTooLarge)
+				return
+			}
 		}
 
 		// Compute MD5 for ETag (md5.Sum of nil/empty = MD5 of empty content)
@@ -175,6 +198,14 @@ func (s3a *S3ApiServer) PutObjectHandler(w http.ResponseWriter, r *http.Request)
 				}
 				entry.Extended[s3_constants.ExtETagKey] = []byte(dirEtag)
 
+				// Only store MIME in extended attributes for true zero-byte directory
+				// markers (the filer may not persist the Mime field for directories).
+				// Directories with content (e.g., Spark _temporary dirs) should NOT
+				// get this flag, so the empty-folder cleaner can still remove them.
+				if len(dirContent) == 0 {
+					entry.Extended[s3_constants.ExtMimeType] = []byte(objectContentType)
+				}
+
 				// Set object owner for directory objects (same as regular objects)
 				s3a.setObjectOwnerFromRequest(r, bucket, entry)
 			}); err != nil {