s3: carry storage class in the cached listing metadata (#10516)

A listing on a versioned bucket is served from metadata cached on the
.versions directory entry so the whole listing is a single scan. The cache
carried size, mtime, ETag, owner and the delete-marker flag but not the
storage class, so newListEntry found none and fell back to STANDARD.

The result was that HEAD and the listings disagreed about the same object:
HEAD reported the class the object was stored with, while ListObjectsV2 and
ListObjectVersions reported STANDARD for every object. Clients that filter or
tier on storage class act on the listing.

Caches the class alongside the other listing fields, clears it with them, and
copies it in the routed RECOMPUTE_LATEST path so both finalize paths agree.

test/s3/versioning/s3_storage_class_consistency_test.go

@@ -0,0 +1,127 @@
+package s3api
+
+import (
+	"bytes"
+	"context"
+	"testing"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// A listing on a versioned bucket is served from metadata cached on the .versions
+// directory entry so the whole listing is one scan. Anything the cache does not
+// carry falls back to a default, so a field missing from the cache makes the
+// listing disagree with HEAD about the same object. Storage class is one such
+// field, and clients that filter or tier on it act on the listing.
+
+func TestStorageClassConsistentBetweenHeadAndListings(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+	enableVersioning(t, client, bucketName)
+
+	cases := []struct {
+		key   string
+		class types.StorageClass
+		want  string
+	}{
+		{"default.blk", "", "STANDARD"},
+		{"standard.blk", types.StorageClassStandard, "STANDARD"},
+		{"reduced.blk", types.StorageClassReducedRedundancy, "REDUCED_REDUNDANCY"},
+		{"glacier.blk", types.StorageClassGlacier, "GLACIER"},
+	}
+
+	for _, tc := range cases {
+		in := &s3.PutObjectInput{
+			Bucket: aws.String(bucketName),
+			Key:    aws.String(tc.key),
+			Body:   bytes.NewReader([]byte("x")),
+		}
+		if tc.class != "" {
+			in.StorageClass = tc.class
+		}
+		_, err := client.PutObject(context.TODO(), in)
+		require.NoError(t, err, "PUT %s with class %q", tc.key, tc.class)
+	}
+
+	listed, err := client.ListObjectsV2(context.TODO(), &s3.ListObjectsV2Input{Bucket: aws.String(bucketName)})
+	require.NoError(t, err)
+	byKeyV2 := make(map[string]string, len(listed.Contents))
+	for _, o := range listed.Contents {
+		require.NotNil(t, o.Key)
+		byKeyV2[*o.Key] = string(o.StorageClass)
+	}
+
+	versions, err := client.ListObjectVersions(context.TODO(), &s3.ListObjectVersionsInput{Bucket: aws.String(bucketName)})
+	require.NoError(t, err)
+	byKeyVersions := make(map[string]string, len(versions.Versions))
+	for _, v := range versions.Versions {
+		require.NotNil(t, v.Key)
+		if v.IsLatest != nil && *v.IsLatest {
+			byKeyVersions[*v.Key] = string(v.StorageClass)
+		}
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.key, func(t *testing.T) {
+			assert.Equal(t, tc.want, byKeyV2[tc.key],
+				"ListObjectsV2 must report the class the object was stored with")
+			assert.Equal(t, tc.want, byKeyVersions[tc.key],
+				"ListObjectVersions must report the class the object was stored with")
+
+			head, err := client.HeadObject(context.TODO(), &s3.HeadObjectInput{
+				Bucket: aws.String(bucketName), Key: aws.String(tc.key),
+			})
+			require.NoError(t, err)
+			// S3 omits the header for STANDARD and sends it otherwise; either way
+			// it must not contradict the listing.
+			if head.StorageClass != "" {
+				assert.Equal(t, tc.want, string(head.StorageClass),
+					"HEAD and the listings must agree on storage class")
+			} else {
+				assert.Equal(t, "STANDARD", tc.want,
+					"HEAD omits the header only for STANDARD")
+			}
+		})
+	}
+}
+
+// Overwriting refreshes the cached listing metadata; the class must follow the
+// new current version rather than stay pinned to the one it replaced.
+func TestStorageClassFollowsLatestVersion(t *testing.T) {
+	client := getS3Client(t)
+	bucketName := getNewBucketName()
+
+	createBucket(t, client, bucketName)
+	defer deleteBucket(t, client, bucketName)
+	enableVersioning(t, client, bucketName)
+
+	key := "rewritten.blk"
+	_, err := client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(bucketName), Key: aws.String(key),
+		Body:         bytes.NewReader([]byte("first")),
+		StorageClass: types.StorageClassGlacier,
+	})
+	require.NoError(t, err)
+
+	_, err = client.PutObject(context.TODO(), &s3.PutObjectInput{
+		Bucket: aws.String(bucketName), Key: aws.String(key),
+		Body:         bytes.NewReader([]byte("second")),
+		StorageClass: types.StorageClassReducedRedundancy,
+	})
+	require.NoError(t, err)
+
+	listed, err := client.ListObjectsV2(context.TODO(), &s3.ListObjectsV2Input{
+		Bucket: aws.String(bucketName), Prefix: aws.String(key),
+	})
+	require.NoError(t, err)
+	require.Len(t, listed.Contents, 1)
+	assert.Equal(t, "REDUCED_REDUNDANCY", string(listed.Contents[0].StorageClass),
+		"the listing must report the current version's class, not the one it replaced")
+}

weed/s3api/s3_constants/extend_key.go

