Merge pull request #1471 from versity/fix/AzureNextMarker

fix: update marker/continuation token to be the azure next marker

backend/azure/azure.go

@@ -68,6 +68,8 @@ const (
 	onameAttr              key = "Objname"
 	onameAttrLower         key = "objname"
 	metaTmpMultipartPrefix key = ".sgwtmp" + "/multipart"
+
+	defaultListingMaxKeys = 1000
 )
 
 func (key) Table() map[string]struct{} {
@@ -579,26 +581,6 @@ func (az *Azure) GetObjectAttributes(ctx context.Context, input *s3.GetObjectAtt
 }
 
 func (az *Azure) ListObjects(ctx context.Context, input *s3.ListObjectsInput) (s3response.ListObjectsResult, error) {
-	client, err := az.getContainerClient(*input.Bucket)
-	if err != nil {
-		return s3response.ListObjectsResult{}, nil
-	}
-	pager := client.NewListBlobsHierarchyPager(*input.Delimiter, &container.ListBlobsHierarchyOptions{
-		Marker:     input.Marker,
-		MaxResults: input.MaxKeys,
-		Prefix:     input.Prefix,
-	})
-
-	var objects []s3response.Object
-	var cPrefixes []types.CommonPrefix
-	var nextMarker *string
-	var isTruncated bool
-	var maxKeys int32 = math.MaxInt32
-
-	if input.MaxKeys != nil {
-		maxKeys = *input.MaxKeys
-	}
-
 	// Retrieve the bucket acl to get the bucket owner
 	// All the objects in the bucket are owner by the bucket owner
 	aclBytes, err := az.getContainerMetaData(ctx, *input.Bucket, string(keyAclCapital))
@@ -611,19 +593,49 @@ func (az *Azure) ListObjects(ctx context.Context, input *s3.ListObjectsInput) (s
 		return s3response.ListObjectsResult{}, err
 	}
 
-Pager:
-	for pager.More() {
+	client, err := az.getContainerClient(*input.Bucket)
+	if err != nil {
+		return s3response.ListObjectsResult{}, nil
+	}
+
+	var maxKeys int32 = defaultListingMaxKeys
+	if input.MaxKeys != nil {
+		maxKeys = *input.MaxKeys
+	}
+
+	pager := client.NewListBlobsHierarchyPager(*input.Delimiter, &container.ListBlobsHierarchyOptions{
+		MaxResults: &maxKeys,
+		Prefix:     input.Prefix,
+	})
+
+	var objects []s3response.Object
+	var cPrefixes []types.CommonPrefix
+	var nextMarker *string
+	var isTruncated bool
+
+	// Convert marker to filter criteria
+	var markerFilter string
+	if input.Marker != nil && *input.Marker != "" {
+		markerFilter = *input.Marker
+	}
+
+	// Loop through pages until we have enough objects or no more pages
+	objectsFound := int32(0)
+	for pager.More() && objectsFound < maxKeys {
 		resp, err := pager.NextPage(ctx)
 		if err != nil {
 			return s3response.ListObjectsResult{}, azureErrToS3Err(err)
 		}
+
+		// Process objects from this page
+		var pageObjects []s3response.Object
 		for _, v := range resp.Segment.BlobItems {
-			if len(objects)+len(cPrefixes) >= int(maxKeys) {
-				nextMarker = objects[len(objects)-1].Key
-				isTruncated = true
-				break Pager
+			// Skip objects that come before or equal to marker
+			if markerFilter != "" && *v.Name <= markerFilter {
+				continue
 			}
-			objects = append(objects, s3response.Object{
+
+			pageObjects = append(pageObjects, s3response.Object{
 				ETag:         backend.GetPtrFromString(convertAzureEtag(v.Properties.ETag)),
 				Key:          v.Name,
 				LastModified: v.Properties.LastModified,
@@ -633,20 +645,22 @@ Pager:
 					ID: &acl.Owner,
 				},
 			})
-		}
-		for _, v := range resp.Segment.BlobPrefixes {
-			if *v.Name <= *input.Marker {
-				continue
-			}
-			if len(objects)+len(cPrefixes) >= int(maxKeys) {
-				nextMarker = cPrefixes[len(cPrefixes)-1].Prefix
-				isTruncated = true
-				break Pager
-			}
 
-			marker := getString(input.Marker)
-			pfx := strings.TrimSuffix(*v.Name, getString(input.Delimiter))
-			if marker != "" && strings.HasPrefix(marker, pfx) {
+			objectsFound++
+			if objectsFound >= maxKeys {
+				// Set next marker to the current object name for pagination
+				nextMarker = v.Name
+				isTruncated = true
+				break
+			}
+		}
+
+		objects = append(objects, pageObjects...)
+
+		// Process common prefixes from this page
+		for _, v := range resp.Segment.BlobPrefixes {
+			// Skip prefixes that come before or equal to marker
+			if markerFilter != "" && *v.Name <= markerFilter {
 				continue
 			}
 
@@ -654,6 +668,16 @@ Pager:
 				Prefix: v.Name,
 			})
 		}
+
+		// If we've reached maxKeys, break
+		if objectsFound >= maxKeys {
+			break
+		}
+
+		// If Azure indicates more pages but we need to continue for more objects
+		if resp.NextMarker != nil && *resp.NextMarker != "" && objectsFound < maxKeys {
+			continue
+		}
 	}
 
 	return s3response.ListObjectsResult{
@@ -670,98 +694,104 @@ Pager:
 }
 
 func (az *Azure) ListObjectsV2(ctx context.Context, input *s3.ListObjectsV2Input) (s3response.ListObjectsV2Result, error) {
-	marker := ""
-	if *input.ContinuationToken > *input.StartAfter {
-		marker = *input.ContinuationToken
-	} else {
-		marker = *input.StartAfter
+	// Retrieve the bucket acl to get the bucket owner
+	// All the objects in the bucket are owner by the bucket owner
+	aclBytes, err := az.getContainerMetaData(ctx, *input.Bucket, string(keyAclCapital))
+	if err != nil {
+		return s3response.ListObjectsV2Result{}, azureErrToS3Err(err)
 	}
+
+	acl, err := auth.ParseACL(aclBytes)
+	if err != nil {
+		return s3response.ListObjectsV2Result{}, err
+	}
+
 	client, err := az.getContainerClient(*input.Bucket)
 	if err != nil {
 		return s3response.ListObjectsV2Result{}, nil
 	}
+
+	var maxKeys int32 = defaultListingMaxKeys
+	if input.MaxKeys != nil {
+		maxKeys = *input.MaxKeys
+	}
+
 	pager := client.NewListBlobsHierarchyPager(*input.Delimiter, &container.ListBlobsHierarchyOptions{
-		Marker:     &marker,
-		MaxResults: input.MaxKeys,
+		Marker:     input.ContinuationToken,
+		MaxResults: &maxKeys,
 		Prefix:     input.Prefix,
 	})
 
 	var objects []s3response.Object
-	var cPrefixes []types.CommonPrefix
-	var nextMarker *string
-	var isTruncated bool
-	var maxKeys int32 = math.MaxInt32
-	var fetchOwner bool
+	var resp container.ListBlobsHierarchyResponse
 
-	if input.MaxKeys != nil {
-		maxKeys = *input.MaxKeys
-	}
-	if input.FetchOwner != nil {
-		fetchOwner = *input.FetchOwner
-	}
-
-	// Retrieve the bucket acl to get the bucket owner, if "fetchOwner" is true
-	// All the objects in the bucket are owner by the bucket owner
-	var acl auth.ACL
-	if fetchOwner {
-		aclBytes, err := az.getContainerMetaData(ctx, *input.Bucket, string(keyAclCapital))
+	// Loop through pages until we find objects or no more pages
+	for {
+		resp, err = pager.NextPage(ctx)
 		if err != nil {
 			return s3response.ListObjectsV2Result{}, azureErrToS3Err(err)
 		}
 
-		acl, err = auth.ParseACL(aclBytes)
-		if err != nil {
-			return s3response.ListObjectsV2Result{}, err
-		}
-	}
-
-Pager:
-	for pager.More() {
-		resp, err := pager.NextPage(ctx)
-		if err != nil {
-			return s3response.ListObjectsV2Result{}, azureErrToS3Err(err)
-		}
+		// Convert Azure objects to S3 objects
+		var pageObjects []s3response.Object
 		for _, v := range resp.Segment.BlobItems {
-			if len(objects)+len(cPrefixes) >= int(maxKeys) {
-				nextMarker = objects[len(objects)-1].Key
-				isTruncated = true
-				break Pager
-			}
-
-			obj := s3response.Object{
+			pageObjects = append(pageObjects, s3response.Object{
 				ETag:         backend.GetPtrFromString(convertAzureEtag(v.Properties.ETag)),
 				Key:          v.Name,
 				LastModified: v.Properties.LastModified,
 				Size:         v.Properties.ContentLength,
 				StorageClass: types.ObjectStorageClassStandard,
-			}
-			if fetchOwner {
-				obj.Owner = &types.Owner{
+				Owner: &types.Owner{
 					ID: &acl.Owner,
-				}
-			}
-			objects = append(objects, obj)
-		}
-		for _, v := range resp.Segment.BlobPrefixes {
-			if *v.Name <= marker {
-				continue
-			}
-			if len(objects)+len(cPrefixes) >= int(maxKeys) {
-				nextMarker = cPrefixes[len(cPrefixes)-1].Prefix
-				isTruncated = true
-				break Pager
-			}
-
-			marker := getString(input.ContinuationToken)
-			pfx := strings.TrimSuffix(*v.Name, getString(input.Delimiter))
-			if marker != "" && strings.HasPrefix(marker, pfx) {
-				continue
-			}
-
-			cPrefixes = append(cPrefixes, types.CommonPrefix{
-				Prefix: v.Name,
+				},
 			})
 		}
+
+		// If StartAfter is specified, filter objects
+		if input.StartAfter != nil && *input.StartAfter != "" {
+			startAfter := *input.StartAfter
+			startIndex := -1
+			for i, obj := range pageObjects {
+				if *obj.Key > startAfter {
+					startIndex = i
+					break
+				}
+			}
+
+			if startIndex != -1 {
+				// Found objects after StartAfter in this page
+				objects = append(objects, pageObjects[startIndex:]...)
+				break
+			} else {
+				// No objects after StartAfter in this page
+				// Check if there are more pages to examine
+				if resp.NextMarker == nil || *resp.NextMarker == "" {
+					// No more pages, so no objects after StartAfter
+					break
+				}
+				// Continue to next page without adding any objects
+				continue
+			}
+		} else {
+			// No StartAfter specified, add all objects from this page
+			objects = append(objects, pageObjects...)
+			break
+		}
+	}
+
+	var cPrefixes []types.CommonPrefix
+	for _, v := range resp.Segment.BlobPrefixes {
+		cPrefixes = append(cPrefixes, types.CommonPrefix{
+			Prefix: v.Name,
+		})
+	}
+
+	var isTruncated bool
+	var nextMarker *string
+	// If Azure returned a NextMarker, set it for the next request
+	if resp.NextMarker != nil && *resp.NextMarker != "" {
+		nextMarker = resp.NextMarker
+		isTruncated = true
 	}
 
 	return s3response.ListObjectsV2Result{

tests/integration/group-tests.go

@@ -236,7 +236,10 @@ func TestListObjects(s *S3Conf) {
 
 func TestListObjectsV2(s *S3Conf) {
 	ListObjectsV2_start_after(s)
-	ListObjectsV2_both_start_after_and_continuation_token(s)
+	// posix continuation token not compatible with azure
+	if !s.azureTests {
+		ListObjectsV2_both_start_after_and_continuation_token(s)
+	}
 	ListObjectsV2_start_after_not_in_list(s)
 	ListObjectsV2_start_after_empty_result(s)
 	ListObjectsV2_both_delimiter_and_prefix(s)
@@ -246,6 +249,7 @@ func TestListObjectsV2(s *S3Conf) {
 	ListObjectsV2_exceeding_max_keys(s)
 	ListObjectsV2_list_all_objs(s)
 	ListObjectsV2_with_owner(s)
+	ListObjectsV2_non_truncated_common_prefixes(s)
 	//TODO: remove the condition after implementing checksums in azure
 	if !s.azureTests {
 		ListObjectsV2_with_checksum(s)
@@ -1146,6 +1150,7 @@ func GetIntTests() IntTests {
 		"ListObjects_nested_dir_file_objs":                                        ListObjects_nested_dir_file_objs,
 		"ListObjects_check_owner":                                                 ListObjects_check_owner,
 		"ListObjects_non_truncated_common_prefixes":                               ListObjects_non_truncated_common_prefixes,
+		"ListObjectsV2_non_truncated_common_prefixes":                             ListObjectsV2_non_truncated_common_prefixes,
 		"ListObjects_with_checksum":                                               ListObjects_with_checksum,
 		"ListObjectsV2_start_after":                                               ListObjectsV2_start_after,
 		"ListObjectsV2_both_start_after_and_continuation_token":                   ListObjectsV2_both_start_after_and_continuation_token,

tests/integration/tests.go

@@ -5142,6 +5142,7 @@ func ListObjects_list_all_objs(s *S3Conf) error {
 			return err
 		}
 
+		// Test 1: List all objects without pagination
 		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
 		out, err := s3client.ListObjects(ctx, &s3.ListObjectsInput{
 			Bucket: &bucket,
@@ -5173,6 +5174,36 @@ func ListObjects_list_all_objs(s *S3Conf) error {
 				contents, out.Contents)
 		}
 
+		// Test 2: List all objects with pagination using ListObjectsV2
+		var marker *string
+		var allObjects []types.Object
+		maxKeys := int32(2)
+
+		for {
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.ListObjects(ctx, &s3.ListObjectsInput{
+				Bucket:  &bucket,
+				MaxKeys: &maxKeys,
+				Marker:  marker,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			allObjects = append(allObjects, out.Contents...)
+
+			if out.NextMarker == nil || !*out.IsTruncated {
+				break
+			}
+			marker = out.NextMarker
+		}
+
+		if !compareObjects(contents, allObjects) {
+			return fmt.Errorf("expected the contents to be %v, instead got %v",
+				contents, allObjects)
+		}
+
 		return nil
 	})
 }
@@ -5585,10 +5616,6 @@ func ListObjectsV2_truncated_common_prefixes(s *S3Conf) error {
 			return fmt.Errorf("expected the max-keys to be %v, instead got %v",
 				maxKeys, *out.MaxKeys)
 		}
-		if getString(out.NextContinuationToken) != "d3/" {
-			return fmt.Errorf("expected the NextContinuationToken to be d3/, instead got %v",
-				getString(out.NextContinuationToken))
-		}
 		if getString(out.Delimiter) != delim {
 			return fmt.Errorf("expected the delimiter to be %v, instead got %v",
 				delim, getString(out.Delimiter))
@@ -5618,6 +5645,59 @@ func ListObjectsV2_truncated_common_prefixes(s *S3Conf) error {
 	})
 }
 
+func ListObjectsV2_non_truncated_common_prefixes(s *S3Conf) error {
+	testName := "ListObjectsV2_non_truncated_common_prefixes"
+	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
+		_, err := putObjects(s3client, []string{"asdf", "boo/bar", "boo/baz/xyzzy", "cquux/thud", "cquux/bla"}, bucket)
+		if err != nil {
+			return err
+		}
+
+		delim, marker, maxKeys := "/", "boo/", int32(1)
+
+		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+		res, err := s3client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+			Bucket:     &bucket,
+			StartAfter: &marker,
+			Delimiter:  &delim,
+			MaxKeys:    &maxKeys,
+		})
+		cancel()
+		if err != nil {
+			return err
+		}
+
+		if res.IsTruncated == nil {
+			return fmt.Errorf("expected non-nil istruncated")
+		}
+		if *res.IsTruncated {
+			return fmt.Errorf("expected non-truncated result")
+		}
+		if res.MaxKeys == nil {
+			return fmt.Errorf("expected non nil max-keys")
+		}
+		if *res.MaxKeys != maxKeys {
+			return fmt.Errorf("expected max-keys to be %v, instead got %v",
+				maxKeys, *res.MaxKeys)
+		}
+		if getString(res.Delimiter) != delim {
+			return fmt.Errorf("expected delimiter to be %v, instead got %v",
+				delim, getString(res.Delimiter))
+		}
+		if len(res.Contents) != 0 {
+			return fmt.Errorf("expected empty contents, instead got %+v",
+				res.Contents)
+		}
+		cPrefs := []string{"cquux/"}
+		if !comparePrefixes(cPrefs, res.CommonPrefixes) {
+			return fmt.Errorf("expected common prefixes to be %v, instead got %+v",
+				cPrefs, res.CommonPrefixes)
+		}
+
+		return nil
+	})
+}
+
 func ListObjectsV2_all_objs_max_keys(s *S3Conf) error {
 	testName := "ListObjectsV2_all_objs_max_keys"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
@@ -5691,11 +5771,12 @@ func ListObjectsV2_exceeding_max_keys(s *S3Conf) error {
 func ListObjectsV2_list_all_objs(s *S3Conf) error {
 	testName := "ListObjectsV2_list_all_objs"
 	return actionHandler(s, testName, func(s3client *s3.Client, bucket string) error {
-		contents, err := putObjects(s3client, []string{"bar", "baz", "foo", "obj1", "hell/", "xyzz/quxx"}, bucket)
+		contents, err := putObjects(s3client, []string{"a", "aa", "aaa", "aaaa", "bar", "baz", "foo", "obj1", "hello/world", "xyzz/quxx"}, bucket)
 		if err != nil {
 			return err
 		}
 
+		// Test 1: List all objects without pagination
 		ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
 		out, err := s3client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
 			Bucket: &bucket,
@@ -5731,6 +5812,36 @@ func ListObjectsV2_list_all_objs(s *S3Conf) error {
 				contents, out.Contents)
 		}
 
+		// Test 2: List all objects with pagination using ListObjectsV2
+		var continuationToken *string
+		var allObjects []types.Object
+		maxKeys := int32(2)
+
+		for {
+			ctx, cancel := context.WithTimeout(context.Background(), shortTimeout)
+			out, err := s3client.ListObjectsV2(ctx, &s3.ListObjectsV2Input{
+				Bucket:            &bucket,
+				MaxKeys:           &maxKeys,
+				ContinuationToken: continuationToken,
+			})
+			cancel()
+			if err != nil {
+				return err
+			}
+
+			allObjects = append(allObjects, out.Contents...)
+
+			if out.NextContinuationToken == nil || !*out.IsTruncated {
+				break
+			}
+			continuationToken = out.NextContinuationToken
+		}
+
+		if !compareObjects(contents, allObjects) {
+			return fmt.Errorf("expected the paginated contents to be %v, instead got %v",
+				contents, allObjects)
+		}
+
 		return nil
 	})
 }