From 143f6970d8639843e19412fe2607bf3bad734bb1 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Wed, 16 Sep 2026 13:16:37 +0800 Subject: [PATCH] fix: make multipart discovery and cancellation explicit and bounded Signed-off-by: Feng Ruohang --- .github/workflows/go.yml | 3 + CHANGELOG.md | 20 +- .../rebrand-guard/compat-baseline.json | 2 + cmd/admin-router.go | 1 + cmd/api-errors.go | 16 + cmd/apierrorcode_string.go | 8 +- cmd/erasure-multipart-listing.go | 385 +++++++++ cmd/erasure-multipart-listing_test.go | 751 ++++++++++++++++++ cmd/erasure-multipart.go | 274 +++---- cmd/erasure-server-pool.go | 47 +- cmd/erasure-sets.go | 16 +- cmd/handler-api.go | 8 + cmd/list-multipart-uploads-compat_test.go | 35 +- cmd/object-api-input-checks.go | 4 + cmd/storage-rest_test.go | 9 + internal/config/api/api.go | 8 + internal/config/api/api_test.go | 41 + internal/config/api/help.go | 6 + 18 files changed, 1426 insertions(+), 208 deletions(-) create mode 100644 cmd/erasure-multipart-listing.go create mode 100644 cmd/erasure-multipart-listing_test.go create mode 100644 internal/config/api/api_test.go diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 2b5ef6bdc..0baad8fe6 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -93,6 +93,9 @@ jobs: - name: Run conditional PUT tests under race detector run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP)' -count=1 -timeout=5m + - name: Run multipart listing and cancellation tests under race detector + run: go test -race ./cmd -run '^Test(MultipartListing|MultipartAbort|PaginateMultipartUploads|ListMultipartUploads)' -count=1 -timeout=5m + crosscompile: name: Cross Compile runs-on: ubuntu-latest diff --git a/CHANGELOG.md b/CHANGELOG.md index 6a022bb3c..fb3b85689 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -51,12 +51,24 @@ and [complete commit range](https://github.com/pgsty/silo/compare/RELEASE.2026-0 - Make `ListMultipartUploads` discover quorum-valid uploads from durable state across pools, erasure sets and drives, then apply S3 prefix, delimiter, - marker, ordering and 1,000-entry pagination semantics globally. New uploads + marker, ordering and 1,000-entry pagination semantics globally (#198). New uploads store their canonical bucket and key as reserved fields in the existing - quorum-written `xl.meta`; completion removes those upload-only fields. During - rolling upgrades, detection of any legacy keyless upload retains the prior - listing behavior until those uploads drain. See [issue #79](https://github.com/pgsty/silo/issues/79) + quorum-written `xl.meta`; completion removes those upload-only fields. Native + markers remain usable after their upload is completed or canceled. Strict + listing returns a diagnostic 503 for legacy uploads or uncertain coverage; + `api multipart_listing=legacy` is an explicit temporary migration mode. + Upgrade every writer, drain old uploads and check the read-only admin + `multipart-preflight` report before relying on strict listing. Per-process + admission, directory-entry, worker and time budgets bound scan scheduling; + each page still scans durable state. See [issue #79](https://github.com/pgsty/silo/issues/79) and its [design record](https://silo.pgsty.com/blog/design/list-multipart-uploads/). + Thanks to mr javad seydi (@mrjavadseydi) for the original implementation. +- Confirm multipart cancellation on a strict majority of each relevant set, + and allow retries after partial deletion. Uncertain pools or insufficient + confirmations return 503 rather than acknowledging a cancellation whose + static remnants can later become readable. **Known boundary:** creation + writes that finish after a storage timeout can still restore an upload after + successful cancellation; this change does not add a durable creation fence. - Preserve object tags during multi-pool metadata reconciliation by reading the resolved tag field together with its revision (#189). Previously, reconciliation could replace existing tags with an empty value. diff --git a/buildscripts/rebrand-guard/compat-baseline.json b/buildscripts/rebrand-guard/compat-baseline.json index e158977f3..df194031f 100644 --- a/buildscripts/rebrand-guard/compat-baseline.json +++ b/buildscripts/rebrand-guard/compat-baseline.json @@ -134,6 +134,7 @@ "MINIO_API_GZIP_OBJECTS", "MINIO_API_LEGACY_BUCKET_RESOURCE_MATCH", "MINIO_API_LIST_QUORUM", + "MINIO_API_MULTIPART_LISTING", "MINIO_API_OBJECT_MAX_VERSIONS", "MINIO_API_ODIRECT", "MINIO_API_REMOTE_TRANSPORT_DEADLINE", @@ -766,6 +767,7 @@ "/metrics/v3", "/minio/grid/", "/minio/grid/lock/", + "/multipart-preflight", "/netperf", "/notification", "/oauth2/callback", diff --git a/cmd/admin-router.go b/cmd/admin-router.go index ebda4d698..cf96ada97 100644 --- a/cmd/admin-router.go +++ b/cmd/admin-router.go @@ -163,6 +163,7 @@ func registerAdminRouter(router *mux.Router, enableConfigOps bool) { // StorageInfo operations adminRouter.Methods(http.MethodGet).Path(adminVersion + "/storageinfo").HandlerFunc(adminMiddleware(adminAPI.StorageInfoHandler, traceAllFlag)) + adminRouter.Methods(http.MethodGet).Path(adminVersion + "/multipart-preflight").HandlerFunc(adminMiddleware(adminAPI.MultipartPreflightHandler, traceAllFlag)) // DataUsageInfo operations adminRouter.Methods(http.MethodGet).Path(adminVersion + "/datausageinfo").HandlerFunc(adminMiddleware(adminAPI.DataUsageInfoHandler, traceAllFlag)) // Metrics operation diff --git a/cmd/api-errors.go b/cmd/api-errors.go index 566fc09b7..52e1beedf 100644 --- a/cmd/api-errors.go +++ b/cmd/api-errors.go @@ -450,6 +450,8 @@ const ( ErrAdminNoSecretKey ErrIAMNotInitialized + ErrMultipartListingLegacy + ErrMultipartListingIdentity apiErrCodeEnd // This is used only for the testing code ) @@ -1336,6 +1338,16 @@ var errorCodes = errorCodeMap{ Description: "IAM sub-system not initialized yet, please try again.", HTTPStatusCode: http.StatusServiceUnavailable, }, + ErrMultipartListingLegacy: { + Code: "MultipartListingNotReady", + Description: "Legacy multipart uploads prevent a complete listing. Upgrade all writers, drain old uploads and run the multipart preflight check.", + HTTPStatusCode: http.StatusServiceUnavailable, + }, + ErrMultipartListingIdentity: { + Code: "MultipartListingMetadataInvalid", + Description: "Multipart upload metadata is inconsistent. Run the multipart preflight check to locate the affected storage set.", + HTTPStatusCode: http.StatusServiceUnavailable, + }, ErrBucketMetadataNotInitialized: { Code: "XMinioBucketMetadataNotInitialized", Description: "Bucket metadata not initialized yet, please try again.", @@ -2173,6 +2185,10 @@ func toAPIErrorCode(ctx context.Context, err error) (apiErr APIErrorCode) { err = unwrapAll(err) switch err { + case errMultipartListingLegacy: + apiErr = ErrMultipartListingLegacy + case errMultipartListingIdentity: + apiErr = ErrMultipartListingIdentity case errCompleteMultipartChecksumMismatch, errCompleteMultipartChecksumTypeMismatch: apiErr = ErrBadDigest case errMissingPartChecksum: diff --git a/cmd/apierrorcode_string.go b/cmd/apierrorcode_string.go index 78723c34c..b3425939b 100644 --- a/cmd/apierrorcode_string.go +++ b/cmd/apierrorcode_string.go @@ -339,12 +339,14 @@ func _() { _ = x[ErrAdminNoAccessKey-328] _ = x[ErrAdminNoSecretKey-329] _ = x[ErrIAMNotInitialized-330] - _ = x[apiErrCodeEnd-331] + _ = x[ErrMultipartListingLegacy-331] + _ = x[ErrMultipartListingIdentity-332] + _ = x[apiErrCodeEnd-333] } -const _APIErrorCode_name = "NoneAccessDeniedBadDigestEntityTooSmallEntityTooLargePolicyTooLargeIncompleteBodyInternalErrorInvalidAccessKeyIDAccessKeyDisabledInvalidArgumentInvalidBucketNameInvalidDigestInvalidRangeInvalidRangePartNumberInvalidCopyPartRangeInvalidCopyPartRangeSourceInvalidMaxKeysInvalidEncodingMethodInvalidMaxUploadsInvalidMaxPartsInvalidPartNumberMarkerInvalidPartNumberInvalidRequestBodyInvalidCopySourceInvalidMetadataDirectiveInvalidCopyDestInvalidPolicyDocumentInvalidObjectStateMalformedXMLMissingContentLengthMissingContentMD5MissingRequestBodyErrorMissingSecurityHeaderNoSuchBucketNoSuchBucketPolicyNoSuchBucketLifecycleNoSuchLifecycleConfigurationInvalidLifecycleWithObjectLockNoSuchBucketSSEConfigNoSuchCORSConfigurationNoSuchWebsiteConfigurationReplicationConfigurationNotFoundErrorRemoteDestinationNotFoundErrorReplicationDestinationMissingLockRemoteTargetNotFoundErrorReplicationRemoteConnectionErrorReplicationBandwidthLimitErrorBucketRemoteIdenticalToSourceBucketRemoteAlreadyExistsBucketRemoteLabelInUseBucketRemoteArnTypeInvalidBucketRemoteArnInvalidBucketRemoteRemoveDisallowedRemoteTargetNotVersionedErrorReplicationSourceNotVersionedErrorReplicationNeedsVersioningErrorReplicationBucketNeedsVersioningErrorReplicationDenyEditErrorRemoteTargetDenyAddErrorReplicationNoExistingObjectsReplicationValidationErrorReplicationPermissionCheckErrorObjectRestoreAlreadyInProgressNoSuchKeyNoSuchUploadInvalidVersionIDNoSuchVersionNotImplementedPreconditionFailedRequestTimeTooSkewedSignatureDoesNotMatchMethodNotAllowedInvalidPartInvalidPartOrderMissingPartAuthorizationHeaderMalformedMalformedPOSTRequestPOSTFileRequiredSignatureVersionNotSupportedBucketNotEmptyAllAccessDisabledPolicyInvalidVersionMissingFieldsMissingCredTagCredMalformedInvalidRegionInvalidServiceS3InvalidServiceSTSInvalidRequestVersionMissingSignTagMissingSignHeadersTagMalformedDateMalformedPresignedDateMalformedCredentialDateMalformedExpiresNegativeExpiresAuthHeaderEmptyExpiredPresignRequestRequestNotReadyYetUnsignedHeadersMissingDateHeaderInvalidQuerySignatureAlgoInvalidQueryParamsBucketAlreadyOwnedByYouInvalidDurationBucketAlreadyExistsMetadataTooLargeUnsupportedMetadataUnsupportedHostHeaderMaximumExpiresSlowDownReadSlowDownWriteMaxVersionsExceededInvalidPrefixMarkerBadRequestKeyTooLongErrorInvalidBucketObjectLockConfigurationObjectLockConfigurationNotFoundObjectLockConfigurationNotAllowedNoSuchObjectLockConfigurationObjectLockedInvalidRetentionDatePastObjectLockRetainDateUnknownWORMModeDirectiveBucketTaggingNotFoundObjectLockInvalidHeadersInvalidTagDirectivePolicyAlreadyAttachedPolicyNotAttachedExcessDataPolicyInvalidNameNoTokenRevokeTypeAdminOpenIDNotEnabledAdminNoSuchAccessKeyInvalidEncryptionMethodInvalidEncryptionKeyIDInsecureSSECustomerRequestSSEMultipartEncryptedSSEEncryptedObjectInvalidEncryptionParametersInvalidEncryptionParametersSSECInvalidSSECustomerAlgorithmInvalidSSECustomerKeyMissingSSECustomerKeyMissingSSECustomerKeyMD5SSECustomerKeyMD5MismatchInvalidSSECustomerParametersIncompatibleEncryptionMethodKMSNotConfiguredKMSKeyNotFoundExceptionKMSDefaultKeyAlreadyConfiguredNoAccessKeyInvalidTokenEventNotificationARNNotificationRegionNotificationOverlappingFilterNotificationFilterNameInvalidFilterNamePrefixFilterNameSuffixFilterValueInvalidOverlappingConfigsUnsupportedNotificationContentSHA256MismatchContentChecksumMismatchStorageFullRequestBodyParseObjectExistsAsDirectoryInvalidObjectNameInvalidObjectNamePrefixSlashInvalidResourceNameInvalidLifecycleQueryParameterServerNotInitializedBucketMetadataNotInitializedRequestTimedoutClientDisconnectedTooManyRequestsInvalidRequestTransitionStorageClassNotFoundErrorInvalidStorageClassBackendDownMalformedJSONAdminNoSuchUserAdminNoSuchUserLDAPWarnAdminLDAPExpectedLoginNameAdminNoSuchGroupAdminGroupNotEmptyAdminGroupDisabledAdminInvalidGroupNameAdminNoSuchJobAdminNoSuchPolicyAdminPolicyChangeAlreadyAppliedAdminInvalidArgumentAdminInvalidAccessKeyAdminInvalidSecretKeyAdminConfigNoQuorumAdminConfigTooLargeAdminConfigBadJSONAdminNoSuchConfigTargetAdminConfigEnvOverriddenAdminConfigDuplicateKeysAdminConfigInvalidIDPTypeAdminConfigLDAPNonDefaultConfigNameAdminConfigLDAPValidationAdminConfigIDPCfgNameAlreadyExistsAdminConfigIDPCfgNameDoesNotExistInsecureClientRequestObjectTamperedAdminLDAPNotEnabledSiteReplicationInvalidRequestSiteReplicationPeerRespSiteReplicationBackendIssueSiteReplicationServiceAccountErrorSiteReplicationBucketConfigErrorSiteReplicationBucketMetaErrorSiteReplicationIAMErrorSiteReplicationConfigMissingSiteReplicationIAMConfigMismatchAdminRebalanceAlreadyStartedAdminRebalanceNotStartedAdminBucketQuotaExceededAdminNoSuchQuotaConfigurationHealNotImplementedHealNoSuchProcessHealInvalidClientTokenHealMissingBucketHealAlreadyRunningHealOverlappingPathsIncorrectContinuationTokenEmptyRequestBodyUnsupportedFunctionInvalidExpressionTypeBusyUnauthorizedAccessExpressionTooLongIllegalSQLFunctionArgumentInvalidKeyPathInvalidCompressionFormatInvalidFileHeaderInfoInvalidJSONTypeInvalidQuoteFieldsInvalidRequestParameterInvalidDataTypeInvalidTextEncodingInvalidDataSourceInvalidTableAliasMissingRequiredParameterObjectSerializationConflictUnsupportedSQLOperationUnsupportedSQLStructureUnsupportedSyntaxUnsupportedRangeHeaderLexerInvalidCharLexerInvalidOperatorLexerInvalidLiteralLexerInvalidIONLiteralParseExpectedDatePartParseExpectedKeywordParseExpectedTokenTypeParseExpected2TokenTypesParseExpectedNumberParseExpectedRightParenBuiltinFunctionCallParseExpectedTypeNameParseExpectedWhenClauseParseUnsupportedTokenParseUnsupportedLiteralsGroupByParseExpectedMemberParseUnsupportedSelectParseUnsupportedCaseParseUnsupportedCaseClauseParseUnsupportedAliasParseUnsupportedSyntaxParseUnknownOperatorParseMissingIdentAfterAtParseUnexpectedOperatorParseUnexpectedTermParseUnexpectedTokenParseUnexpectedKeywordParseExpectedExpressionParseExpectedLeftParenAfterCastParseExpectedLeftParenValueConstructorParseExpectedLeftParenBuiltinFunctionCallParseExpectedArgumentDelimiterParseCastArityParseInvalidTypeParamParseEmptySelectParseSelectMissingFromParseExpectedIdentForGroupNameParseExpectedIdentForAliasParseUnsupportedCallWithStarParseNonUnaryAggregateFunctionCallParseMalformedJoinParseExpectedIdentForAtParseAsteriskIsNotAloneInSelectListParseCannotMixSqbAndWildcardInSelectListParseInvalidContextForWildcardInSelectListIncorrectSQLFunctionArgumentTypeValueParseFailureEvaluatorInvalidArgumentsIntegerOverflowLikeInvalidInputsCastFailedInvalidCastEvaluatorInvalidTimestampFormatPatternEvaluatorInvalidTimestampFormatPatternSymbolForParsingEvaluatorTimestampFormatPatternDuplicateFieldsEvaluatorTimestampFormatPatternHourClockAmPmMismatchEvaluatorUnterminatedTimestampFormatPatternTokenEvaluatorInvalidTimestampFormatPatternTokenEvaluatorInvalidTimestampFormatPatternSymbolEvaluatorBindingDoesNotExistMissingHeadersInvalidColumnIndexAdminConfigNotificationTargetsFailedAdminProfilerNotEnabledInvalidDecompressedSizeAddUserInvalidArgumentAddUserValidUTFAdminResourceInvalidArgumentAdminAccountNotEligibleAccountNotEligibleAdminServiceAccountNotFoundPostPolicyConditionInvalidFormatInvalidChecksumLambdaARNInvalidLambdaARNNotFoundInvalidAttributeNameAdminNoAccessKeyAdminNoSecretKeyIAMNotInitializedapiErrCodeEnd" +const _APIErrorCode_name = "NoneAccessDeniedBadDigestEntityTooSmallEntityTooLargePolicyTooLargeIncompleteBodyInternalErrorInvalidAccessKeyIDAccessKeyDisabledInvalidArgumentInvalidBucketNameInvalidDigestInvalidRangeInvalidRangePartNumberInvalidCopyPartRangeInvalidCopyPartRangeSourceInvalidMaxKeysInvalidEncodingMethodInvalidMaxUploadsInvalidMaxPartsInvalidPartNumberMarkerInvalidPartNumberInvalidRequestBodyInvalidCopySourceInvalidMetadataDirectiveInvalidCopyDestInvalidPolicyDocumentInvalidObjectStateMalformedXMLMissingContentLengthMissingContentMD5MissingRequestBodyErrorMissingSecurityHeaderNoSuchBucketNoSuchBucketPolicyNoSuchBucketLifecycleNoSuchLifecycleConfigurationInvalidLifecycleWithObjectLockNoSuchBucketSSEConfigNoSuchCORSConfigurationNoSuchWebsiteConfigurationReplicationConfigurationNotFoundErrorRemoteDestinationNotFoundErrorReplicationDestinationMissingLockRemoteTargetNotFoundErrorReplicationRemoteConnectionErrorReplicationBandwidthLimitErrorBucketRemoteIdenticalToSourceBucketRemoteAlreadyExistsBucketRemoteLabelInUseBucketRemoteArnTypeInvalidBucketRemoteArnInvalidBucketRemoteRemoveDisallowedRemoteTargetNotVersionedErrorReplicationSourceNotVersionedErrorReplicationNeedsVersioningErrorReplicationBucketNeedsVersioningErrorReplicationDenyEditErrorRemoteTargetDenyAddErrorReplicationNoExistingObjectsReplicationValidationErrorReplicationPermissionCheckErrorObjectRestoreAlreadyInProgressNoSuchKeyNoSuchUploadInvalidVersionIDNoSuchVersionNotImplementedPreconditionFailedRequestTimeTooSkewedSignatureDoesNotMatchMethodNotAllowedInvalidPartInvalidPartOrderMissingPartAuthorizationHeaderMalformedMalformedPOSTRequestPOSTFileRequiredSignatureVersionNotSupportedBucketNotEmptyAllAccessDisabledPolicyInvalidVersionMissingFieldsMissingCredTagCredMalformedInvalidRegionInvalidServiceS3InvalidServiceSTSInvalidRequestVersionMissingSignTagMissingSignHeadersTagMalformedDateMalformedPresignedDateMalformedCredentialDateMalformedExpiresNegativeExpiresAuthHeaderEmptyExpiredPresignRequestRequestNotReadyYetUnsignedHeadersMissingDateHeaderInvalidQuerySignatureAlgoInvalidQueryParamsBucketAlreadyOwnedByYouInvalidDurationBucketAlreadyExistsMetadataTooLargeUnsupportedMetadataUnsupportedHostHeaderMaximumExpiresSlowDownReadSlowDownWriteMaxVersionsExceededInvalidPrefixMarkerBadRequestKeyTooLongErrorInvalidBucketObjectLockConfigurationObjectLockConfigurationNotFoundObjectLockConfigurationNotAllowedNoSuchObjectLockConfigurationObjectLockedInvalidRetentionDatePastObjectLockRetainDateUnknownWORMModeDirectiveBucketTaggingNotFoundObjectLockInvalidHeadersInvalidTagDirectivePolicyAlreadyAttachedPolicyNotAttachedExcessDataPolicyInvalidNameNoTokenRevokeTypeAdminOpenIDNotEnabledAdminNoSuchAccessKeyInvalidEncryptionMethodInvalidEncryptionKeyIDInsecureSSECustomerRequestSSEMultipartEncryptedSSEEncryptedObjectInvalidEncryptionParametersInvalidEncryptionParametersSSECInvalidSSECustomerAlgorithmInvalidSSECustomerKeyMissingSSECustomerKeyMissingSSECustomerKeyMD5SSECustomerKeyMD5MismatchInvalidSSECustomerParametersIncompatibleEncryptionMethodKMSNotConfiguredKMSKeyNotFoundExceptionKMSDefaultKeyAlreadyConfiguredNoAccessKeyInvalidTokenEventNotificationARNNotificationRegionNotificationOverlappingFilterNotificationFilterNameInvalidFilterNamePrefixFilterNameSuffixFilterValueInvalidOverlappingConfigsUnsupportedNotificationContentSHA256MismatchContentChecksumMismatchStorageFullRequestBodyParseObjectExistsAsDirectoryInvalidObjectNameInvalidObjectNamePrefixSlashInvalidResourceNameInvalidLifecycleQueryParameterServerNotInitializedBucketMetadataNotInitializedRequestTimedoutClientDisconnectedTooManyRequestsInvalidRequestTransitionStorageClassNotFoundErrorInvalidStorageClassBackendDownMalformedJSONAdminNoSuchUserAdminNoSuchUserLDAPWarnAdminLDAPExpectedLoginNameAdminNoSuchGroupAdminGroupNotEmptyAdminGroupDisabledAdminInvalidGroupNameAdminNoSuchJobAdminNoSuchPolicyAdminPolicyChangeAlreadyAppliedAdminInvalidArgumentAdminInvalidAccessKeyAdminInvalidSecretKeyAdminConfigNoQuorumAdminConfigTooLargeAdminConfigBadJSONAdminNoSuchConfigTargetAdminConfigEnvOverriddenAdminConfigDuplicateKeysAdminConfigInvalidIDPTypeAdminConfigLDAPNonDefaultConfigNameAdminConfigLDAPValidationAdminConfigIDPCfgNameAlreadyExistsAdminConfigIDPCfgNameDoesNotExistInsecureClientRequestObjectTamperedAdminLDAPNotEnabledSiteReplicationInvalidRequestSiteReplicationPeerRespSiteReplicationBackendIssueSiteReplicationServiceAccountErrorSiteReplicationBucketConfigErrorSiteReplicationBucketMetaErrorSiteReplicationIAMErrorSiteReplicationConfigMissingSiteReplicationIAMConfigMismatchAdminRebalanceAlreadyStartedAdminRebalanceNotStartedAdminBucketQuotaExceededAdminNoSuchQuotaConfigurationHealNotImplementedHealNoSuchProcessHealInvalidClientTokenHealMissingBucketHealAlreadyRunningHealOverlappingPathsIncorrectContinuationTokenEmptyRequestBodyUnsupportedFunctionInvalidExpressionTypeBusyUnauthorizedAccessExpressionTooLongIllegalSQLFunctionArgumentInvalidKeyPathInvalidCompressionFormatInvalidFileHeaderInfoInvalidJSONTypeInvalidQuoteFieldsInvalidRequestParameterInvalidDataTypeInvalidTextEncodingInvalidDataSourceInvalidTableAliasMissingRequiredParameterObjectSerializationConflictUnsupportedSQLOperationUnsupportedSQLStructureUnsupportedSyntaxUnsupportedRangeHeaderLexerInvalidCharLexerInvalidOperatorLexerInvalidLiteralLexerInvalidIONLiteralParseExpectedDatePartParseExpectedKeywordParseExpectedTokenTypeParseExpected2TokenTypesParseExpectedNumberParseExpectedRightParenBuiltinFunctionCallParseExpectedTypeNameParseExpectedWhenClauseParseUnsupportedTokenParseUnsupportedLiteralsGroupByParseExpectedMemberParseUnsupportedSelectParseUnsupportedCaseParseUnsupportedCaseClauseParseUnsupportedAliasParseUnsupportedSyntaxParseUnknownOperatorParseMissingIdentAfterAtParseUnexpectedOperatorParseUnexpectedTermParseUnexpectedTokenParseUnexpectedKeywordParseExpectedExpressionParseExpectedLeftParenAfterCastParseExpectedLeftParenValueConstructorParseExpectedLeftParenBuiltinFunctionCallParseExpectedArgumentDelimiterParseCastArityParseInvalidTypeParamParseEmptySelectParseSelectMissingFromParseExpectedIdentForGroupNameParseExpectedIdentForAliasParseUnsupportedCallWithStarParseNonUnaryAggregateFunctionCallParseMalformedJoinParseExpectedIdentForAtParseAsteriskIsNotAloneInSelectListParseCannotMixSqbAndWildcardInSelectListParseInvalidContextForWildcardInSelectListIncorrectSQLFunctionArgumentTypeValueParseFailureEvaluatorInvalidArgumentsIntegerOverflowLikeInvalidInputsCastFailedInvalidCastEvaluatorInvalidTimestampFormatPatternEvaluatorInvalidTimestampFormatPatternSymbolForParsingEvaluatorTimestampFormatPatternDuplicateFieldsEvaluatorTimestampFormatPatternHourClockAmPmMismatchEvaluatorUnterminatedTimestampFormatPatternTokenEvaluatorInvalidTimestampFormatPatternTokenEvaluatorInvalidTimestampFormatPatternSymbolEvaluatorBindingDoesNotExistMissingHeadersInvalidColumnIndexAdminConfigNotificationTargetsFailedAdminProfilerNotEnabledInvalidDecompressedSizeAddUserInvalidArgumentAddUserValidUTFAdminResourceInvalidArgumentAdminAccountNotEligibleAccountNotEligibleAdminServiceAccountNotFoundPostPolicyConditionInvalidFormatInvalidChecksumLambdaARNInvalidLambdaARNNotFoundInvalidAttributeNameAdminNoAccessKeyAdminNoSecretKeyIAMNotInitializedMultipartListingLegacyMultipartListingIdentityapiErrCodeEnd" -var _APIErrorCode_index = [...]uint16{0, 4, 16, 25, 39, 53, 67, 81, 94, 112, 129, 144, 161, 174, 186, 208, 228, 254, 268, 289, 306, 321, 344, 361, 379, 396, 420, 435, 456, 474, 486, 506, 523, 546, 567, 579, 597, 618, 646, 676, 697, 720, 746, 783, 813, 846, 871, 903, 933, 962, 987, 1009, 1035, 1057, 1085, 1114, 1148, 1179, 1216, 1240, 1264, 1292, 1318, 1349, 1379, 1388, 1400, 1416, 1429, 1443, 1461, 1481, 1502, 1518, 1529, 1545, 1556, 1584, 1604, 1620, 1648, 1662, 1679, 1699, 1712, 1726, 1739, 1752, 1768, 1785, 1806, 1820, 1841, 1854, 1876, 1899, 1915, 1930, 1945, 1966, 1984, 1999, 2016, 2041, 2059, 2082, 2097, 2116, 2132, 2151, 2172, 2186, 2198, 2211, 2230, 2249, 2259, 2274, 2310, 2341, 2374, 2403, 2415, 2435, 2459, 2483, 2504, 2528, 2547, 2568, 2585, 2595, 2612, 2629, 2650, 2670, 2693, 2715, 2741, 2762, 2780, 2807, 2838, 2865, 2886, 2907, 2931, 2956, 2984, 3012, 3028, 3051, 3081, 3092, 3104, 3121, 3136, 3154, 3183, 3200, 3216, 3232, 3250, 3268, 3291, 3312, 3335, 3346, 3362, 3385, 3402, 3430, 3449, 3479, 3499, 3527, 3542, 3560, 3575, 3589, 3624, 3643, 3654, 3667, 3682, 3705, 3731, 3747, 3765, 3783, 3804, 3818, 3835, 3866, 3886, 3907, 3928, 3947, 3966, 3984, 4007, 4031, 4055, 4080, 4115, 4140, 4174, 4207, 4228, 4242, 4261, 4290, 4313, 4340, 4374, 4406, 4436, 4459, 4487, 4519, 4547, 4571, 4595, 4624, 4642, 4659, 4681, 4698, 4716, 4736, 4762, 4778, 4797, 4818, 4822, 4840, 4857, 4883, 4897, 4921, 4942, 4957, 4975, 4998, 5013, 5032, 5049, 5066, 5090, 5117, 5140, 5163, 5180, 5202, 5218, 5238, 5257, 5279, 5300, 5320, 5342, 5366, 5385, 5427, 5448, 5471, 5492, 5523, 5542, 5564, 5584, 5610, 5631, 5653, 5673, 5697, 5720, 5739, 5759, 5781, 5804, 5835, 5873, 5914, 5944, 5958, 5979, 5995, 6017, 6047, 6073, 6101, 6135, 6153, 6176, 6211, 6251, 6293, 6325, 6342, 6367, 6382, 6399, 6409, 6420, 6458, 6512, 6558, 6610, 6658, 6701, 6745, 6773, 6787, 6805, 6841, 6864, 6887, 6909, 6924, 6952, 6975, 6993, 7020, 7052, 7067, 7083, 7100, 7120, 7136, 7152, 7169, 7182} +var _APIErrorCode_index = [...]uint16{0, 4, 16, 25, 39, 53, 67, 81, 94, 112, 129, 144, 161, 174, 186, 208, 228, 254, 268, 289, 306, 321, 344, 361, 379, 396, 420, 435, 456, 474, 486, 506, 523, 546, 567, 579, 597, 618, 646, 676, 697, 720, 746, 783, 813, 846, 871, 903, 933, 962, 987, 1009, 1035, 1057, 1085, 1114, 1148, 1179, 1216, 1240, 1264, 1292, 1318, 1349, 1379, 1388, 1400, 1416, 1429, 1443, 1461, 1481, 1502, 1518, 1529, 1545, 1556, 1584, 1604, 1620, 1648, 1662, 1679, 1699, 1712, 1726, 1739, 1752, 1768, 1785, 1806, 1820, 1841, 1854, 1876, 1899, 1915, 1930, 1945, 1966, 1984, 1999, 2016, 2041, 2059, 2082, 2097, 2116, 2132, 2151, 2172, 2186, 2198, 2211, 2230, 2249, 2259, 2274, 2310, 2341, 2374, 2403, 2415, 2435, 2459, 2483, 2504, 2528, 2547, 2568, 2585, 2595, 2612, 2629, 2650, 2670, 2693, 2715, 2741, 2762, 2780, 2807, 2838, 2865, 2886, 2907, 2931, 2956, 2984, 3012, 3028, 3051, 3081, 3092, 3104, 3121, 3136, 3154, 3183, 3200, 3216, 3232, 3250, 3268, 3291, 3312, 3335, 3346, 3362, 3385, 3402, 3430, 3449, 3479, 3499, 3527, 3542, 3560, 3575, 3589, 3624, 3643, 3654, 3667, 3682, 3705, 3731, 3747, 3765, 3783, 3804, 3818, 3835, 3866, 3886, 3907, 3928, 3947, 3966, 3984, 4007, 4031, 4055, 4080, 4115, 4140, 4174, 4207, 4228, 4242, 4261, 4290, 4313, 4340, 4374, 4406, 4436, 4459, 4487, 4519, 4547, 4571, 4595, 4624, 4642, 4659, 4681, 4698, 4716, 4736, 4762, 4778, 4797, 4818, 4822, 4840, 4857, 4883, 4897, 4921, 4942, 4957, 4975, 4998, 5013, 5032, 5049, 5066, 5090, 5117, 5140, 5163, 5180, 5202, 5218, 5238, 5257, 5279, 5300, 5320, 5342, 5366, 5385, 5427, 5448, 5471, 5492, 5523, 5542, 5564, 5584, 5610, 5631, 5653, 5673, 5697, 5720, 5739, 5759, 5781, 5804, 5835, 5873, 5914, 5944, 5958, 5979, 5995, 6017, 6047, 6073, 6101, 6135, 6153, 6176, 6211, 6251, 6293, 6325, 6342, 6367, 6382, 6399, 6409, 6420, 6458, 6512, 6558, 6610, 6658, 6701, 6745, 6773, 6787, 6805, 6841, 6864, 6887, 6909, 6924, 6952, 6975, 6993, 7020, 7052, 7067, 7083, 7100, 7120, 7136, 7152, 7169, 7191, 7215, 7228} func (i APIErrorCode) String() string { idx := int(i) - 0 diff --git a/cmd/erasure-multipart-listing.go b/cmd/erasure-multipart-listing.go new file mode 100644 index 000000000..cbc9b15fa --- /dev/null +++ b/cmd/erasure-multipart-listing.go @@ -0,0 +1,385 @@ +// Copyright (c) 2026 mr javad seydi and Ruohang Feng +// +// This file is part of Silo Object Storage stack. +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "strings" + "sync" + "sync/atomic" + "time" + + "github.com/pgsty/silo-pkg/v3/policy" +) + +const multipartScanEntryLimit = 100_000 + +var ( + multipartScanSlots = make(chan struct{}, 2) + errMultipartListingLegacy = errors.New("legacy multipart uploads require a coordinated upgrade and drain") + errMultipartListingIdentity = errors.New("multipart upload identity is invalid") +) + +// One budget and admission slot cover the entire request, including all pools. +// A slot is released only after the scan workers have actually stopped. +type multipartScan struct { + ctx context.Context + cancel context.CancelFunc + remaining atomic.Int64 + metadataSlots chan struct{} + preflight bool + sets []multipartScanSet +} + +type multipartScanSet struct { + Pool int `json:"pool"` + Set int `json:"set"` + Drives int `json:"drives"` + ScannedDrives int `json:"scannedDrives"` + UncoveredDrives []int `json:"uncoveredDrives,omitempty"` + Candidates int `json:"candidates"` + LegacyUploads int `json:"legacyUploads"` + OldestLegacy time.Time `json:"oldestLegacy,omitempty"` + Error string `json:"error,omitempty"` +} + +type multipartPreflightReport struct { + Ready bool `json:"ready"` + Complete bool `json:"complete"` + Mode string `json:"mode"` + ScannedEntries int64 `json:"scannedEntries"` + LegacyUploads int `json:"legacyUploads"` + Sets []multipartScanSet `json:"sets"` +} + +func startMultipartScan(ctx context.Context, preflight bool) (*multipartScan, error) { + select { + case multipartScanSlots <- struct{}{}: + case <-ctx.Done(): + return nil, ctx.Err() + default: + return nil, SlowDown{} + } + ctx, cancel := context.WithTimeout(ctx, 30*time.Second) + s := &multipartScan{ctx: ctx, cancel: cancel, preflight: preflight, metadataSlots: make(chan struct{}, multipartMetadataScanConcurrency)} + s.remaining.Store(multipartScanEntryLimit) + return s, nil +} + +func (s *multipartScan) close() { + s.cancel() + <-multipartScanSlots +} + +func (s *multipartScan) listDir(disk StorageAPI, bucket, dir string) ([]string, error) { + if err := s.ctx.Err(); err != nil { + return nil, SlowDown{} + } + remaining := s.remaining.Load() + if remaining <= 0 { + return nil, SlowDown{} + } + // Request one extra entry to detect overflow, never a silently partial page. + entries, err := disk.ListDir(s.ctx, bucket, minioMetaMultipartBucket, dir, int(remaining)+1) + if errors.Is(err, errFileNotFound) { + return nil, nil + } + if err != nil { + return nil, err + } + if s.remaining.Add(-int64(len(entries))) < 0 { + return nil, SlowDown{} + } + return entries, nil +} + +func (s *multipartScan) listUploadDirs(disk StorageAPI, bucket string) ([]string, error) { + hashDirs, err := s.listDir(disk, bucket, "") + if err != nil { + return nil, err + } + var candidates []string + for _, hashDir := range hashDirs { + if !strings.HasSuffix(hashDir, SlashSeparator) { + continue + } + hashDir = strings.TrimSuffix(hashDir, SlashSeparator) + uploadDirs, err := s.listDir(disk, bucket, hashDir) + if err != nil { + return nil, err + } + for _, uploadDir := range uploadDirs { + if strings.HasSuffix(uploadDir, SlashSeparator) { + candidates = append(candidates, pathJoin(hashDir, strings.TrimSuffix(uploadDir, SlashSeparator))) + } + } + } + return candidates, nil +} + +// Identity is immutable at hash/uploadUUID. Check the hash before using even +// a single source drive to exclude another bucket. Missing/bad identities must +// fall back to the full metadata read; they cannot prove absence. +func (er erasureObjects) multipartIdentity(fi FileInfo, shaDir string) (string, string, bool) { + bucket, object := fi.Metadata[multipartMetaBucket], fi.Metadata[multipartMetaObject] + return bucket, object, bucket != "" && object != "" && IsValidBucketName(bucket) && + IsValidObjectPrefix(object) && er.getMultipartSHADir(bucket, object) == shaDir +} + +func (er erasureObjects) readMultipartUploadCandidate(s *multipartScan, bucket, candidate string, source StorageAPI) (MultipartInfo, bool, bool, error) { + if s.ctx.Err() != nil { + return MultipartInfo{}, false, false, SlowDown{} + } + shaDir, uploadUUID, ok := strings.Cut(candidate, SlashSeparator) + if !ok || shaDir == "" || uploadUUID == "" || strings.Contains(uploadUUID, SlashSeparator) { + return MultipartInfo{}, false, false, errMultipartListingIdentity + } + if !s.preflight && bucket != "" && source != nil { + fi, err := source.ReadVersion(s.ctx, bucket, minioMetaMultipartBucket, candidate, "", ReadOptions{}) + if err == nil { + storedBucket, _, valid := er.multipartIdentity(fi, shaDir) + if valid && storedBucket != bucket { + return MultipartInfo{}, false, false, nil + } + } + } + if s.ctx.Err() != nil { + return MultipartInfo{}, false, false, SlowDown{} + } + select { + case s.metadataSlots <- struct{}{}: + defer func() { <-s.metadataSlots }() + case <-s.ctx.Done(): + return MultipartInfo{}, false, false, SlowDown{} + } + disks := er.getDisks() + metadata, errs := readAllFileInfo(s.ctx, disks, bucket, minioMetaMultipartBucket, candidate, "", false, false) + if s.preflight { + // Readiness is stronger than listing liveness: even one readable legacy + // copy must be drained, and an unreadable copy cannot certify readiness. + var oldest MultipartInfo + legacy := false + _, nativeID := multipartUploadTime(uploadUUID) + for i, err := range errs { + if errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound) { + continue + } + if err != nil { + return MultipartInfo{}, false, false, err + } + fi := metadata[i] + storedBucket, storedObject, valid := er.multipartIdentity(fi, shaDir) + if storedBucket != "" && storedObject != "" && !valid { + return MultipartInfo{}, false, false, errMultipartListingIdentity + } + if !valid || !nativeID { + info := multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime) + if !legacy || info.Initiated.Before(oldest.Initiated) { + oldest = info + } + legacy = true + } + } + if legacy { + return oldest, false, true, nil + } + } + readQuorum, _, err := objectQuorumFromMeta(s.ctx, metadata, errs, er.defaultParityCount) + if err != nil { + return MultipartInfo{}, false, false, err + } + _, modTime, etag := listOnlineDisks(disks, metadata, errs, readQuorum) + if err := reduceReadQuorumErrs(s.ctx, errs, objectOpIgnoredErrs, readQuorum); err != nil { + return MultipartInfo{}, false, false, err + } + fi, err := pickValidFileInfo(s.ctx, metadata, modTime, etag, readQuorum) + if err != nil { + return MultipartInfo{}, false, false, err + } + storedBucket, storedObject, valid := er.multipartIdentity(fi, shaDir) + info := multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime) + if storedBucket == "" || storedObject == "" { + return info, false, true, nil + } + if !valid { + return MultipartInfo{}, false, false, fmt.Errorf("%w: %s", errMultipartListingIdentity, candidate) + } + if _, ok := multipartUploadTime(uploadUUID); !ok { + return info, false, true, nil + } + if bucket != "" && bucket != storedBucket { + return MultipartInfo{}, false, false, nil + } + return info, true, false, nil +} + +func (er erasureObjects) scanMultipartUploads(s *multipartScan, bucket string, poolIdx, setIdx int) ([]MultipartInfo, bool, error) { + disks := er.getDisks() + report := multipartScanSet{Pool: poolIdx, Set: setIdx, Drives: er.setDriveCount} + var resultErr error + defer func() { + if resultErr != nil { + report.Error = resultErr.Error() + } + s.sets = append(s.sets, report) + }() + var candidateMu sync.Mutex + candidates := make(map[string]StorageAPI) + errs := make([]error, len(disks)) + var wg sync.WaitGroup + for i, disk := range disks { + wg.Add(1) + go func() { + defer wg.Done() + if disk == nil || !disk.IsOnline() { + errs[i] = errDiskNotFound + return + } + paths, err := s.listUploadDirs(disk, bucket) + errs[i] = err + if err != nil { + return + } + candidateMu.Lock() + defer candidateMu.Unlock() + for _, p := range paths { + candidates[p] = disk + } + }() + } + wg.Wait() + for i, err := range errs { + if err == nil { + report.ScannedDrives++ + } else { + report.UncoveredDrives = append(report.UncoveredDrives, i) + if _, limited := err.(SlowDown); limited { + resultErr = err + } + } + } + report.Candidates = len(candidates) + // A successful scan must intersect every metadata read quorum, including + // records left with R copies by a partially failed cancellation. + if report.ScannedDrives < er.setDriveCount/2+1 && resultErr == nil { + resultErr = toObjectErr(errErasureReadQuorum, bucket) + } + if resultErr != nil { + return nil, false, resultErr + } + type candidate struct { + path string + source StorageAPI + } + jobs := make(chan candidate) + var mu sync.Mutex + var uploads []MultipartInfo + for range min(16, len(candidates)) { + wg.Add(1) + go func() { + defer wg.Done() + for c := range jobs { + mu.Lock() + stopped := resultErr != nil + mu.Unlock() + if stopped || s.ctx.Err() != nil { + continue + } + upload, found, legacy, err := er.readMultipartUploadCandidate(s, bucket, c.path, c.source) + if errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound) { + continue + } + mu.Lock() + switch { + case err != nil && resultErr == nil: + resultErr = err + case legacy: + report.LegacyUploads++ + if report.OldestLegacy.IsZero() || upload.Initiated.Before(report.OldestLegacy) { + report.OldestLegacy = upload.Initiated + } + case found && !s.preflight: + uploads = append(uploads, upload) + } + mu.Unlock() + } + }() + } +dispatch: + for p, source := range candidates { + select { + case jobs <- candidate{p, source}: + case <-s.ctx.Done(): + break dispatch + } + } + close(jobs) + wg.Wait() + if s.ctx.Err() != nil { + resultErr = SlowDown{} + } + return uploads, report.LegacyUploads != 0, resultErr +} + +// multipartPreflight scans all pools, sets and drives, independent of caches. +// A majority suffices for normal listing; upgrade readiness requires every +// drive to have been inspected. The operator must also upgrade all writers. +func (z *erasureServerPools) multipartPreflight(ctx context.Context) (multipartPreflightReport, error) { + s, err := startMultipartScan(ctx, true) + if err != nil { + return multipartPreflightReport{}, err + } + defer s.close() + report := multipartPreflightReport{Complete: true, Mode: "strict"} + if globalAPIConfig.getMultipartListingLegacy() { + report.Mode = "legacy" + } + for p, pool := range z.serverPools { + for i, set := range pool.sets { + _, _, _ = set.scanMultipartUploads(s, "", p, i) + } + } + report.Sets = s.sets + for _, set := range s.sets { + report.LegacyUploads += set.LegacyUploads + if set.Error != "" || set.ScannedDrives != set.Drives { + report.Complete = false + } + } + report.ScannedEntries = multipartScanEntryLimit - s.remaining.Load() + report.Ready = report.Complete && report.LegacyUploads == 0 + return report, nil +} + +// MultipartPreflightHandler is a read-only storage-admin diagnostic. It never +// accepts an arbitrary deletion path or changes upload lifetime settings. +func (a adminAPIHandlers) MultipartPreflightHandler(w http.ResponseWriter, r *http.Request) { + ctx := r.Context() + obj, _ := validateAdminReq(ctx, w, r, policy.StorageInfoAdminAction) + if obj == nil { + return + } + z, ok := obj.(*erasureServerPools) + if !ok { + writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrNotImplemented), r.URL) + return + } + report, err := z.multipartPreflight(ctx) + if err != nil { + writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL) + return + } + b, err := json.Marshal(report) + if err != nil { + writeErrorResponseJSON(ctx, w, toAdminAPIErr(ctx, err), r.URL) + return + } + writeSuccessResponseJSON(w, b) +} diff --git a/cmd/erasure-multipart-listing_test.go b/cmd/erasure-multipart-listing_test.go new file mode 100644 index 000000000..6566252ee --- /dev/null +++ b/cmd/erasure-multipart-listing_test.go @@ -0,0 +1,751 @@ +// Copyright (c) 2026 Ruohang Feng +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "context" + "encoding/base64" + "encoding/json" + "encoding/xml" + "errors" + "fmt" + "net/http" + "net/http/httptest" + "slices" + "strconv" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/minio/minio/internal/config/storageclass" +) + +// Check the real handler and storage path, including a successful abort between +// pages. Choose IDs increasing in both initiation time and lexical order so the +// failure does not depend on differing interpretations of S3 marker ordering. +func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads", "AbortMultipart"}, MakeBucketOptions{}) + if err != nil { + t.Fatal(err) + } + var firstID, secondID string + for attempt := 0; attempt < 32; attempt++ { + one, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + two, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + if one.UploadID < two.UploadID { + firstID, secondID = one.UploadID, two.UploadID + break + } + for _, id := range []string{one.UploadID, two.UploadID} { + if err := z.AbortMultipartUpload(t.Context(), bucket, "a", id, ObjectOptions{}); err != nil { + t.Fatal(err) + } + } + } + if firstID == "" { + t.Fatal("could not construct increasing upload IDs") + } + if _, err := z.NewMultipartUpload(t.Context(), bucket, "b", ObjectOptions{}); err != nil { + t.Fatal(err) + } + request := func(method, u string) *httptest.ResponseRecorder { + t.Helper() + req, err := newTestSignedRequestV4(method, u, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + return rec + } + list := func(keyMarker, uploadMarker string, limit int) ListMultipartUploadsResponse { + t.Helper() + rec := request(http.MethodGet, getListMultipartUploadsURLWithParams("", bucket, "", keyMarker, uploadMarker, "", strconv.Itoa(limit))) + if rec.Code != http.StatusOK { + t.Fatalf("list HTTP %d: %s", rec.Code, rec.Body.String()) + } + var result ListMultipartUploadsResponse + if err := xml.Unmarshal(rec.Body.Bytes(), &result); err != nil { + t.Fatal(err) + } + return result + } + first := list("", "", 1) + if len(first.Uploads) != 1 || first.Uploads[0].UploadID != firstID || !first.IsTruncated { + t.Fatalf("unexpected first page: %+v", first) + } + rec := request(http.MethodDelete, getAbortMultipartUploadURL("", bucket, "a", firstID)) + if rec.Code != http.StatusNoContent { + t.Fatalf("abort HTTP %d: %s", rec.Code, rec.Body.String()) + } + rest := list(first.NextKeyMarker, first.NextUploadIDMarker, 10) + var keys []string + for _, upload := range rest.Uploads { + keys = append(keys, upload.Key) + } + t.Logf("GET first page=200; DELETE marker=204; GET next page=200 keys=%v truncated=%v", keys, rest.IsTruncated) + if _, err := z.GetMultipartInfo(t.Context(), bucket, "a", secondID, ObjectOptions{}); err != nil { + t.Fatalf("remaining upload is not valid: %v", err) + } + if len(rest.Uploads) != 2 || rest.Uploads[0].UploadID != secondID { + t.Fatalf("valid remaining upload for key a is missing from continuation: %+v", rest.Uploads) + } +} + +type multipartListingFaultDisk struct { + StorageAPI + read func(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) + delete func(context.Context, string, string, DeleteOptions) error +} + +func (d multipartListingFaultDisk) ReadVersion(ctx context.Context, original, volume, object, version string, opts ReadOptions) (FileInfo, error) { + if d.read != nil { + return d.read(ctx, original, volume, object, version, opts) + } + return d.StorageAPI.ReadVersion(ctx, original, volume, object, version, opts) +} + +func (d multipartListingFaultDisk) Delete(ctx context.Context, volume, object string, opts DeleteOptions) error { + if d.delete != nil { + return d.delete(ctx, volume, object, opts) + } + return d.StorageAPI.Delete(ctx, volume, object, opts) +} + +func TestMultipartListingLegacyPreflight(t *testing.T) { + z, set, bucket := multipartListingFixture(t) + const other = "multipart-legacy-other" + if err := z.MakeBucket(t.Context(), other, MakeBucketOptions{}); err != nil { + t.Fatal(err) + } + old, err := z.NewMultipartUpload(t.Context(), other, "old", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + fi, metadata, err := set.checkUploadIDExists(t.Context(), other, "old", old.UploadID, true) + if err != nil { + t.Fatal(err) + } + for i := range metadata { + delete(metadata[i].Metadata, multipartMetaBucket) + delete(metadata[i].Metadata, multipartMetaObject) + } + if _, err = writeAllMetadata(t.Context(), set.getDisks(), other, minioMetaMultipartBucket, + set.getUploadIDDir(other, "old", old.UploadID), metadata, fi.WriteQuorum(set.defaultWQuorum())); err != nil { + t.Fatal(err) + } + if _, err = z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil { + t.Fatal(err) + } + z.mpCache.Clear() + _, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10) + if !errors.Is(err, errMultipartListingLegacy) { + t.Fatalf("old upload in another bucket: %v", err) + } + report, err := z.multipartPreflight(t.Context()) + if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 { + t.Fatalf("legacy preflight: %+v %v", report, err) + } + if err = z.AbortMultipartUpload(t.Context(), other, "old", old.UploadID, ObjectOptions{}); err != nil { + t.Fatal(err) + } + report, err = z.multipartPreflight(t.Context()) + if err != nil || !report.Ready || report.LegacyUploads != 0 { + t.Fatalf("drained preflight: %+v %v", report, err) + } + got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10) + if err != nil { + t.Fatal(err) + } + requireMultipartUploadKeys(t, got, "a") +} + +func TestMultipartListingIdentityFallback(t *testing.T) { + for _, kind := range []string{"old", "corrupt", "read-failure", "wrong-bucket"} { + t.Run(kind, func(t *testing.T) { + z, set, bucket := multipartListingFixture(t) + if _, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil { + t.Fatal(err) + } + original := set.getDisks + t.Cleanup(func() { set.getDisks = original }) + var first atomic.Bool + set.getDisks = func() []StorageAPI { + disks := append([]StorageAPI(nil), original()...) + for i, d := range disks { + disks[i] = multipartListingFaultDisk{StorageAPI: d, read: func(ctx context.Context, b, v, p, version string, opts ReadOptions) (FileInfo, error) { + fi, err := d.ReadVersion(ctx, b, v, p, version, opts) + if v == minioMetaMultipartBucket && first.CompareAndSwap(false, true) { + if kind == "read-failure" { + return FileInfo{}, errDiskNotFound + } + if kind == "corrupt" { + return FileInfo{}, errFileCorrupt + } + fi.Metadata = cloneMSS(fi.Metadata) + if kind == "old" { + delete(fi.Metadata, multipartMetaBucket) + } else { + fi.Metadata[multipartMetaBucket] = "another-bucket" + } + } + return fi, err + }} + } + return disks + } + got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10) + if err != nil { + t.Fatal(err) + } + requireMultipartUploadKeys(t, got, "a") + }) + } +} + +func TestMultipartAbortPoolsAndRetry(t *testing.T) { + z, bucket := consistencyPools(t) + mp, err := z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + emptySet := z.serverPools[0].getHashedSet("a") + original := emptySet.getDisks + t.Cleanup(func() { emptySet.getDisks = original }) + emptySet.getDisks = func() []StorageAPI { + disks := append([]StorageAPI(nil), original()...) + for i, d := range disks { + disks[i] = multipartListingFaultDisk{StorageAPI: d, read: func(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) { + return FileInfo{}, errDiskNotFound + }} + } + return disks + } + err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}) + if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 { + t.Fatalf("unknown pool must not acknowledge cancellation: %v", err) + } + emptySet.getDisks = original + err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}) + if _, ok := err.(InvalidUploadID); !ok { + t.Fatalf("retry after all pools confirm absence: %v", err) + } + mp, err = z.serverPools[1].NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatal(err) + } + if _, err = z.serverPools[1].GetMultipartInfo(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err == nil { + t.Fatal("empty first pool hid the actual upload") + } +} + +func TestMultipartAbortRetryBelowReadQuorum(t *testing.T) { + z, set, bucket := multipartListingFixture(t) + mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + original := set.getDisks + t.Cleanup(func() { set.getDisks = original }) + set.getDisks = func() []StorageAPI { + disks := append([]StorageAPI(nil), original()...) + disks[3] = nil + d := disks[2] + disks[2] = multipartListingFaultDisk{StorageAPI: d, delete: func(ctx context.Context, v, p string, opts DeleteOptions) error { + if err := d.Delete(ctx, v, p, opts); err != nil { + return err + } + return context.DeadlineExceeded // operation finished, acknowledgement lost + }} + return disks + } + if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err == nil { + t.Fatal("lost acknowledgement must fail") + } + set.getDisks = original + if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatalf("one remaining metadata copy prevented retry: %v", err) + } +} + +func TestMultipartListingMarkerHTTP(t *testing.T) { + z, _ := consistencyPools(t) + bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{}) + if err != nil { + t.Fatal(err) + } + for _, tc := range []struct { + key, marker string + status int + }{ + {"", "not-base64=", 200}, + {"a", "not-base64=", 404}, + {"a", base64.RawURLEncoding.EncodeToString([]byte("not-native")), 400}, + {"a", multipartListingTestID(time.Unix(100, 0), 1), 200}, + } { + u := getListMultipartUploadsURLWithParams("", bucket, "", tc.key, tc.marker, "", "10") + req, err := newTestSignedRequestV4(http.MethodGet, u, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if rec.Code != tc.status { + t.Fatalf("key=%q marker=%q: %d %s", tc.key, tc.marker, rec.Code, rec.Body.String()) + } + } +} + +func TestMultipartPreflightAdminHTTP(t *testing.T) { + bed, err := prepareAdminErasureTestBed(t.Context()) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { bed.done(); bed.objLayer.Shutdown(context.Background()); removeRoots(bed.erasureDirs) }) + const target = "/minio/admin/v3/multipart-preflight" + rec := httptest.NewRecorder() + bed.router.ServeHTTP(rec, httptest.NewRequest(http.MethodGet, target, nil)) + if rec.Code != 403 { + t.Fatalf("anonymous preflight: %d", rec.Code) + } + req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + rec = httptest.NewRecorder() + bed.router.ServeHTTP(rec, req) + if rec.Code != 200 { + t.Fatalf("admin preflight: %d %s", rec.Code, rec.Body.String()) + } + var report multipartPreflightReport + if err = json.Unmarshal(rec.Body.Bytes(), &report); err != nil || !report.Ready || !report.Complete { + t.Fatalf("preflight: %+v %v", report, err) + } +} + +type multipartLateCreateDisk struct { + StorageAPI + late chan<- func() error +} + +func (d multipartLateCreateDisk) WriteMetadata(_ context.Context, original, volume, object string, fi FileInfo) error { + if volume == minioMetaMultipartBucket { + d.late <- func() error { return d.StorageAPI.WriteMetadata(context.Background(), original, volume, object, fi) } + return context.DeadlineExceeded + } + return d.StorageAPI.WriteMetadata(context.Background(), original, volume, object, fi) +} + +// This characterization is deliberately NOT an assertion of terminal abort +// correctness. It preserves an executable example of the separately scoped +// late-creation-write limitation. Change the expectation when fencing is added. +func TestMultipartAbortLateCreateBoundary(t *testing.T) { + obj, dirs, err := prepareErasure(t.Context(), 16) + if err != nil { + t.Fatal(err) + } + z := obj.(*erasureServerPools) + t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) }) + saved := globalStorageClass + globalStorageClass.Update(storageclass.Config{Standard: storageclass.StorageClass{Parity: 8}}) + t.Cleanup(func() { globalStorageClass.Update(saved) }) + const bucket = "multipart-late-create" + if err = z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil { + t.Fatal(err) + } + set := z.serverPools[0].getHashedSet("a") + original := set.getDisks + t.Cleanup(func() { set.getDisks = original }) + late := make(chan func() error, 16) + set.getDisks = func() []StorageAPI { + disks := append([]StorageAPI(nil), original()...) + for i := 9; i < 16; i++ { + disks[i] = multipartLateCreateDisk{disks[i], late} + } + return disks + } + mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + if len(late) != 7 { + t.Fatalf("expected seven timed-out creation writes, got %d", len(late)) + } + set.getDisks = func() []StorageAPI { + disks := append([]StorageAPI(nil), original()...) + for i := 2; i < 9; i++ { + disks[i] = nil + } + return disks + } + if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatal(err) + } + for range 7 { + if err = (<-late)(); err != nil { + t.Fatal(err) + } + } + set.getDisks = original + _, metadata, err := set.checkUploadIDExists(t.Context(), bucket, "a", mp.UploadID, true) + if err != nil { + t.Fatalf("late creation boundary changed; review and remove the documented limitation: %v", err) + } + count := 0 + for _, fi := range metadata { + if fi.IsValid() { + count++ + } + } + if count != 14 { + t.Fatalf("expected fourteen resurrected copies, got %d", count) + } + t.Log("KNOWN UNRESOLVED BOUNDARY: seven delayed creation writes plus seven offline copies restore a writable upload after acknowledged cancellation") +} + +type multipartListingCountingDisk struct { + StorageAPI + directoryCalls *atomic.Int64 + metadataCalls *atomic.Int64 +} + +func (d multipartListingCountingDisk) ListDir(ctx context.Context, original, volume, dir string, count int) ([]string, error) { + if volume == minioMetaMultipartBucket { + d.directoryCalls.Add(1) + } + return d.StorageAPI.ListDir(ctx, original, volume, dir, count) +} + +func (d multipartListingCountingDisk) ReadVersion(ctx context.Context, original, volume, object, version string, opts ReadOptions) (FileInfo, error) { + if volume == minioMetaMultipartBucket { + d.metadataCalls.Add(1) + } + return d.StorageAPI.ReadVersion(ctx, original, volume, object, version, opts) +} + +// Counts storage API calls; this is not a deployment throughput benchmark. +func TestMultipartListingScanCosts(t *testing.T) { + obj, dirs, err := prepareErasure(t.Context(), 4) + if err != nil { + t.Fatal(err) + } + z := obj.(*erasureServerPools) + t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) }) + const bucket, otherBucket = "r9-scan-target", "r9-scan-unrelated" + for _, name := range []string{bucket, otherBucket} { + if err := z.MakeBucket(t.Context(), name, MakeBucketOptions{}); err != nil { + t.Fatal(err) + } + } + if _, err := z.NewMultipartUpload(t.Context(), bucket, "dir/target", ObjectOptions{}); err != nil { + t.Fatal(err) + } + var directoryCalls, metadataCalls atomic.Int64 + for _, pool := range z.serverPools { + for _, set := range pool.sets { + original := set.getDisks + set.getDisks = func() []StorageAPI { + disks := original() + wrapped := make([]StorageAPI, len(disks)) + for i, disk := range disks { + if disk != nil { + wrapped[i] = multipartListingCountingDisk{disk, &directoryCalls, &metadataCalls} + } + } + return wrapped + } + t.Cleanup(func() { set.getDisks = original }) + } + } + measure := func(label string) (int64, int64) { + t.Helper() + directoryCalls.Store(0) + metadataCalls.Store(0) + result, err := z.ListMultipartUploads(t.Context(), bucket, "dir/", "", "", "", 1) + if err != nil { + t.Fatal(err) + } + requireMultipartUploadKeys(t, result, "dir/target") + d, m := directoryCalls.Load(), metadataCalls.Load() + t.Logf("%s: max-uploads=1 returned=%d ListDir=%d ReadVersion=%d", label, len(result.Uploads), d, m) + return d, m + } + _, baseline := measure("no unrelated uploads") + for n := 0; n < 32; n++ { + if _, err := z.NewMultipartUpload(t.Context(), otherBucket, fmt.Sprintf("other/%03d", n), ObjectOptions{}); err != nil { + t.Fatal(err) + } + } + _, loaded := measure("32 uploads in another bucket") + _, repeated := measure("same query repeated") + if loaded != baseline+32 || repeated != loaded { + t.Fatalf("unexpected scan accounting: baseline=%d loaded=%d repeated=%d", baseline, loaded, repeated) + } +} + +func multipartListingTestID(created time.Time, n int) string { + return base64.RawURLEncoding.EncodeToString([]byte(fmt.Sprintf("00000000-0000-4000-8000-000000000000.00000000-0000-4000-8000-%012xx%d", n, created.UnixNano()))) +} + +func multipartListingFixture(t *testing.T) (*erasureServerPools, *erasureObjects, string) { + t.Helper() + obj, dirs, err := prepareErasure(t.Context(), 4) + if err != nil { + t.Fatal(err) + } + z := obj.(*erasureServerPools) + t.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) }) + const bucket = "multipart-listing-test" + if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil { + t.Fatal(err) + } + return z, z.serverPools[0].getHashedSet("a"), bucket +} + +func TestMultipartListingMissingMarkers(t *testing.T) { + base := time.Unix(100, 0) + sameTime := []string{multipartListingTestID(base.Add(time.Second), 1), multipartListingTestID(base.Add(time.Second), 2), multipartListingTestID(base.Add(time.Second), 3)} + slices.Sort(sameTime) + uploads := []MultipartInfo{ + {Bucket: "bucket", Object: "a", UploadID: multipartListingTestID(base, 1), Initiated: base}, + {Bucket: "bucket", Object: "a", UploadID: sameTime[1], Initiated: base.Add(time.Second)}, + {Bucket: "bucket", Object: "b", UploadID: multipartListingTestID(base, 4), Initiated: base}, + } + for _, tc := range []struct { + name string + created time.Time + id int + want []string + }{ + {"before", base.Add(-time.Second), 0, []string{"a", "a", "b"}}, + {"between", base.Add(time.Second / 2), 2, []string{"a", "b"}}, + {"same-time-before", base.Add(time.Second), 2, []string{"a", "b"}}, + {"same-time-after", base.Add(time.Second), 4, []string{"b"}}, + {"after", base.Add(2 * time.Second), 5, []string{"b"}}, + } { + t.Run(tc.name, func(t *testing.T) { + marker := multipartListingTestID(tc.created, tc.id) + if tc.name == "same-time-before" { + marker = sameTime[0] + } + if tc.name == "same-time-after" { + marker = sameTime[2] + } + if err := checkListMultipartArgs(t.Context(), "bucket", "", "a", marker, ""); err != nil { + t.Fatal(err) + } + got := paginateMultipartUploads(uploads, "", "a", marker, "", 10) + if !slices.Equal(multipartUploadKeys(got.Uploads), tc.want) { + t.Fatalf("got %v want %v", multipartUploadKeys(got.Uploads), tc.want) + } + }) + } +} + +func TestMultipartListingAbortRecovery(t *testing.T) { + for _, offline := range []int{1, 2} { + t.Run(fmt.Sprint(offline), func(t *testing.T) { + z, set, bucket := multipartListingFixture(t) + mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + original := set.getDisks + t.Cleanup(func() { set.getDisks = original }) + set.getDisks = func() []StorageAPI { + disks := append([]StorageAPI(nil), original()...) + for i := 0; i < offline; i++ { + disks[i] = nil + } + return disks + } + err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}) + if offline == 1 && err != nil { + t.Fatal(err) + } + if offline == 2 && (err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503) { + t.Fatalf("two offline drives must not acknowledge cancellation: %v", err) + } + set.getDisks = original + if offline == 2 { + // Retry a partially completed deletion after the original disks return. + if err = z.AbortMultipartUpload(t.Context(), bucket, "a", mp.UploadID, ObjectOptions{}); err != nil { + t.Fatal(err) + } + } + got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100) + if err != nil || len(got.Uploads) != 0 { + t.Fatalf("canceled upload reappeared: %+v %v", got, err) + } + }) + } +} + +func TestMultipartListingCoverage(t *testing.T) { + z, set, bucket := multipartListingFixture(t) + mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + original := set.getDisks + t.Cleanup(func() { set.getDisks = original }) + disks := original() + // Leave a readable two-copy record, simulating a partially failed abort. + for _, d := range disks[:2] { + if err = d.Delete(t.Context(), minioMetaMultipartBucket, set.getUploadIDDir(bucket, "a", mp.UploadID), DeleteOptions{Recursive: true}); err != nil { + t.Fatal(err) + } + } + set.getDisks = func() []StorageAPI { return []StorageAPI{disks[0], disks[1], nil, nil} } + _, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10) + if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 { + t.Fatalf("incomplete discovery returned success: %v", err) + } + set.getDisks = func() []StorageAPI { return []StorageAPI{nil, disks[1], disks[2], disks[3]} } + got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10) + if err != nil { + t.Fatal(err) + } + requireMultipartUploadKeys(t, got, "a") + report, err := z.multipartPreflight(t.Context()) + if err != nil || report.Ready || report.Complete || len(report.Sets[0].UncoveredDrives) != 1 { + t.Fatalf("offline upgrade preflight: %+v %v", report, err) + } +} + +func TestMultipartListingBudgetAndAdmission(t *testing.T) { + _, set, bucket := multipartListingFixture(t) + if _, err := set.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil { + t.Fatal(err) + } + scan, err := startMultipartScan(t.Context(), false) + if err != nil { + t.Fatal(err) + } + defer scan.close() + scan.remaining.Store(1) + _, _, err = set.scanMultipartUploads(scan, bucket, 0, 0) + var limited SlowDown + if !errors.As(err, &limited) { + t.Fatalf("budget: %v", err) + } + second, err := startMultipartScan(t.Context(), false) + if err != nil { + t.Fatal(err) + } + defer second.close() + if third, err := startMultipartScan(t.Context(), false); err == nil { + third.close() + t.Fatal("third scan admitted") + } +} + +func TestMultipartListingPreflightMinorityLegacy(t *testing.T) { + z, set, bucket := multipartListingFixture(t) + mp, err := z.NewMultipartUpload(t.Context(), bucket, "old", ObjectOptions{}) + if err != nil { + t.Fatal(err) + } + path := set.getUploadIDDir(bucket, "old", mp.UploadID) + disks := set.getDisks() + fi, err := disks[0].ReadVersion(t.Context(), bucket, minioMetaMultipartBucket, path, "", ReadOptions{}) + if err != nil { + t.Fatal(err) + } + delete(fi.Metadata, multipartMetaBucket) + delete(fi.Metadata, multipartMetaObject) + if err = disks[0].WriteMetadata(t.Context(), bucket, minioMetaMultipartBucket, path, fi); err != nil { + t.Fatal(err) + } + for _, d := range disks[1:] { + if err = d.Delete(t.Context(), minioMetaMultipartBucket, path, DeleteOptions{Recursive: true}); err != nil { + t.Fatal(err) + } + } + report, err := z.multipartPreflight(t.Context()) + if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 { + t.Fatalf("minority legacy copy must prevent readiness: %+v %v", report, err) + } +} + +func TestMultipartListingCancellationRetainsAdmission(t *testing.T) { + z, set, bucket := multipartListingFixture(t) + for i := range 40 { + if _, err := z.NewMultipartUpload(t.Context(), bucket, fmt.Sprintf("key-%02d", i), ObjectOptions{}); err != nil { + t.Fatal(err) + } + } + original := set.getDisks + t.Cleanup(func() { set.getDisks = original }) + entered := make(chan struct{}, 40) + release := make(chan struct{}) + var once sync.Once + defer once.Do(func() { close(release) }) + var reads atomic.Int32 + set.getDisks = func() []StorageAPI { + disks := append([]StorageAPI(nil), original()...) + for i, d := range disks { + disks[i] = multipartListingFaultDisk{StorageAPI: d, read: func(ctx context.Context, b, v, p, version string, opts ReadOptions) (FileInfo, error) { + if v != minioMetaMultipartBucket { + return d.ReadVersion(ctx, b, v, p, version, opts) + } + reads.Add(1) + entered <- struct{}{} + // Model an RPC which does not return immediately on cancellation. + <-release + return FileInfo{}, ctx.Err() + }} + } + return disks + } + second, err := startMultipartScan(t.Context(), false) + if err != nil { + t.Fatal(err) + } + defer second.close() + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + done := make(chan error, 1) + go func() { + _, err := z.ListMultipartUploads(ctx, bucket, "", "", "", "", 10) + done <- err + }() + for range 16 { + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("identity workers did not start") + } + } + cancel() + if extra, err := startMultipartScan(t.Context(), false); err == nil { + extra.close() + t.Fatal("canceled request released admission while RPCs were still running") + } + start := time.Now() + once.Do(func() { close(release) }) + select { + case err := <-done: + if err == nil { + t.Fatal("canceled scan returned a successful partial list") + } + case <-time.After(2 * time.Second): + t.Fatal("scan did not stop after blocked RPCs returned") + } + if got := reads.Load(); got != 16 { + t.Fatalf("scheduled more identity/metadata reads after cancellation: %d", got) + } + t.Logf("16 identity RPCs bounded; admission held until return; cancellation settled in %s", time.Since(start)) +} diff --git a/cmd/erasure-multipart.go b/cmd/erasure-multipart.go index 0a1e70718..bccd16001 100644 --- a/cmd/erasure-multipart.go +++ b/cmd/erasure-multipart.go @@ -31,6 +31,7 @@ import ( "sync" "time" + "github.com/google/uuid" "github.com/klauspost/readahead" "github.com/minio/minio-go/v7/pkg/set" "github.com/minio/minio/internal/config/storageclass" @@ -262,13 +263,8 @@ func (er erasureObjects) cleanupStaleUploadsOnDisk(ctx context.Context, disk Sto func multipartUploadInfo(bucket, object, uploadUUID string, fallback time.Time) MultipartInfo { initiated := fallback - if i := strings.LastIndexByte(uploadUUID, 'x'); i >= 0 { - if parsed, err := strconv.ParseInt(uploadUUID[i+1:], 10, 64); err == nil { - initiated = time.Unix(0, parsed) - } - } - if initiated.IsZero() { - initiated = UTCNow() + if parsed, ok := multipartUploadTime(uploadUUID); ok { + initiated = parsed } return MultipartInfo{ Bucket: bucket, @@ -278,152 +274,33 @@ func multipartUploadInfo(bucket, object, uploadUUID string, fallback time.Time) } } -func listMultipartUploadDirs(ctx context.Context, disk StorageAPI, bucket string) ([]string, error) { - hashDirs, err := disk.ListDir(ctx, bucket, minioMetaMultipartBucket, "", -1) - if errors.Is(err, errFileNotFound) { - return nil, nil +// multipartUploadTime is shared by stored records and continuation markers. +// The time is part of the immutable upload ID, so a removed marker still +// identifies the same ordering boundary. +func multipartUploadTime(uploadUUID string) (time.Time, bool) { + if len(uploadUUID) < 38 || uploadUUID[36] != 'x' { + return time.Time{}, false } - if err != nil { - return nil, err + if _, err := uuid.Parse(uploadUUID[:36]); err != nil { + return time.Time{}, false } - - var candidates []string - for _, hashDir := range hashDirs { - if !strings.HasSuffix(hashDir, SlashSeparator) { - continue - } - hashDir = strings.TrimSuffix(hashDir, SlashSeparator) - uploadDirs, err := disk.ListDir(ctx, bucket, minioMetaMultipartBucket, hashDir, -1) - if errors.Is(err, errFileNotFound) { - continue - } - if err != nil { - return nil, err - } - for _, uploadDir := range uploadDirs { - if strings.HasSuffix(uploadDir, SlashSeparator) { - candidates = append(candidates, pathJoin(hashDir, strings.TrimSuffix(uploadDir, SlashSeparator))) - } - } + ns, err := strconv.ParseInt(uploadUUID[37:], 10, 64) + if err != nil || ns <= 0 || strconv.FormatInt(ns, 10) != uploadUUID[37:] { + return time.Time{}, false } - return candidates, nil + return time.Unix(0, ns), true } -func (er erasureObjects) readMultipartUploadCandidate(ctx context.Context, bucket, candidate string) (MultipartInfo, bool, bool, error) { - shaDir, uploadUUID, ok := strings.Cut(candidate, SlashSeparator) - if !ok || shaDir == "" || uploadUUID == "" || strings.Contains(uploadUUID, SlashSeparator) { - return MultipartInfo{}, false, false, fmt.Errorf("invalid multipart upload path %q", candidate) - } - - disks := er.getDisks() - partsMetadata, errs := readAllFileInfo(ctx, disks, bucket, minioMetaMultipartBucket, candidate, "", false, false) - readQuorum, _, err := objectQuorumFromMeta(ctx, partsMetadata, errs, er.defaultParityCount) +func multipartMarkerTime(uploadID string) (time.Time, bool) { + b, err := base64.RawURLEncoding.DecodeString(uploadID) if err != nil { - return MultipartInfo{}, false, false, err + return time.Time{}, false } - _, modTime, etag := listOnlineDisks(disks, partsMetadata, errs, readQuorum) - if err = reduceReadQuorumErrs(ctx, errs, objectOpIgnoredErrs, readQuorum); err != nil { - return MultipartInfo{}, false, false, err + _, uploadUUID, ok := strings.Cut(string(b), ".") + if !ok { + return time.Time{}, false } - fi, err := pickValidFileInfo(ctx, partsMetadata, modTime, etag, readQuorum) - if err != nil { - return MultipartInfo{}, false, false, err - } - - storedBucket, hasBucket := fi.Metadata[multipartMetaBucket] - storedObject, hasObject := fi.Metadata[multipartMetaObject] - if !hasBucket || !hasObject || storedBucket == "" || storedObject == "" { - return MultipartInfo{}, false, true, nil - } - if storedBucket != bucket { - return MultipartInfo{}, false, false, nil - } - if !IsValidObjectPrefix(storedObject) || er.getMultipartSHADir(storedBucket, storedObject) != shaDir { - return MultipartInfo{}, false, false, fmt.Errorf("multipart upload %q has invalid stored identity", candidate) - } - return multipartUploadInfo(storedBucket, storedObject, uploadUUID, fi.ModTime), true, false, nil -} - -// scanMultipartUploads discovers durable multipart state from every online -// drive in this erasure set and accepts only quorum-valid upload metadata. -func (er erasureObjects) scanMultipartUploads(ctx context.Context, bucket string) ([]MultipartInfo, bool, error) { - disks := er.getOnlineDisks() - if len(disks) < (er.setDriveCount+1)/2 { - return nil, false, errErasureReadQuorum - } - - candidateSet := make(map[string]struct{}) - var candidateMu sync.Mutex - g := errgroup.WithNErrs(len(disks)) - for i := range disks { - g.Go(func() error { - candidates, err := listMultipartUploadDirs(ctx, disks[i], bucket) - if err != nil { - return err - } - candidateMu.Lock() - for _, candidate := range candidates { - candidateSet[candidate] = struct{}{} - } - candidateMu.Unlock() - return nil - }, i) - } - if err := reduceReadQuorumErrs(ctx, g.Wait(), nil, (er.setDriveCount+1)/2); err != nil { - return nil, false, err - } - - candidates := make([]string, 0, len(candidateSet)) - for candidate := range candidateSet { - candidates = append(candidates, candidate) - } - sort.Strings(candidates) - if len(candidates) == 0 { - return nil, false, nil - } - - jobs := make(chan string) - workerCount := min(multipartMetadataScanConcurrency, len(candidates)) - var wg sync.WaitGroup - var resultMu sync.Mutex - var uploads []MultipartInfo - var keyless bool - var firstErr error - for range workerCount { - wg.Add(1) - go func() { - defer wg.Done() - for candidate := range jobs { - resultMu.Lock() - stopped := firstErr != nil - resultMu.Unlock() - if stopped { - continue - } - - upload, found, legacy, err := er.readMultipartUploadCandidate(ctx, bucket, candidate) - if errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound) { - continue - } - resultMu.Lock() - switch { - case err != nil && firstErr == nil: - firstErr = err - case legacy: - keyless = true - case found: - uploads = append(uploads, upload) - } - resultMu.Unlock() - } - }() - } - for _, candidate := range candidates { - jobs <- candidate - } - close(jobs) - wg.Wait() - return uploads, keyless, firstErr + return multipartUploadTime(uploadUUID) } type multipartListEntry struct { @@ -465,7 +342,7 @@ func paginateMultipartUploads(uploads []MultipartInfo, prefix, keyMarker, upload return deduplicated[i].UploadID < deduplicated[j].UploadID }) - markerPassed := keyMarker == "" + markerTime, _ := multipartMarkerTime(uploadIDMarker) seenPrefixes := make(map[string]struct{}) entries := make([]multipartListEntry, 0, len(deduplicated)) for i := range deduplicated { @@ -473,17 +350,15 @@ func paginateMultipartUploads(uploads []MultipartInfo, prefix, keyMarker, upload if !strings.HasPrefix(upload.Object, prefix) { continue } - if !markerPassed { + if keyMarker != "" { switch strings.Compare(upload.Object, keyMarker) { case -1: continue case 0: - if uploadIDMarker != "" && upload.UploadID == uploadIDMarker { - markerPassed = true + if uploadIDMarker == "" || upload.Initiated.Before(markerTime) || + (upload.Initiated.Equal(markerTime) && upload.UploadID <= uploadIDMarker) { + continue } - continue - default: - markerPassed = true } } @@ -584,7 +459,16 @@ func (er erasureObjects) listMultipartUploadsExact(ctx context.Context, bucket, if populatedUploadIDs.Contains(uploadID) { continue } - uploads = append(uploads, multipartUploadInfo(bucket, object, uploadID, time.Time{})) + var fallback time.Time + if _, ok := multipartUploadTime(uploadID); !ok { + fi, err := disk.ReadVersion(ctx, bucket, minioMetaMultipartBucket, + pathJoin(er.getMultipartSHADir(bucket, object), uploadID), "", ReadOptions{}) + if err != nil { + return result, toObjectErr(err, bucket, object) + } + fallback = fi.ModTime + } + uploads = append(uploads, multipartUploadInfo(bucket, object, uploadID, fallback)) populatedUploadIDs.Add(uploadID) } @@ -629,10 +513,18 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, prefi if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil { return ListMultipartsInfo{}, err } - uploads, _, err := er.scanMultipartUploads(ctx, bucket) + scan, err := startMultipartScan(ctx, false) if err != nil { return ListMultipartsInfo{}, err } + defer scan.close() + uploads, legacy, err := er.scanMultipartUploads(scan, bucket, 0, 0) + if err != nil { + return ListMultipartsInfo{}, err + } + if legacy { + return ListMultipartsInfo{}, errMultipartListingLegacy + } return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil } @@ -1861,22 +1753,66 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str return fi.ToObjectInfo(bucket, object, opts.Versioned || opts.VersionSuspended), nil } -// AbortMultipartUpload - aborts an ongoing multipart operation -// signified by the input uploadID. This is an atomic operation -// doesn't require clients to initiate multiple such requests. -// -// All parts are purged from all disks and reference to the uploadID -// would be removed from the system, rollback is not possible on this -// operation. -func (er erasureObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (err error) { +// abortMultipartUpload confirms absence on a strict majority of this set. +// Unlike an existence read followed by best-effort deletion, it also permits +// retrying a partial deletion which no longer has a readable metadata quorum. +// This does not fence creation writes still executing after a storage timeout. +func (er erasureObjects) abortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (bool, error) { if !opts.NoAuditLog { auditObjectErasureSet(ctx, "AbortMultipartUpload", object, &er) } - - // Cleanup all uploaded parts. - defer er.deleteAll(ctx, minioMetaMultipartBucket, er.getUploadIDDir(bucket, object, uploadID)) - - // Validates if upload ID exists. - _, _, err = er.checkUploadIDExists(ctx, bucket, object, uploadID, false) - return toObjectErr(err, bucket, object, uploadID) + b, err := base64.RawURLEncoding.DecodeString(uploadID) + if err != nil { + return false, MalformedUploadID{UploadID: uploadID} + } + _, internalID, ok := strings.Cut(string(b), ".") + if !ok || internalID == "" || internalID == "." || internalID == ".." || strings.ContainsAny(internalID, "/\\") { + return false, InvalidUploadID{Bucket: bucket, Object: object, UploadID: uploadID} + } + disks := er.getDisks() + uploadPath := er.getUploadIDDir(bucket, object, uploadID) + _, errs := readAllFileInfo(ctx, disks, bucket, minioMetaMultipartBucket, uploadPath, "", false, false) + quorum := er.setDriveCount/2 + 1 + found, absent := false, 0 + for _, err := range errs { + switch { + case err == nil, errors.Is(err, errFileCorrupt): + found = true + case errors.Is(err, errFileNotFound), errors.Is(err, errFileVersionNotFound): + absent++ + } + } + if absent >= quorum { + return found, nil + } + if !found { + return false, toObjectErr(errErasureReadQuorum, bucket, object, uploadID) + } + g := errgroup.WithNErrs(len(disks)) + for i, disk := range disks { + g.Go(func() error { + if disk == nil { + return errDiskNotFound + } + err := disk.Delete(ctx, minioMetaMultipartBucket, uploadPath, DeleteOptions{Recursive: true, Immediate: false}) + if errors.Is(err, errFileNotFound) || errors.Is(err, errFileVersionNotFound) { + return nil + } + return err + }, i) + } + return true, toObjectErr(reduceWriteQuorumErrs(ctx, g.Wait(), nil, quorum), bucket, object, uploadID) +} + +// AbortMultipartUpload confirms logical cancellation. Offline part data may +// still need stale-upload cleanup after its drives return. +func (er erasureObjects) AbortMultipartUpload(ctx context.Context, bucket, object, uploadID string, opts ObjectOptions) (err error) { + found, err := er.abortMultipartUpload(ctx, bucket, object, uploadID, opts) + if err != nil { + return err + } + if !found { + return InvalidUploadID{Bucket: bucket, Object: object, UploadID: uploadID} + } + return nil } diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go index f18b5b906..bdc3bfedd 100644 --- a/cmd/erasure-server-pool.go +++ b/cmd/erasure-server-pool.go @@ -1894,6 +1894,14 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil { return ListMultipartsInfo{}, toObjectErr(err, bucket) } + if globalAPIConfig.getMultipartListingLegacy() { + return z.listMultipartUploadsLegacy(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads) + } + scan, err := startMultipartScan(ctx, false) + if err != nil { + return ListMultipartsInfo{}, err + } + defer scan.close() var uploads []MultipartInfo var keyless bool @@ -1901,7 +1909,7 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p if z.IsSuspended(idx) { continue } - poolUploads, poolKeyless, err := pool.scanMultipartUploads(ctx, bucket) + poolUploads, poolKeyless, err := pool.scanMultipartUploads(scan, bucket, idx) if err != nil { return ListMultipartsInfo{}, err } @@ -1909,11 +1917,10 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p keyless = keyless || poolKeyless } - // Old writers did not persist the bucket and object key. Until every such - // upload has drained, retain the old response behavior instead of silently - // claiming that a partial durable scan is complete. + // The old format cannot be enumerated authoritatively. Migration mode is + // explicit: another bucket must never silently change this API's semantics. if keyless { - return z.listMultipartUploadsLegacy(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads) + return ListMultipartsInfo{}, errMultipartListingLegacy } return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil } @@ -2149,9 +2156,13 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o if err := checkAbortMultipartArgs(ctx, bucket, object, uploadID); err != nil { return err } + if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil { + return toObjectErr(err, bucket) + } defer func() { - if err == nil { + _, absent := err.(InvalidUploadID) + if err == nil || absent { z.mpCache.Delete(uploadID) globalNotificationSys.DeleteUploadID(ctx, uploadID) } @@ -2165,23 +2176,23 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o ctx = lkctx.Context() defer lk.Unlock(lkctx) - if z.SinglePool() { - return z.serverPools[0].AbortMultipartUpload(ctx, bucket, object, uploadID, opts) - } - + found := false + var firstErr error for idx, pool := range z.serverPools { if z.IsSuspended(idx) { continue } - err := pool.AbortMultipartUpload(ctx, bucket, object, uploadID, opts) - if err == nil { - return nil + poolFound, err := pool.getHashedSet(object).abortMultipartUpload(ctx, bucket, object, uploadID, opts) + found = found || poolFound + if err != nil && firstErr == nil { + firstErr = err } - if _, ok := err.(InvalidUploadID); ok { - // upload id not found move to next pool - continue - } - return err + } + if firstErr != nil { + return firstErr + } + if found { + return nil } return InvalidUploadID{ Bucket: bucket, diff --git a/cmd/erasure-sets.go b/cmd/erasure-sets.go index 20b730451..f655804cd 100644 --- a/cmd/erasure-sets.go +++ b/cmd/erasure-sets.go @@ -883,18 +883,26 @@ func (s *erasureSets) ListMultipartUploads(ctx context.Context, bucket, prefix, if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil { return ListMultipartsInfo{}, err } - uploads, _, err := s.scanMultipartUploads(ctx, bucket) + scan, err := startMultipartScan(ctx, false) if err != nil { return ListMultipartsInfo{}, err } + defer scan.close() + uploads, legacy, err := s.scanMultipartUploads(scan, bucket, 0) + if err != nil { + return ListMultipartsInfo{}, err + } + if legacy { + return ListMultipartsInfo{}, errMultipartListingLegacy + } return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil } -func (s *erasureSets) scanMultipartUploads(ctx context.Context, bucket string) ([]MultipartInfo, bool, error) { +func (s *erasureSets) scanMultipartUploads(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, bool, error) { var uploads []MultipartInfo var keyless bool - for _, set := range s.sets { - setUploads, setKeyless, err := set.scanMultipartUploads(ctx, bucket) + for i, set := range s.sets { + setUploads, setKeyless, err := set.scanMultipartUploads(scan, bucket, poolIdx, i) if err != nil { return nil, false, err } diff --git a/cmd/handler-api.go b/cmd/handler-api.go index 09790a60a..acd8f97af 100644 --- a/cmd/handler-api.go +++ b/cmd/handler-api.go @@ -50,6 +50,7 @@ type apiConfig struct { transitionWorkers int staleUploadsExpiry time.Duration + multipartListingLegacy bool staleUploadsCleanupInterval time.Duration deleteCleanupInterval time.Duration enableODirect bool @@ -181,6 +182,7 @@ func (t *apiConfig) init(cfg api.Config, setDriveCounts []int, legacy bool) { t.transitionWorkers = cfg.TransitionWorkers t.staleUploadsExpiry = cfg.StaleUploadsExpiry + t.multipartListingLegacy = cfg.MultipartListing == "legacy" t.deleteCleanupInterval = cfg.DeleteCleanupInterval t.enableODirect = cfg.EnableODirect t.gzipObjects = cfg.GzipObjects @@ -206,6 +208,12 @@ func (t *apiConfig) odirectEnabled() bool { return t.enableODirect } +func (t *apiConfig) getMultipartListingLegacy() bool { + t.mu.RLock() + defer t.mu.RUnlock() + return t.multipartListingLegacy +} + func (t *apiConfig) shouldGzipObjects() bool { t.mu.RLock() defer t.mu.RUnlock() diff --git a/cmd/list-multipart-uploads-compat_test.go b/cmd/list-multipart-uploads-compat_test.go index db0f49d6b..49aab94b3 100644 --- a/cmd/list-multipart-uploads-compat_test.go +++ b/cmd/list-multipart-uploads-compat_test.go @@ -19,6 +19,7 @@ package cmd import ( "bytes" + "errors" "fmt" "slices" "testing" @@ -180,8 +181,8 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) { } // Simulate an upload written by a pre-upgrade server. Its key cannot be - // recovered by scanning the hashed namespace, so detection must retain the - // exact-key legacy path until such uploads have drained. + // recovered by scanning the hashed namespace. Strict listing must fail; + // the old path is available only through an explicit migration setting. legacyObject := objects[1] er := sets.getHashedSet(legacyObject) fi, metadata, err := er.checkUploadIDExists(t.Context(), bucket, legacyObject, uploadIDs[legacyObject], true) @@ -196,6 +197,19 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) { er.getUploadIDDir(bucket, legacyObject, uploadIDs[legacyObject]), metadata, fi.WriteQuorum(er.defaultWQuorum())); err != nil { t.Fatal(err) } + _, err = z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100) + if !errors.Is(err, errMultipartListingLegacy) { + t.Fatalf("legacy strict listing: %v", err) + } + globalAPIConfig.mu.Lock() + oldLegacy := globalAPIConfig.multipartListingLegacy + globalAPIConfig.multipartListingLegacy = true + globalAPIConfig.mu.Unlock() + t.Cleanup(func() { + globalAPIConfig.mu.Lock() + globalAPIConfig.multipartListingLegacy = oldLegacy + globalAPIConfig.mu.Unlock() + }) legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100) if err != nil { t.Fatal(err) @@ -205,24 +219,25 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) { func TestPaginateMultipartUploads(t *testing.T) { base := time.Unix(100, 0) + id1, id2, id3 := multipartListingTestID(base, 1), multipartListingTestID(base.Add(time.Second), 2), multipartListingTestID(base, 3) uploads := []MultipartInfo{ - {Bucket: "bucket", Object: "b", UploadID: "b1", Initiated: base}, - {Bucket: "bucket", Object: "a", UploadID: "a2", Initiated: base.Add(time.Second)}, - {Bucket: "bucket", Object: "a", UploadID: "a1", Initiated: base}, - {Bucket: "bucket", Object: "a", UploadID: "a1", Initiated: base}, // duplicate discovery + {Bucket: "bucket", Object: "b", UploadID: id3, Initiated: base}, + {Bucket: "bucket", Object: "a", UploadID: id2, Initiated: base.Add(time.Second)}, + {Bucket: "bucket", Object: "a", UploadID: id1, Initiated: base}, + {Bucket: "bucket", Object: "a", UploadID: id1, Initiated: base}, // duplicate discovery } first := paginateMultipartUploads(uploads, "", "", "", "", 1) requireMultipartUploadKeys(t, first, "a") - if !first.IsTruncated || first.NextKeyMarker != "a" || first.NextUploadIDMarker != "a1" { + if !first.IsTruncated || first.NextKeyMarker != "a" || first.NextUploadIDMarker != id1 { t.Fatalf("first page = %+v", first) } second := paginateMultipartUploads(uploads, "", first.NextKeyMarker, first.NextUploadIDMarker, "", 1) - if len(second.Uploads) != 1 || second.Uploads[0].Object != "a" || second.Uploads[0].UploadID != "a2" { + if len(second.Uploads) != 1 || second.Uploads[0].Object != "a" || second.Uploads[0].UploadID != id2 { t.Fatalf("second page uploads = %+v", second.Uploads) } - if !second.IsTruncated || second.NextKeyMarker != "a" || second.NextUploadIDMarker != "a2" { + if !second.IsTruncated || second.NextKeyMarker != "a" || second.NextUploadIDMarker != id2 { t.Fatalf("second page = %+v", second) } @@ -232,7 +247,7 @@ func TestPaginateMultipartUploads(t *testing.T) { t.Fatalf("last page = %+v", last) } - missingUploadMarker := paginateMultipartUploads(uploads, "", "a", "missing", "", 10) + missingUploadMarker := paginateMultipartUploads(uploads, "", "a", multipartListingTestID(base.Add(2*time.Second), 4), "", 10) requireMultipartUploadKeys(t, missingUploadMarker, "b") if err := checkListMultipartArgs(t.Context(), "bucket", "", "", "not-base64=", ""); err != nil { diff --git a/cmd/object-api-input-checks.go b/cmd/object-api-input-checks.go index 7ffdd617b..27d1d033c 100644 --- a/cmd/object-api-input-checks.go +++ b/cmd/object-api-input-checks.go @@ -20,6 +20,7 @@ package cmd import ( "context" "encoding/base64" + "errors" "runtime" "strings" @@ -97,6 +98,9 @@ func checkListMultipartArgs(ctx context.Context, bucket, prefix, keyMarker, uplo UploadID: uploadIDMarker, } } + if _, ok := multipartMarkerTime(uploadIDMarker); !ok { + return InvalidArgument{Bucket: bucket, Object: keyMarker, Err: errors.New("upload-id-marker must contain a native multipart upload ID")} + } } return nil } diff --git a/cmd/storage-rest_test.go b/cmd/storage-rest_test.go index d38ad819b..6609cbfa3 100644 --- a/cmd/storage-rest_test.go +++ b/cmd/storage-rest_test.go @@ -109,6 +109,15 @@ func testStorageAPIListDir(t *testing.T, storage StorageAPI) { } } } + for _, name := range []string{"one", "two", "three"} { + if err := storage.AppendFile(t.Context(), "foo", "bounded/"+name, []byte("x")); err != nil { + t.Fatal(err) + } + } + entries, err := storage.ListDir(t.Context(), "", "foo", "bounded", 2) + if err != nil || len(entries) != 2 { + t.Fatalf("ListDir count was lost in storage/RPC path: %v %v", entries, err) + } } func testStorageAPIReadAll(t *testing.T, storage StorageAPI) { diff --git a/internal/config/api/api.go b/internal/config/api/api.go index e6b6a1eb5..26d479873 100644 --- a/internal/config/api/api.go +++ b/internal/config/api/api.go @@ -45,6 +45,7 @@ const ( apiTransitionWorkers = "transition_workers" apiStaleUploadsCleanupInterval = "stale_uploads_cleanup_interval" apiStaleUploadsExpiry = "stale_uploads_expiry" + apiMultipartListing = "multipart_listing" apiDeleteCleanupInterval = "delete_cleanup_interval" apiDisableODirect = "disable_odirect" apiODirect = "odirect" @@ -67,6 +68,7 @@ const ( EnvAPIStaleUploadsCleanupInterval = "MINIO_API_STALE_UPLOADS_CLEANUP_INTERVAL" EnvAPIStaleUploadsExpiry = "MINIO_API_STALE_UPLOADS_EXPIRY" + EnvAPIMultipartListing = "MINIO_API_MULTIPART_LISTING" EnvAPIDeleteCleanupInterval = "MINIO_API_DELETE_CLEANUP_INTERVAL" EnvDeleteCleanupInterval = "MINIO_DELETE_CLEANUP_INTERVAL" EnvAPIODirect = "MINIO_API_ODIRECT" @@ -133,6 +135,7 @@ var ( Key: apiStaleUploadsExpiry, Value: "24h", }, + config.KV{Key: apiMultipartListing, Value: "strict"}, config.KV{ Key: apiDeleteCleanupInterval, Value: "5m", @@ -178,6 +181,7 @@ type Config struct { TransitionWorkers int `json:"transition_workers"` StaleUploadsCleanupInterval time.Duration `json:"stale_uploads_cleanup_interval"` StaleUploadsExpiry time.Duration `json:"stale_uploads_expiry"` + MultipartListing string `json:"multipart_listing"` DeleteCleanupInterval time.Duration `json:"delete_cleanup_interval"` EnableODirect bool `json:"enable_odirect"` GzipObjects bool `json:"gzip_objects"` @@ -320,6 +324,10 @@ func LookupConfig(kvs config.KVS) (cfg Config, err error) { return cfg, err } cfg.StaleUploadsExpiry = staleUploadsExpiry + cfg.MultipartListing = env.Get(EnvAPIMultipartListing, kvs.GetWithDefault(apiMultipartListing, DefaultKVS)) + if cfg.MultipartListing != "strict" && cfg.MultipartListing != "legacy" { + return cfg, fmt.Errorf("%s must be strict or legacy", apiMultipartListing) + } cfg.SyncEvents = env.Get(EnvAPISyncEvents, kvs.Get(apiSyncEvents)) == config.EnableOn diff --git a/internal/config/api/api_test.go b/internal/config/api/api_test.go new file mode 100644 index 000000000..0c659f65b --- /dev/null +++ b/internal/config/api/api_test.go @@ -0,0 +1,41 @@ +// Copyright (c) 2026 Ruohang Feng +// SPDX-License-Identifier: AGPL-3.0-or-later + +package api + +import ( + "testing" + + "github.com/minio/minio/internal/config" +) + +func TestMultipartListingMigrationMode(t *testing.T) { + for _, tc := range []struct { + name, stored, override, want string + invalid bool + }{ + {name: "default", want: "strict"}, + {name: "explicit-migration", stored: "legacy", want: "legacy"}, + {name: "environment-override", stored: "strict", override: "legacy", want: "legacy"}, + {name: "invalid-stored", stored: "automatic", invalid: true}, + {name: "invalid-environment", stored: "strict", override: "automatic", invalid: true}, + } { + t.Run(tc.name, func(t *testing.T) { + t.Setenv(EnvAPIMultipartListing, tc.override) + var kvs config.KVS + if tc.stored != "" { + kvs = config.KVS{{Key: apiMultipartListing, Value: tc.stored}} + } + cfg, err := LookupConfig(kvs) + if tc.invalid { + if err == nil { + t.Fatal("invalid migration mode silently accepted") + } + return + } + if err != nil || cfg.MultipartListing != tc.want { + t.Fatalf("mode=%q err=%v, want %q", cfg.MultipartListing, err, tc.want) + } + }) + } +} diff --git a/internal/config/api/help.go b/internal/config/api/help.go index 2c4b8b2bb..650a4064d 100644 --- a/internal/config/api/help.go +++ b/internal/config/api/help.go @@ -26,6 +26,12 @@ var ( // Help holds configuration keys and their default values for api subsystem. Help = config.HelpKVS{ + config.HelpKV{ + Key: apiMultipartListing, + Description: "multipart listing mode: strict, or temporary legacy mode during coordinated upgrade" + defaultHelpPostfix(apiMultipartListing), + Type: "string", + Optional: true, + }, config.HelpKV{ Key: apiRequestsMax, Description: `set the maximum number of concurrent requests (default: auto)`,