@@ -13,12 +13,13 @@ const (
 	ExtLatestVersionFileNameKey = "Seaweed-X-Amz-Latest-Version-File-Name"
 	ExtAllowEmptyFolders        = "Seaweed-X-Amz-Allow-Empty-Folders"
 	// Cached list metadata in .versions directory for single-scan efficiency
-	ExtLatestVersionSizeKey        = "Seaweed-X-Amz-Latest-Version-Size"
-	ExtLatestVersionETagKey        = "Seaweed-X-Amz-Latest-Version-ETag"
-	ExtLatestVersionMtimeKey       = "Seaweed-X-Amz-Latest-Version-Mtime"
-	ExtLatestVersionOwnerKey       = "Seaweed-X-Amz-Latest-Version-Owner"
-	ExtLatestVersionIsDeleteMarker = "Seaweed-X-Amz-Latest-Version-Is-Delete-Marker"
-	ExtMultipartObjectKey          = "key"
+	ExtLatestVersionSizeKey         = "Seaweed-X-Amz-Latest-Version-Size"
+	ExtLatestVersionETagKey         = "Seaweed-X-Amz-Latest-Version-ETag"
+	ExtLatestVersionMtimeKey        = "Seaweed-X-Amz-Latest-Version-Mtime"
+	ExtLatestVersionOwnerKey        = "Seaweed-X-Amz-Latest-Version-Owner"
+	ExtLatestVersionIsDeleteMarker  = "Seaweed-X-Amz-Latest-Version-Is-Delete-Marker"
+	ExtLatestVersionStorageClassKey = "Seaweed-X-Amz-Latest-Version-Storage-Class"
+	ExtMultipartObjectKey           = "key"
 	// Wall-clock nanoseconds (int64 as decimal string) captured at the
 	// moment a versioned entry was demoted from current to noncurrent
 	// by a later PUT or delete marker. Read by the s3 lifecycle engine

weed/s3api/s3api_object_versioned_finalize.go

@@ -39,10 +39,11 @@ func (s3a *S3ApiServer) latestPointerRecompute(bucket, object string, useInverte
 		SizeToKey:  s3_constants.ExtLatestVersionSizeKey,
 		MtimeToKey: s3_constants.ExtLatestVersionMtimeKey,
 		CopyExtended: map[string]string{
-			s3_constants.ExtLatestVersionIdKey:          s3_constants.ExtVersionIdKey,
-			s3_constants.ExtLatestVersionETagKey:        s3_constants.ExtETagKey,
-			s3_constants.ExtLatestVersionOwnerKey:       s3_constants.ExtAmzOwnerKey,
-			s3_constants.ExtLatestVersionIsDeleteMarker: s3_constants.ExtDeleteMarkerKey,
+			s3_constants.ExtLatestVersionIdKey:           s3_constants.ExtVersionIdKey,
+			s3_constants.ExtLatestVersionETagKey:         s3_constants.ExtETagKey,
+			s3_constants.ExtLatestVersionOwnerKey:        s3_constants.ExtAmzOwnerKey,
+			s3_constants.ExtLatestVersionIsDeleteMarker:  s3_constants.ExtDeleteMarkerKey,
+			s3_constants.ExtLatestVersionStorageClassKey: s3_constants.AmzStorageClass,
 		},
 		ExcludeName: excludeName,
 	}

weed/s3api/s3api_object_versioning.go

@@ -37,6 +37,7 @@ func clearCachedVersionMetadata(extended map[string][]byte) {
 	delete(extended, s3_constants.ExtLatestVersionETagKey)
 	delete(extended, s3_constants.ExtLatestVersionOwnerKey)
 	delete(extended, s3_constants.ExtLatestVersionIsDeleteMarker)
+	delete(extended, s3_constants.ExtLatestVersionStorageClassKey)
 }
 
 // markVersionNoncurrent stamps ExtNoncurrentSinceNsKey on the named entry
@@ -98,6 +99,9 @@ func setCachedListMetadata(versionsEntry, versionEntry *filer_pb.Entry) {
 		if owner, ok := versionEntry.Extended[s3_constants.ExtAmzOwnerKey]; ok {
 			versionsEntry.Extended[s3_constants.ExtLatestVersionOwnerKey] = owner
 		}
+		if storageClass, ok := versionEntry.Extended[s3_constants.AmzStorageClass]; ok {
+			versionsEntry.Extended[s3_constants.ExtLatestVersionStorageClassKey] = storageClass
+		}
 		if deleteMarker, ok := versionEntry.Extended[s3_constants.ExtDeleteMarkerKey]; ok {
 			versionsEntry.Extended[s3_constants.ExtLatestVersionIsDeleteMarker] = deleteMarker
 		} else {
@@ -2076,6 +2080,13 @@ func (s3a *S3ApiServer) getLatestVersionEntryFromDirectoryEntry(bucket, object s
 				logicalEntry.Extended[s3_constants.ExtAmzOwnerKey] = ownerBytes
 			}
 
+			// Add storage class if cached. Without it the listing falls back to
+			// the default and reports a different class than HEAD does for the
+			// same object.
+			if storageClassBytes, hasStorageClass := versionsDirEntry.Extended[s3_constants.ExtLatestVersionStorageClassKey]; hasStorageClass {
+				logicalEntry.Extended[s3_constants.AmzStorageClass] = storageClassBytes
+			}
+
 			return logicalEntry, nil
 		}
 		glog.Warningf("getLatestVersionEntryFromDirectoryEntry: failed to parse cached metadata for %s/%s, falling back. sizeErr:%v, mtimeErr:%v", bucket, normalizedObject, sizeErr, mtimeErr)