mirror of
https://github.com/pgsty/minio.git
synced 2026-10-18 07:26:10 +03:00
Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 7007c164a5 | |||
| 51b7de6a0d | |||
| 4733e1423c | |||
| 4189869e43 | |||
| bb60871f57 | |||
| 2c820895d7 | |||
| 2080b3d031 | |||
| 9954392605 | |||
| 3dd53eb81f |
@@ -93,8 +93,10 @@ jobs:
|
|||||||
- name: Run conditional PUT tests under race detector
|
- name: Run conditional PUT tests under race detector
|
||||||
run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP)' -count=1 -timeout=5m
|
run: go test -race ./cmd -run '^Test(PoolsConditionalPut|SinglePoolConditionalPutHTTP)' -count=1 -timeout=5m
|
||||||
|
|
||||||
- name: Run multipart listing and cancellation tests under race detector
|
- name: Run multipart listing, cancellation and cleanup tests under race detector
|
||||||
run: go test -race ./cmd -run '^Test(MultipartListing|MultipartAbort|PaginateMultipartUploads|ListMultipartUploads)' -count=1 -timeout=5m
|
env:
|
||||||
|
MINIO_API_REQUESTS_MAX: "10000"
|
||||||
|
run: go test -race ./cmd -run '^Test(Multipart|PaginateMultipartUploads|ListMultipartUploads|CleanupStaleUploads|ExtractModTime)' -count=1 -timeout=5m
|
||||||
|
|
||||||
- name: Run listing metadata recovery, pagination and cancellation tests under race detector
|
- name: Run listing metadata recovery, pagination and cancellation tests under race detector
|
||||||
run: go test -race ./cmd -run '^Test(ListFallback|ListingCache|ListingMerge|ListingRecovery|ListingPending|ListingEntry|ListingSecondLock|StreamMetadataParts|LocalLockerListing|DistributedListing|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity)' -count=1 -timeout=5m
|
run: go test -race ./cmd -run '^Test(ListFallback|ListingCache|ListingMerge|ListingRecovery|ListingPending|ListingEntry|ListingSecondLock|StreamMetadataParts|LocalLockerListing|DistributedListing|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity)' -count=1 -timeout=5m
|
||||||
|
|||||||
+14
-14
@@ -11,9 +11,12 @@ See the [component matrix](https://silo.pgsty.com/compatibility/versions/) and
|
|||||||
|
|
||||||
### October 18 dependency update
|
### October 18 dependency update
|
||||||
|
|
||||||
- Select silo-pkg v3.14.2, upstream minio-go v7.4.1, published mcli 20261018
|
- Select silo-pkg v3.14.3, upstream minio-go v7.4.1, published mcli 20261018
|
||||||
and Console v2.4.2 source (`ef2dd2c41d21`). The SDK includes cancellation cleanup, UTC retention
|
and Console v2.4.2 source (`ef2dd2c41d21`). The SDK includes cancellation cleanup, UTC retention
|
||||||
timestamps and preserved STS duration; parallel GET/Compose remain opt-in.
|
timestamps and preserved STS duration; parallel GET/Compose remain opt-in.
|
||||||
|
- Adopt silo-pkg's faster admin-action classification during policy evaluation,
|
||||||
|
preserving authorization behavior and compatibility with existing clients.
|
||||||
|
Update the standalone `reorder-disks` debugging module to the same package version.
|
||||||
- Build with Go 1.27.2 and refresh Go x/* modules. Apply go-jose v4.1.6,
|
- Build with Go 1.27.2 and refresh Go x/* modules. Apply go-jose v4.1.6,
|
||||||
LDAP v3.4.15, compress v1.20.3, msgp v1.6.5, etcd v3.7.2, fwd v1.2.1 and
|
LDAP v3.4.15, compress v1.20.3, msgp v1.6.5, etcd v3.7.2, fwd v1.2.1 and
|
||||||
cpuid v2.4.1 bug/security fixes. OIDC JWK RSA keys must not exceed 4096 bits.
|
cpuid v2.4.1 bug/security fixes. OIDC JWK RSA keys must not exceed 4096 bits.
|
||||||
@@ -73,19 +76,16 @@ candidate's exact delta.
|
|||||||
|
|
||||||
### Object storage and replication
|
### Object storage and replication
|
||||||
|
|
||||||
- Make durable `ListMultipartUploads` scanning the only listing path (#79):
|
- Withdraw default persistent `ListMultipartUploads` scanning (#79). Restore
|
||||||
apply prefix, delimiter, markers, ordering and the 1,000-entry page limit
|
node-local cache listing for an empty prefix and exact-key lookup otherwise;
|
||||||
globally across pools and sets, independently of node-local caches. Omit
|
restart omissions, true-prefix and cache-pagination limitations remain.
|
||||||
readable historical uploads only when both bucket/key identity fields are
|
Preserve bucket/key identity metadata, path safety and stale-upload cleanup.
|
||||||
absent; malformed identities, uncertain coverage and resource limits remain
|
`MINIO_API_MULTIPART_LISTING` stays retired and ignored with a startup warning;
|
||||||
errors. Old uploads still support part upload, completion and cancellation
|
obsolete shared `api multipart_listing` keys remain tolerated. **Existing
|
||||||
with a known bucket/key/upload ID. No backfill or new storage layout is added.
|
opt-in strict users lose durable listing and strict cancellation.** The
|
||||||
`MINIO_API_MULTIPART_LISTING` is retired and ignored with a startup warning;
|
read-only admin `multipart-preflight` reports `mode=legacy`; its scan budgets
|
||||||
historical shared `api multipart_listing` keys remain tolerated. The read-only
|
and readiness do not gate public listing or startup. Durable, complete,
|
||||||
admin `multipart-preflight` reports `mode=persistent`; `ready` diagnoses the
|
scalable enumeration remains deferred. See the
|
||||||
absence of old records and no longer gates listing. Each page still scans
|
|
||||||
durable state within the existing 100,000-directory-entry, 30-second and
|
|
||||||
per-process concurrency budgets. See the
|
|
||||||
[design and limits](https://silo.pgsty.com/blog/design/list-multipart-uploads/#implementation).
|
[design and limits](https://silo.pgsty.com/blog/design/list-multipart-uploads/#implementation).
|
||||||
Thanks to mr javad seydi (@mrjavadseydi) for the original metadata-and-scan
|
Thanks to mr javad seydi (@mrjavadseydi) for the original metadata-and-scan
|
||||||
implementation (#198).
|
implementation (#198).
|
||||||
|
|||||||
@@ -11,6 +11,33 @@ fail() {
|
|||||||
exit 1
|
exit 1
|
||||||
}
|
}
|
||||||
|
|
||||||
|
# Git is already a build prerequisite. --no-index also scans untracked files
|
||||||
|
# and exported source trees. As before, honor .gitignore only in a worktree.
|
||||||
|
search_ignore=--no-exclude-standard
|
||||||
|
if git rev-parse --show-toplevel >/dev/null 2>&1; then
|
||||||
|
search_ignore=--exclude-standard
|
||||||
|
fi
|
||||||
|
search_errors="$(mktemp)"
|
||||||
|
trap 'rm -f "${search_errors}"' EXIT
|
||||||
|
|
||||||
|
search_text() {
|
||||||
|
local status=0
|
||||||
|
git --no-pager grep --no-index "${search_ignore}" --no-color -nIHE "$@" 2>"${search_errors}" || status=$?
|
||||||
|
# Git can return 1 for both no matches and a file read error. Check stderr
|
||||||
|
# as well, before callers filter matches through their allowlists.
|
||||||
|
if [ "${status}" -gt 1 ] || [ -s "${search_errors}" ]; then
|
||||||
|
cat "${search_errors}" >&2
|
||||||
|
fail "could not scan source files"
|
||||||
|
fi
|
||||||
|
}
|
||||||
|
|
||||||
|
# Keep hidden files and build output out of the repository-wide scans.
|
||||||
|
source_excludes=(
|
||||||
|
':(exclude,glob)**/.*'
|
||||||
|
':(exclude,glob)**/.*/**'
|
||||||
|
':(exclude)dist/**'
|
||||||
|
)
|
||||||
|
|
||||||
require_file() {
|
require_file() {
|
||||||
[ -f "$1" ] || fail "missing required file: $1"
|
[ -f "$1" ] || fail "missing required file: $1"
|
||||||
}
|
}
|
||||||
@@ -152,7 +179,11 @@ if grep -Ev '^[[:space:]]*(#|$)' silo.env | grep -q '='; then
|
|||||||
fail "silo.env must not contain active assignments that shadow /etc/default/minio"
|
fail "silo.env must not contain active assignments that shadow /etc/default/minio"
|
||||||
fi
|
fi
|
||||||
|
|
||||||
if rg -n 'pgsty/minio:' .github/workflows Dockerfile.goreleaser helm/silo; then
|
delivery_hits="$(search_text 'pgsty/minio:' -- .github/workflows Dockerfile.goreleaser helm/silo \
|
||||||
|
':(exclude,glob).github/workflows/**/.*' ':(exclude,glob).github/workflows/**/.*/**' \
|
||||||
|
':(exclude,glob)helm/silo/**/.*' ':(exclude,glob)helm/silo/**/.*/**')"
|
||||||
|
if [ -n "${delivery_hits}" ]; then
|
||||||
|
printf '%s\n' "${delivery_hits}" >&2
|
||||||
fail "an active delivery surface still publishes the frozen pgsty/minio image"
|
fail "an active delivery surface still publishes the frozen pgsty/minio image"
|
||||||
fi
|
fi
|
||||||
|
|
||||||
@@ -165,10 +196,10 @@ fi
|
|||||||
# So two rules. First, no live URL may resolve to the old repository anywhere,
|
# So two rules. First, no live URL may resolve to the old repository anywhere,
|
||||||
# READMEs and CONTRIBUTORS.md included.
|
# READMEs and CONTRIBUTORS.md included.
|
||||||
old_repo_pattern='pgsty/minio(\.git)?([^[:alnum:]_.-]|$)'
|
old_repo_pattern='pgsty/minio(\.git)?([^[:alnum:]_.-]|$)'
|
||||||
stale_repo_url="$(rg -n -e "github\.com/${old_repo_pattern}" -e "hub\.docker\.com/r/${old_repo_pattern}" \
|
stale_repo_url="$(search_text -e "github\.com/${old_repo_pattern}" -e "hub\.docker\.com/r/${old_repo_pattern}" -- . \
|
||||||
--glob '!.git/**' --glob '!dist/**' \
|
"${source_excludes[@]}" ':(exclude)SILO_REBRANDING_MIGRATION.md' \
|
||||||
--glob '!SILO_REBRANDING_MIGRATION.md' \
|
':(exclude)buildscripts/rebrand-guard/compat-baseline.json')"
|
||||||
--glob '!buildscripts/rebrand-guard/compat-baseline.json' . |
|
stale_repo_url="$(printf '%s\n' "${stale_repo_url}" |
|
||||||
sed 's#^\./##' | grep -v '^buildscripts/verify-rebrand\.sh:' || true)"
|
sed 's#^\./##' | grep -v '^buildscripts/verify-rebrand\.sh:' || true)"
|
||||||
if [ -n "${stale_repo_url}" ]; then
|
if [ -n "${stale_repo_url}" ]; then
|
||||||
printf '%s\n' "${stale_repo_url}" >&2
|
printf '%s\n' "${stale_repo_url}" >&2
|
||||||
@@ -180,32 +211,32 @@ fi
|
|||||||
# legacy image, the two READMEs that document the rename and the archived
|
# legacy image, the two READMEs that document the rename and the archived
|
||||||
# minio branch, and historical issue titles in CONTRIBUTORS.md.
|
# minio branch, and historical issue titles in CONTRIBUTORS.md.
|
||||||
repo_guard_allowlist='^(buildscripts/minio-upgrade\.sh|buildscripts/verify-rebrand\.sh|buildscripts/helm-migration-guard/main\.go|README\.md|README_ZH\.md|CONTRIBUTORS\.md):'
|
repo_guard_allowlist='^(buildscripts/minio-upgrade\.sh|buildscripts/verify-rebrand\.sh|buildscripts/helm-migration-guard/main\.go|README\.md|README_ZH\.md|CONTRIBUTORS\.md):'
|
||||||
stale_repo="$(rg -n "${old_repo_pattern}" --glob '!.git/**' --glob '!dist/**' \
|
stale_repo="$(search_text "${old_repo_pattern}" -- . "${source_excludes[@]}" \
|
||||||
--glob '!SILO_REBRANDING_MIGRATION.md' \
|
':(exclude)SILO_REBRANDING_MIGRATION.md' ':(exclude)buildscripts/rebrand-guard/compat-baseline.json')"
|
||||||
--glob '!buildscripts/rebrand-guard/compat-baseline.json' . |
|
stale_repo="$(printf '%s\n' "${stale_repo}" |
|
||||||
sed 's#^\./##' | grep -Ev "${repo_guard_allowlist}" || true)"
|
sed 's#^\./##' | grep -Ev "${repo_guard_allowlist}" || true)"
|
||||||
if [ -n "${stale_repo}" ]; then
|
if [ -n "${stale_repo}" ]; then
|
||||||
printf '%s\n' "${stale_repo}" >&2
|
printf '%s\n' "${stale_repo}" >&2
|
||||||
fail "a source reference still names the pre-rename pgsty/minio repository"
|
fail "a source reference still names the pre-rename pgsty/minio repository"
|
||||||
fi
|
fi
|
||||||
|
|
||||||
stale_branch="$(rg -n 'pgsty/silo/(blob/|tree/|raw/)?master' \
|
stale_branch="$(search_text 'pgsty/silo/(blob/|tree/|raw/)?master' -- . "${source_excludes[@]}")"
|
||||||
--glob '!.git/**' --glob '!dist/**' . || true)"
|
|
||||||
if [ -n "${stale_branch}" ]; then
|
if [ -n "${stale_branch}" ]; then
|
||||||
printf '%s\n' "${stale_branch}" >&2
|
printf '%s\n' "${stale_branch}" >&2
|
||||||
fail "a link still targets the retired master branch; raw and Actions URLs do not follow a branch rename"
|
fail "a link still targets the retired master branch; raw and Actions URLs do not follow a branch rename"
|
||||||
fi
|
fi
|
||||||
|
|
||||||
for workflow in .github/workflows/go.yml .github/workflows/vulncheck.yml; do
|
for workflow in .github/workflows/go.yml .github/workflows/vulncheck.yml; do
|
||||||
if rg -q '^\s+- master$' "${workflow}"; then
|
stale_filter="$(search_text '^[[:space:]]+- master$' -- "${workflow}")"
|
||||||
|
if [ -n "${stale_filter}" ]; then
|
||||||
fail "${workflow} still filters on master and would go silently dormant on main"
|
fail "${workflow} still filters on master and would go silently dormant on main"
|
||||||
fi
|
fi
|
||||||
require_text "${workflow}" " - main"
|
require_text "${workflow}" " - main"
|
||||||
done
|
done
|
||||||
|
|
||||||
network_hits="$(rg -n --glob '*.go' --glob '!**/*_test.go' \
|
network_hits="$(search_text \
|
||||||
'https?://[^"`[:space:]]*(dl\.min\.io|subnet\.min\.io|api\.min\.io|slack\.min\.io|play\.min\.io)' \
|
'https?://[^"`[:space:]]*(dl\.min\.io|subnet\.min\.io|api\.min\.io|slack\.min\.io|play\.min\.io)' -- \
|
||||||
cmd internal || true)"
|
':(glob)cmd/**/*.go' ':(glob)internal/**/*.go' ':(exclude,glob)**/*_test.go' "${source_excludes[@]}")"
|
||||||
network_hits="$(printf '%s\n' "${network_hits}" | grep -Ev '^[^:]+:[0-9]+:[[:space:]]*//' || true)"
|
network_hits="$(printf '%s\n' "${network_hits}" | grep -Ev '^[^:]+:[0-9]+:[[:space:]]*//' || true)"
|
||||||
if [ -n "${network_hits}" ]; then
|
if [ -n "${network_hits}" ]; then
|
||||||
printf '%s\n' "${network_hits}" >&2
|
printf '%s\n' "${network_hits}" >&2
|
||||||
|
|||||||
@@ -11,6 +11,7 @@ import (
|
|||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
"net/http/httptest"
|
||||||
|
"net/url"
|
||||||
"strings"
|
"strings"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
@@ -248,8 +249,20 @@ func TestIAMAdminPolicylessSelf(t *testing.T) {
|
|||||||
ctx, sys := iamAdminFixture(t)
|
ctx, sys := iamAdminFixture(t)
|
||||||
caller := iamAdminUser(ctx, t, sys, "review-policyless", "")
|
caller := iamAdminUser(ctx, t, sys, "review-policyless", "")
|
||||||
svc := iamAdminServiceAccount(ctx, t, sys, caller.AccessKey, "rv-policyless-child", "")
|
svc := iamAdminServiceAccount(ctx, t, sys, caller.AccessKey, "rv-policyless-child", "")
|
||||||
|
other := iamAdminUser(ctx, t, sys, "bulk-other", "readwrite")
|
||||||
|
_ = iamAdminServiceAccount(ctx, t, sys, other.AccessKey, "bulk-other-child", "")
|
||||||
api := adminAPIHandlers{}
|
api := adminAPIHandlers{}
|
||||||
assertIAMAdminStatus(t, "list", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), 200)
|
assertIAMAdminStatus(t, "list", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), 200)
|
||||||
|
rec := iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-access-keys-bulk?listType=all", nil, nil, api.ListAccessKeysBulk)
|
||||||
|
assertIAMAdminStatus(t, "list bulk self", rec, 200)
|
||||||
|
data, err := madmin.DecryptData(caller.SecretKey, rec.Body)
|
||||||
|
mustIAM(t, err)
|
||||||
|
var listed map[string]madmin.ListAccessKeysResp
|
||||||
|
mustIAM(t, json.Unmarshal(data, &listed))
|
||||||
|
entry, ok := listed[caller.AccessKey]
|
||||||
|
if len(listed) != 1 || !ok || len(entry.ServiceAccounts) != 1 || entry.ServiceAccounts[0].AccessKey != svc.AccessKey || len(entry.STSKeys) != 0 {
|
||||||
|
t.Fatalf("policyless bulk self returned unexpected identities: %+v", listed)
|
||||||
|
}
|
||||||
assertIAMAdminStatus(t, "info service", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.InfoServiceAccount), 200)
|
assertIAMAdminStatus(t, "info service", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.InfoServiceAccount), 200)
|
||||||
assertIAMAdminStatus(t, "info access key", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-access-key?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.InfoAccessKey), 200)
|
assertIAMAdminStatus(t, "info access key", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-access-key?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.InfoAccessKey), 200)
|
||||||
assertIAMAdminStatus(t, "delete", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.DeleteServiceAccount), 204)
|
assertIAMAdminStatus(t, "delete", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.DeleteServiceAccount), 204)
|
||||||
@@ -384,6 +397,7 @@ func TestIAMAdminSelfDenials(t *testing.T) {
|
|||||||
}
|
}
|
||||||
api := adminAPIHandlers{}
|
api := adminAPIHandlers{}
|
||||||
assertIAMAdminStatus(t, "list", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), 403)
|
assertIAMAdminStatus(t, "list", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), 403)
|
||||||
|
assertIAMAdminStatus(t, "list bulk self", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-access-keys-bulk?listType=all", nil, nil, api.ListAccessKeysBulk), 403)
|
||||||
assertIAMAdminStatus(t, "info service", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.InfoServiceAccount), 403)
|
assertIAMAdminStatus(t, "info service", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.InfoServiceAccount), 403)
|
||||||
assertIAMAdminStatus(t, "info access key", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-access-key?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.InfoAccessKey), 403)
|
assertIAMAdminStatus(t, "info access key", iamAdminCall(t, caller, "GET", "/minio/admin/v3/info-access-key?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.InfoAccessKey), 403)
|
||||||
assertIAMAdminStatus(t, "delete", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.DeleteServiceAccount), 403)
|
assertIAMAdminStatus(t, "delete", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.DeleteServiceAccount), 403)
|
||||||
@@ -477,7 +491,10 @@ func TestIAMAdminListSelfAliases(t *testing.T) {
|
|||||||
ctx, sys := iamAdminFixture(t)
|
ctx, sys := iamAdminFixture(t)
|
||||||
parent := iamAdminUser(ctx, t, sys, "alias-parent", "readwrite")
|
parent := iamAdminUser(ctx, t, sys, "alias-parent", "readwrite")
|
||||||
target := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-target", "")
|
target := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-target", "")
|
||||||
caller := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-caller", "")
|
serviceCaller := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-caller", "")
|
||||||
|
caller := serviceCaller
|
||||||
|
other := iamAdminUser(ctx, t, sys, "alias-other", "readwrite")
|
||||||
|
_ = iamAdminServiceAccount(ctx, t, sys, other.AccessKey, "alias-other-child", "")
|
||||||
if kind == "sts" {
|
if kind == "sts" {
|
||||||
var err error
|
var err error
|
||||||
caller, err = auth.GetNewCredentialsWithMetadata(map[string]any{parentClaim: parent.AccessKey, "exp": UTCNow().Add(time.Hour).Unix()}, globalActiveCred.SecretKey)
|
caller, err = auth.GetNewCredentialsWithMetadata(map[string]any{parentClaim: parent.AccessKey, "exp": UTCNow().Add(time.Hour).Unix()}, globalActiveCred.SecretKey)
|
||||||
@@ -501,6 +518,38 @@ func TestIAMAdminListSelfAliases(t *testing.T) {
|
|||||||
if !found {
|
if !found {
|
||||||
t.Fatalf("self alias %q did not list the parent's service accounts", alias)
|
t.Fatalf("self alias %q did not list the parent's service accounts", alias)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
query := url.Values{"listType": {madmin.AccessKeyListAll}}
|
||||||
|
if alias != "" {
|
||||||
|
query.Set("users", alias)
|
||||||
|
}
|
||||||
|
bulk := iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-access-keys-bulk?"+query.Encode(), nil, nil, api.ListAccessKeysBulk)
|
||||||
|
assertIAMAdminStatus(t, "list bulk self alias", bulk, 200)
|
||||||
|
data, err = madmin.DecryptData(caller.SecretKey, bulk.Body)
|
||||||
|
mustIAM(t, err)
|
||||||
|
var listed map[string]madmin.ListAccessKeysResp
|
||||||
|
mustIAM(t, json.Unmarshal(data, &listed))
|
||||||
|
entry, ok := listed[parent.AccessKey]
|
||||||
|
if len(listed) != 1 || !ok || len(entry.ServiceAccounts) != 2 {
|
||||||
|
t.Fatalf("bulk self alias %q returned unexpected identities: %+v", alias, listed)
|
||||||
|
}
|
||||||
|
want := map[string]bool{target.AccessKey: true, serviceCaller.AccessKey: true}
|
||||||
|
for _, account := range entry.ServiceAccounts {
|
||||||
|
if !want[account.AccessKey] {
|
||||||
|
t.Fatalf("bulk self alias %q returned unexpected or duplicate service account %s", alias, account.AccessKey)
|
||||||
|
}
|
||||||
|
delete(want, account.AccessKey)
|
||||||
|
}
|
||||||
|
if kind == "sts" {
|
||||||
|
if len(entry.STSKeys) != 1 || entry.STSKeys[0].AccessKey != caller.AccessKey {
|
||||||
|
t.Fatalf("bulk self alias %q returned unexpected STS accounts: %+v", alias, entry.STSKeys)
|
||||||
|
}
|
||||||
|
} else if len(entry.STSKeys) != 0 {
|
||||||
|
t.Fatalf("service caller returned unexpected STS accounts: %+v", entry.STSKeys)
|
||||||
|
}
|
||||||
|
if bytes.Contains(data, []byte(serviceCaller.SecretKey)) || bytes.Contains(data, []byte(`"secretKey"`)) || bytes.Contains(data, []byte(`"sessionToken"`)) {
|
||||||
|
t.Fatal("bulk listing returned credential secrets")
|
||||||
|
}
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1237,7 +1237,7 @@ func (a adminAPIHandlers) ListAccessKeysBulk(w http.ResponseWriter, r *http.Requ
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
if !globalIAMSys.IsAllowed(policy.Args{
|
if !isAllowedServiceAccountAction(cred, policy.Args{
|
||||||
AccountName: cred.AccessKey,
|
AccountName: cred.AccessKey,
|
||||||
Groups: cred.Groups,
|
Groups: cred.Groups,
|
||||||
Action: policy.ListServiceAccountsAdminAction,
|
Action: policy.ListServiceAccountsAdminAction,
|
||||||
@@ -1250,12 +1250,12 @@ func (a adminAPIHandlers) ListAccessKeysBulk(w http.ResponseWriter, r *http.Requ
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if selfOnly && len(users) == 0 {
|
if selfOnly {
|
||||||
selfUser := cred.AccessKey
|
selfUser := cred.AccessKey
|
||||||
if cred.ParentUser != "" {
|
if cred.ParentUser != "" {
|
||||||
selfUser = cred.ParentUser
|
selfUser = cred.ParentUser
|
||||||
}
|
}
|
||||||
users = append(users, selfUser)
|
users = []string{selfUser}
|
||||||
}
|
}
|
||||||
|
|
||||||
var checkedUserList []string
|
var checkedUserList []string
|
||||||
|
|||||||
+1
-1
@@ -883,7 +883,7 @@ func serverHandleEnvVars() {
|
|||||||
}
|
}
|
||||||
globalInplaceUpdateDisabled = true
|
globalInplaceUpdateDisabled = true
|
||||||
if env.IsSet(api.EnvAPIMultipartListing) {
|
if env.IsSet(api.EnvAPIMultipartListing) {
|
||||||
logger.Warning("%s is ignored: multipart listing always uses persistent metadata; uploads without identity fields are omitted", api.EnvAPIMultipartListing)
|
logger.Warning("%s is ignored: multipart listing uses the legacy cache/exact-key behavior; persistent listing is deferred", api.EnvAPIMultipartListing)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Check if the supported credential env vars,
|
// Check if the supported credential env vars,
|
||||||
|
|||||||
@@ -14,7 +14,7 @@ import (
|
|||||||
// Run explicitly with -benchtime=1x. Fixture creation is excluded from timing;
|
// Run explicitly with -benchtime=1x. Fixture creation is excluded from timing;
|
||||||
// each request still discovers real on-disk metadata. The 16-drive/10k case
|
// each request still discovers real on-disk metadata. The 16-drive/10k case
|
||||||
// deliberately exceeds the existing aggregate directory-entry budget.
|
// deliberately exceeds the existing aggregate directory-entry budget.
|
||||||
func BenchmarkMultipartListingCapacity(b *testing.B) {
|
func BenchmarkMultipartDiagnosticScanCapacity(b *testing.B) {
|
||||||
for _, tc := range []struct {
|
for _, tc := range []struct {
|
||||||
drives, uploads int
|
drives, uploads int
|
||||||
mixed, limited bool
|
mixed, limited bool
|
||||||
@@ -67,7 +67,7 @@ func BenchmarkMultipartListingCapacity(b *testing.B) {
|
|||||||
}
|
}
|
||||||
b.ResetTimer()
|
b.ResetTimer()
|
||||||
for range b.N {
|
for range b.N {
|
||||||
got, err := z.ListMultipartUploads(b.Context(), bucket, "dir/", "", "", "", 1000)
|
got, err := z.scanMultipartUploadsForTest(b.Context(), bucket, "dir/", "", "", "", 1000)
|
||||||
if tc.limited {
|
if tc.limited {
|
||||||
var limited SlowDown
|
var limited SlowDown
|
||||||
if !errors.As(err, &limited) || len(got.Uploads) != 0 {
|
if !errors.As(err, &limited) || len(got.Uploads) != 0 {
|
||||||
@@ -83,3 +83,88 @@ func BenchmarkMultipartListingCapacity(b *testing.B) {
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// The public path is measured with a populated cache as well as after cache
|
||||||
|
// loss. An empty post-restart response alone is not capacity acceptance.
|
||||||
|
func BenchmarkMultipartPublicListingCapacity(b *testing.B) {
|
||||||
|
for _, tc := range []struct {
|
||||||
|
drives, uploads int
|
||||||
|
sameKey bool
|
||||||
|
}{
|
||||||
|
{4, 10_000, false},
|
||||||
|
{16, 3_126, false},
|
||||||
|
{16, 10_000, true},
|
||||||
|
} {
|
||||||
|
b.Run(fmt.Sprintf("drives=%d/uploads=%d/same-key=%t", tc.drives, tc.uploads, tc.sameKey), func(b *testing.B) {
|
||||||
|
obj, dirs, err := prepareErasure(b.Context(), tc.drives)
|
||||||
|
if err != nil {
|
||||||
|
b.Fatal(err)
|
||||||
|
}
|
||||||
|
z := obj.(*erasureServerPools)
|
||||||
|
b.Cleanup(func() { z.Shutdown(context.Background()); removeRoots(dirs) })
|
||||||
|
const bucket, other = "multipart-public-capacity", "multipart-public-other"
|
||||||
|
for _, name := range []string{bucket, other} {
|
||||||
|
if err := z.MakeBucket(b.Context(), name, MakeBucketOptions{}); err != nil {
|
||||||
|
b.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
wantCached, wantExact := 0, 1
|
||||||
|
for n := range tc.uploads {
|
||||||
|
target, key := bucket, fmt.Sprintf("dir/%05d", n)
|
||||||
|
if tc.sameKey {
|
||||||
|
key = "dir/00000"
|
||||||
|
} else if n%2 != 0 {
|
||||||
|
target = other
|
||||||
|
}
|
||||||
|
if target == bucket {
|
||||||
|
wantCached++
|
||||||
|
}
|
||||||
|
if _, err := z.NewMultipartUpload(b.Context(), target, key, ObjectOptions{}); err != nil {
|
||||||
|
b.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if tc.sameKey {
|
||||||
|
wantExact = 1000
|
||||||
|
}
|
||||||
|
if tc.drives == 16 && !tc.sameKey {
|
||||||
|
_, err := z.scanMultipartUploadsForTest(b.Context(), bucket, "", "", "", "", 1000)
|
||||||
|
var limited SlowDown
|
||||||
|
if !errors.As(err, &limited) {
|
||||||
|
b.Fatalf("fixture did not reproduce the previous global budget failure: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
var directoryCalls, metadataCalls atomic.Int64
|
||||||
|
for _, set := range z.serverPools[0].sets {
|
||||||
|
original := set.getDisks
|
||||||
|
set.getDisks = func() []StorageAPI {
|
||||||
|
disks := append([]StorageAPI(nil), original()...)
|
||||||
|
for i, disk := range disks {
|
||||||
|
disks[i] = multipartListingCountingDisk{disk, &directoryCalls, &metadataCalls}
|
||||||
|
}
|
||||||
|
return disks
|
||||||
|
}
|
||||||
|
b.Cleanup(func() { set.getDisks = original })
|
||||||
|
}
|
||||||
|
measure := func(name, prefix string, want int) {
|
||||||
|
b.Run(name, func(b *testing.B) {
|
||||||
|
directoryCalls.Store(0)
|
||||||
|
metadataCalls.Store(0)
|
||||||
|
b.ResetTimer()
|
||||||
|
for range b.N {
|
||||||
|
got, err := z.ListMultipartUploads(b.Context(), bucket, prefix, "", "", "", 1000)
|
||||||
|
if err != nil || len(got.Uploads) != want {
|
||||||
|
b.Fatalf("public listing: count=%d want=%d err=%v", len(got.Uploads), want, err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
b.StopTimer()
|
||||||
|
b.ReportMetric(float64(directoryCalls.Load())/float64(b.N), "ListDir/op")
|
||||||
|
b.ReportMetric(float64(metadataCalls.Load())/float64(b.N), "ReadVersion/op")
|
||||||
|
})
|
||||||
|
}
|
||||||
|
measure("populated-cache", "", wantCached)
|
||||||
|
z.mpCache.Clear()
|
||||||
|
measure("lost-cache", "", 0)
|
||||||
|
measure("known-key", "dir/00000", wantExact)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -359,7 +359,7 @@ func (z *erasureServerPools) multipartPreflight(ctx context.Context) (multipartP
|
|||||||
return multipartPreflightReport{}, err
|
return multipartPreflightReport{}, err
|
||||||
}
|
}
|
||||||
defer s.close()
|
defer s.close()
|
||||||
report := multipartPreflightReport{Complete: true, Mode: "persistent"}
|
report := multipartPreflightReport{Complete: true, Mode: "legacy"}
|
||||||
for p, pool := range z.serverPools {
|
for p, pool := range z.serverPools {
|
||||||
for i, set := range pool.sets {
|
for i, set := range pool.sets {
|
||||||
_, _ = set.scanMultipartUploads(s, "", p, i)
|
_, _ = set.scanMultipartUploads(s, "", p, i)
|
||||||
|
|||||||
@@ -22,10 +22,53 @@ import (
|
|||||||
"github.com/minio/minio/internal/config/storageclass"
|
"github.com/minio/minio/internal/config/storageclass"
|
||||||
)
|
)
|
||||||
|
|
||||||
// Check the real handler and storage path, including a successful abort between
|
// The persistent scanner is retained for diagnostics. This test-only wrapper
|
||||||
// pages. Choose IDs increasing in both initiation time and lexical order so the
|
// preserves its pagination coverage without exposing it through public listing.
|
||||||
// failure does not depend on differing interpretations of S3 marker ordering.
|
func (z *erasureServerPools) scanMultipartUploadsForTest(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (ListMultipartsInfo, error) {
|
||||||
func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
|
if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
|
||||||
|
return ListMultipartsInfo{}, err
|
||||||
|
}
|
||||||
|
if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
|
||||||
|
return ListMultipartsInfo{}, toObjectErr(err, bucket)
|
||||||
|
}
|
||||||
|
scan, err := startMultipartScan(ctx, false)
|
||||||
|
if err != nil {
|
||||||
|
return ListMultipartsInfo{}, err
|
||||||
|
}
|
||||||
|
defer scan.close()
|
||||||
|
|
||||||
|
var uploads []MultipartInfo
|
||||||
|
for idx, pool := range z.serverPools {
|
||||||
|
if z.IsSuspended(idx) {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
poolUploads, err := pool.scanMultipartUploadsForTest(scan, bucket, idx)
|
||||||
|
if err != nil {
|
||||||
|
return ListMultipartsInfo{}, err
|
||||||
|
}
|
||||||
|
uploads = append(uploads, poolUploads...)
|
||||||
|
}
|
||||||
|
|
||||||
|
// Confirmed historical records lack a recoverable bucket/key and are
|
||||||
|
// omitted by the scan. They do not block listing identifiable uploads.
|
||||||
|
return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
|
||||||
|
}
|
||||||
|
|
||||||
|
func (s *erasureSets) scanMultipartUploadsForTest(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, error) {
|
||||||
|
var uploads []MultipartInfo
|
||||||
|
for i, set := range s.sets {
|
||||||
|
setUploads, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
uploads = append(uploads, setUploads...)
|
||||||
|
}
|
||||||
|
return uploads, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
// Check that a successful HTTP abort removes the cached upload. The restored
|
||||||
|
// cache path still ignores page limits; this is a documented legacy limitation.
|
||||||
|
func TestMultipartListingAbortHTTPUpdatesCache(t *testing.T) {
|
||||||
z, _ := consistencyPools(t)
|
z, _ := consistencyPools(t)
|
||||||
bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads", "AbortMultipart"}, MakeBucketOptions{})
|
bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads", "AbortMultipart"}, MakeBucketOptions{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -81,8 +124,8 @@ func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
|
|||||||
return result
|
return result
|
||||||
}
|
}
|
||||||
first := list("", "", 1)
|
first := list("", "", 1)
|
||||||
if len(first.Uploads) != 1 || first.Uploads[0].UploadID != firstID || !first.IsTruncated {
|
if len(first.Uploads) != 3 || first.Uploads[0].UploadID != firstID || first.IsTruncated {
|
||||||
t.Fatalf("unexpected first page: %+v", first)
|
t.Fatalf("unexpected legacy cache view: %+v", first)
|
||||||
}
|
}
|
||||||
rec := request(http.MethodDelete, getAbortMultipartUploadURL("", bucket, "a", firstID))
|
rec := request(http.MethodDelete, getAbortMultipartUploadURL("", bucket, "a", firstID))
|
||||||
if rec.Code != http.StatusNoContent {
|
if rec.Code != http.StatusNoContent {
|
||||||
@@ -93,12 +136,12 @@ func TestMultipartListingAbortBetweenHTTPPages(t *testing.T) {
|
|||||||
for _, upload := range rest.Uploads {
|
for _, upload := range rest.Uploads {
|
||||||
keys = append(keys, upload.Key)
|
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)
|
t.Logf("GET cache=200; DELETE upload=204; GET cache=200 keys=%v truncated=%v", keys, rest.IsTruncated)
|
||||||
if _, err := z.GetMultipartInfo(t.Context(), bucket, "a", secondID, ObjectOptions{}); err != nil {
|
if _, err := z.GetMultipartInfo(t.Context(), bucket, "a", secondID, ObjectOptions{}); err != nil {
|
||||||
t.Fatalf("remaining upload is not valid: %v", err)
|
t.Fatalf("remaining upload is not valid: %v", err)
|
||||||
}
|
}
|
||||||
if len(rest.Uploads) != 2 || rest.Uploads[0].UploadID != secondID {
|
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)
|
t.Fatalf("valid remaining upload for key a is missing from cache: %+v", rest.Uploads)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -149,7 +192,7 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
|
|||||||
}
|
}
|
||||||
z.mpCache.Clear()
|
z.mpCache.Clear()
|
||||||
for _, target := range []string{bucket, other} {
|
for _, target := range []string{bucket, other} {
|
||||||
got, err := z.ListMultipartUploads(t.Context(), target, "", "", "", "", 10)
|
got, err := z.scanMultipartUploadsForTest(t.Context(), target, "", "", "", "", 10)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -159,14 +202,14 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
|
|||||||
requireMultipartUploadKeys(t, got)
|
requireMultipartUploadKeys(t, got)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
// Exact-looking prefixes now have the same durable semantics.
|
// Diagnostic discovery cannot recover an unidentified historical key.
|
||||||
exact, err := z.ListMultipartUploads(t.Context(), other, "old", "", "", "", 10)
|
exact, err := z.scanMultipartUploadsForTest(t.Context(), other, "old", "", "", "", 10)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
requireMultipartUploadKeys(t, exact)
|
requireMultipartUploadKeys(t, exact)
|
||||||
report, err := z.multipartPreflight(t.Context())
|
report, err := z.multipartPreflight(t.Context())
|
||||||
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" {
|
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "legacy" {
|
||||||
t.Fatalf("legacy preflight: %+v %v", report, err)
|
t.Fatalf("legacy preflight: %+v %v", report, err)
|
||||||
}
|
}
|
||||||
if err = z.AbortMultipartUpload(t.Context(), other, "old", old.UploadID, ObjectOptions{}); err != nil {
|
if err = z.AbortMultipartUpload(t.Context(), other, "old", old.UploadID, ObjectOptions{}); err != nil {
|
||||||
@@ -176,14 +219,14 @@ func TestMultipartListingLegacyPreflight(t *testing.T) {
|
|||||||
if err != nil || !report.Ready || report.LegacyUploads != 0 {
|
if err != nil || !report.Ready || report.LegacyUploads != 0 {
|
||||||
t.Fatalf("drained preflight: %+v %v", report, err)
|
t.Fatalf("drained preflight: %+v %v", report, err)
|
||||||
}
|
}
|
||||||
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
|
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
requireMultipartUploadKeys(t, got, "a")
|
requireMultipartUploadKeys(t, got, "a")
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMultipartListingIdentityFallback(t *testing.T) {
|
func TestMultipartDiagnosticIdentityFallback(t *testing.T) {
|
||||||
for _, kind := range []string{"old", "partial", "corrupt", "read-failure", "wrong-bucket"} {
|
for _, kind := range []string{"old", "partial", "corrupt", "read-failure", "wrong-bucket"} {
|
||||||
t.Run(kind, func(t *testing.T) {
|
t.Run(kind, func(t *testing.T) {
|
||||||
z, set, bucket := multipartListingFixture(t)
|
z, set, bucket := multipartListingFixture(t)
|
||||||
@@ -220,7 +263,7 @@ func TestMultipartListingIdentityFallback(t *testing.T) {
|
|||||||
}
|
}
|
||||||
return disks
|
return disks
|
||||||
}
|
}
|
||||||
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
|
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -469,7 +512,7 @@ func (d multipartListingCountingDisk) ReadVersion(ctx context.Context, original,
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Counts storage API calls; this is not a deployment throughput benchmark.
|
// Counts storage API calls; this is not a deployment throughput benchmark.
|
||||||
func TestMultipartListingScanCosts(t *testing.T) {
|
func TestMultipartDiagnosticScanCosts(t *testing.T) {
|
||||||
obj, dirs, err := prepareErasure(t.Context(), 4)
|
obj, dirs, err := prepareErasure(t.Context(), 4)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
@@ -507,7 +550,7 @@ func TestMultipartListingScanCosts(t *testing.T) {
|
|||||||
t.Helper()
|
t.Helper()
|
||||||
directoryCalls.Store(0)
|
directoryCalls.Store(0)
|
||||||
metadataCalls.Store(0)
|
metadataCalls.Store(0)
|
||||||
result, err := z.ListMultipartUploads(t.Context(), bucket, "dir/", "", "", "", 1)
|
result, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "dir/", "", "", "", 1)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -589,7 +632,7 @@ func TestMultipartListingMissingMarkers(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMultipartListingCoverage(t *testing.T) {
|
func TestMultipartDiagnosticCoverage(t *testing.T) {
|
||||||
z, set, bucket := multipartListingFixture(t)
|
z, set, bucket := multipartListingFixture(t)
|
||||||
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
|
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -605,12 +648,12 @@ func TestMultipartListingCoverage(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
set.getDisks = func() []StorageAPI { return []StorageAPI{disks[0], disks[1], nil, nil} }
|
set.getDisks = func() []StorageAPI { return []StorageAPI{disks[0], disks[1], nil, nil} }
|
||||||
_, err = z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
|
_, err = z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
|
||||||
if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
|
if err == nil || toAPIError(t.Context(), err).HTTPStatusCode != 503 {
|
||||||
t.Fatalf("incomplete discovery returned success: %v", err)
|
t.Fatalf("incomplete discovery returned success: %v", err)
|
||||||
}
|
}
|
||||||
set.getDisks = func() []StorageAPI { return []StorageAPI{nil, disks[1], disks[2], disks[3]} }
|
set.getDisks = func() []StorageAPI { return []StorageAPI{nil, disks[1], disks[2], disks[3]} }
|
||||||
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
|
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -621,7 +664,7 @@ func TestMultipartListingCoverage(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMultipartListingBudgetAndAdmission(t *testing.T) {
|
func TestMultipartDiagnosticBudgetAndAdmission(t *testing.T) {
|
||||||
_, set, bucket := multipartListingFixture(t)
|
_, set, bucket := multipartListingFixture(t)
|
||||||
if _, err := set.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil {
|
if _, err := set.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{}); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
@@ -651,7 +694,7 @@ func TestMultipartListingBudgetAndAdmission(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMultipartListingAdmissionHTTP(t *testing.T) {
|
func TestMultipartPublicListingAdmissionHTTP(t *testing.T) {
|
||||||
z, _, _ := multipartListingFixture(t)
|
z, _, _ := multipartListingFixture(t)
|
||||||
bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
|
bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -673,12 +716,12 @@ func TestMultipartListingAdmissionHTTP(t *testing.T) {
|
|||||||
}
|
}
|
||||||
rec := httptest.NewRecorder()
|
rec := httptest.NewRecorder()
|
||||||
router.ServeHTTP(rec, req)
|
router.ServeHTTP(rec, req)
|
||||||
var response APIErrorResponse
|
var response ListMultipartUploadsResponse
|
||||||
if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
|
if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if rec.Code != http.StatusServiceUnavailable || response.Code != "SlowDown" {
|
if rec.Code != http.StatusOK || len(response.Uploads) != 0 {
|
||||||
t.Fatalf("admission returned %d %s", rec.Code, rec.Body.String())
|
t.Fatalf("diagnostic admission blocked public listing: %d %s", rec.Code, rec.Body.String())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -705,12 +748,12 @@ func TestMultipartListingPreflightMinorityLegacy(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
report, err := z.multipartPreflight(t.Context())
|
report, err := z.multipartPreflight(t.Context())
|
||||||
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "persistent" {
|
if err != nil || report.Ready || !report.Complete || report.LegacyUploads != 1 || report.Mode != "legacy" {
|
||||||
t.Fatalf("minority legacy copy must prevent readiness: %+v %v", report, err)
|
t.Fatalf("minority legacy copy must prevent readiness: %+v %v", report, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMultipartListingCancellationRetainsAdmission(t *testing.T) {
|
func TestMultipartDiagnosticCancellationRetainsAdmission(t *testing.T) {
|
||||||
z, set, bucket := multipartListingFixture(t)
|
z, set, bucket := multipartListingFixture(t)
|
||||||
for i := range 40 {
|
for i := range 40 {
|
||||||
if _, err := z.NewMultipartUpload(t.Context(), bucket, fmt.Sprintf("key-%02d", i), ObjectOptions{}); err != nil {
|
if _, err := z.NewMultipartUpload(t.Context(), bucket, fmt.Sprintf("key-%02d", i), ObjectOptions{}); err != nil {
|
||||||
@@ -749,7 +792,7 @@ func TestMultipartListingCancellationRetainsAdmission(t *testing.T) {
|
|||||||
defer cancel()
|
defer cancel()
|
||||||
done := make(chan error, 1)
|
done := make(chan error, 1)
|
||||||
go func() {
|
go func() {
|
||||||
_, err := z.ListMultipartUploads(ctx, bucket, "", "", "", "", 10)
|
_, err := z.scanMultipartUploadsForTest(ctx, bucket, "", "", "", "", 10)
|
||||||
done <- err
|
done <- err
|
||||||
}()
|
}()
|
||||||
for range 16 {
|
for range 16 {
|
||||||
|
|||||||
@@ -0,0 +1,106 @@
|
|||||||
|
// Copyright (c) 2026 Ruohang Feng
|
||||||
|
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||||
|
|
||||||
|
package cmd
|
||||||
|
|
||||||
|
import (
|
||||||
|
"context"
|
||||||
|
"encoding/xml"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"sync/atomic"
|
||||||
|
"testing"
|
||||||
|
)
|
||||||
|
|
||||||
|
type multipartRootScanGuard struct {
|
||||||
|
StorageAPI
|
||||||
|
roots *atomic.Int64
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d multipartRootScanGuard) ListDir(ctx context.Context, original, volume, dir string, count int) ([]string, error) {
|
||||||
|
if volume == minioMetaMultipartBucket && dir == "" {
|
||||||
|
d.roots.Add(1)
|
||||||
|
return nil, errFaultyDisk
|
||||||
|
}
|
||||||
|
return d.StorageAPI.ListDir(ctx, original, volume, dir, count)
|
||||||
|
}
|
||||||
|
|
||||||
|
// A busy or unreadable diagnostic namespace must not become a prerequisite for
|
||||||
|
// ordinary listing, including the lower-level exact-object entry points.
|
||||||
|
func TestMultipartPublicListingIndependentOfDiagnostics(t *testing.T) {
|
||||||
|
for _, busy := range []bool{false, true} {
|
||||||
|
t.Run(map[bool]string{false: "unreadable-root", true: "busy-diagnostics"}[busy], func(t *testing.T) {
|
||||||
|
z, _ := consistencyPools(t)
|
||||||
|
bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
const key = "dir/known"
|
||||||
|
mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
var roots atomic.Int64
|
||||||
|
for _, pool := range z.serverPools {
|
||||||
|
for _, set := range pool.sets {
|
||||||
|
original := set.getDisks
|
||||||
|
set.getDisks = func() []StorageAPI {
|
||||||
|
disks := append([]StorageAPI(nil), original()...)
|
||||||
|
for i, disk := range disks {
|
||||||
|
if disk != nil {
|
||||||
|
disks[i] = multipartRootScanGuard{disk, &roots}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return disks
|
||||||
|
}
|
||||||
|
t.Cleanup(func() { set.getDisks = original })
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if busy {
|
||||||
|
for range cap(multipartScanSlots) {
|
||||||
|
scan, err := startMultipartScan(t.Context(), true)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
defer scan.close()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, prefix := range []string{"", key, "dir/"} {
|
||||||
|
req, err := newTestSignedRequestV4(http.MethodGet,
|
||||||
|
getListMultipartUploadsURLWithParams("", bucket, prefix, "", "", "", "1"),
|
||||||
|
0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
router.ServeHTTP(rec, req)
|
||||||
|
if rec.Code != http.StatusOK {
|
||||||
|
t.Fatalf("prefix %q: HTTP %d %s", prefix, rec.Code, rec.Body.String())
|
||||||
|
}
|
||||||
|
var got ListMultipartUploadsResponse
|
||||||
|
if err := xml.Unmarshal(rec.Body.Bytes(), &got); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
if prefix == "dir/" {
|
||||||
|
if len(got.Uploads) != 0 {
|
||||||
|
t.Fatalf("partial prefix must retain the documented exact-key limitation: %+v", got)
|
||||||
|
}
|
||||||
|
} else if len(got.Uploads) != 1 || got.Uploads[0].UploadID != mp.UploadID {
|
||||||
|
t.Fatalf("known upload missing: %+v", got)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, pool := range z.serverPools {
|
||||||
|
for _, list := range []func(context.Context, string, string, string, string, string, int) (ListMultipartsInfo, error){
|
||||||
|
pool.ListMultipartUploads, pool.getHashedSet(key).ListMultipartUploads,
|
||||||
|
} {
|
||||||
|
if _, err := list(t.Context(), bucket, key, "", "", "", 10); err != nil {
|
||||||
|
t.Fatalf("lower-level exact listing: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if roots.Load() != 0 {
|
||||||
|
t.Fatalf("public listing attempted %d global multipart scans", roots.Load())
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -4,11 +4,10 @@
|
|||||||
package cmd
|
package cmd
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"bytes"
|
||||||
"context"
|
"context"
|
||||||
"encoding/xml"
|
|
||||||
"errors"
|
"errors"
|
||||||
"net/http"
|
"net/http"
|
||||||
"net/http/httptest"
|
|
||||||
"slices"
|
"slices"
|
||||||
"testing"
|
"testing"
|
||||||
)
|
)
|
||||||
@@ -33,14 +32,10 @@ func removeMultipartIdentity(metadata map[string]string) {
|
|||||||
delete(metadata, multipartMetaObject)
|
delete(metadata, multipartMetaObject)
|
||||||
}
|
}
|
||||||
|
|
||||||
// The actual S3 handler must use the same semantics for every prefix and page,
|
// The retained diagnostic scanner handles mixed records independently of a
|
||||||
// even with old records and a stale, misleading node-local cache.
|
// stale cache. Public HTTP behavior is tested separately.
|
||||||
func TestMultipartListingMixedHTTP(t *testing.T) {
|
func TestMultipartDiagnosticMixed(t *testing.T) {
|
||||||
z, set, _ := multipartListingFixture(t)
|
z, set, bucket := multipartListingFixture(t)
|
||||||
bucket, router, err := initAPIHandlerTest(t.Context(), z, []string{"ListMultipartUploads"}, MakeBucketOptions{})
|
|
||||||
if err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
for _, key := range []string{"dir/old", "dir/a", "dir/sub/b", "outside"} {
|
for _, key := range []string{"dir/old", "dir/a", "dir/sub/b", "outside"} {
|
||||||
mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
|
mp, err := z.NewMultipartUpload(t.Context(), bucket, key, ObjectOptions{})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -54,22 +49,11 @@ func TestMultipartListingMixedHTTP(t *testing.T) {
|
|||||||
z.mpCache.Store("stale", MultipartInfo{Bucket: bucket, Object: "ghost", UploadID: "stale"})
|
z.mpCache.Store("stale", MultipartInfo{Bucket: bucket, Object: "ghost", UploadID: "stale"})
|
||||||
list := func(prefix, keyMarker, uploadMarker, delimiter string) ListMultipartUploadsResponse {
|
list := func(prefix, keyMarker, uploadMarker, delimiter string) ListMultipartUploadsResponse {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
req, err := newTestSignedRequestV4(http.MethodGet,
|
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, prefix, keyMarker, uploadMarker, delimiter, 1)
|
||||||
getListMultipartUploadsURLWithParams("", bucket, prefix, keyMarker, uploadMarker, delimiter, "1"),
|
|
||||||
0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
rec := httptest.NewRecorder()
|
return generateListMultipartUploadsResponse(bucket, got, "")
|
||||||
router.ServeHTTP(rec, req)
|
|
||||||
var got ListMultipartUploadsResponse
|
|
||||||
if rec.Code != http.StatusOK {
|
|
||||||
t.Fatalf("list %q: HTTP %d %s", prefix, rec.Code, rec.Body.String())
|
|
||||||
}
|
|
||||||
if err := xml.Unmarshal(rec.Body.Bytes(), &got); err != nil {
|
|
||||||
t.Fatal(err)
|
|
||||||
}
|
|
||||||
return got
|
|
||||||
}
|
}
|
||||||
var keys []string
|
var keys []string
|
||||||
keyMarker, uploadMarker := "", ""
|
keyMarker, uploadMarker := "", ""
|
||||||
@@ -103,7 +87,7 @@ func TestMultipartListingMixedHTTP(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMultipartListingInvalidIdentity(t *testing.T) {
|
func TestMultipartDiagnosticInvalidIdentity(t *testing.T) {
|
||||||
for _, kind := range []string{"missing-bucket", "missing-object", "empty-bucket", "empty-object", "empty-both", "wrong-hash", "read-failure"} {
|
for _, kind := range []string{"missing-bucket", "missing-object", "empty-bucket", "empty-object", "empty-both", "wrong-hash", "read-failure"} {
|
||||||
t.Run(kind, func(t *testing.T) {
|
t.Run(kind, func(t *testing.T) {
|
||||||
z, set, bucket := multipartListingFixture(t)
|
z, set, bucket := multipartListingFixture(t)
|
||||||
@@ -142,7 +126,7 @@ func TestMultipartListingInvalidIdentity(t *testing.T) {
|
|||||||
return disks
|
return disks
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
|
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
|
||||||
if err == nil || len(got.Uploads) != 0 || toAPIError(t.Context(), err).HTTPStatusCode != http.StatusServiceUnavailable {
|
if err == nil || len(got.Uploads) != 0 || toAPIError(t.Context(), err).HTTPStatusCode != http.StatusServiceUnavailable {
|
||||||
t.Fatalf("invalid/unreadable record returned a successful page: %+v %v", got, err)
|
t.Fatalf("invalid/unreadable record returned a successful page: %+v %v", got, err)
|
||||||
}
|
}
|
||||||
@@ -185,7 +169,14 @@ func TestMultipartListingRestart(t *testing.T) {
|
|||||||
if _, cached := obj.(*erasureServerPools).mpCache.Load(mp.UploadID); cached {
|
if _, cached := obj.(*erasureServerPools).mpCache.Load(mp.UploadID); cached {
|
||||||
t.Fatal("restart unexpectedly preserved the upload cache")
|
t.Fatal("restart unexpectedly preserved the upload cache")
|
||||||
}
|
}
|
||||||
got, err := obj.ListMultipartUploads(t.Context(), bucket, "dir/", "", "", "", 10)
|
for _, prefix := range []string{"", "dir/"} {
|
||||||
|
got, err := obj.ListMultipartUploads(t.Context(), bucket, prefix, "", "", "", 10)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
requireMultipartUploadKeys(t, got)
|
||||||
|
}
|
||||||
|
got, err := obj.ListMultipartUploads(t.Context(), bucket, key, "", "", "", 10)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -193,9 +184,18 @@ func TestMultipartListingRestart(t *testing.T) {
|
|||||||
if got.Uploads[0].UploadID != mp.UploadID {
|
if got.Uploads[0].UploadID != mp.UploadID {
|
||||||
t.Fatal("restart changed upload identity")
|
t.Fatal("restart changed upload identity")
|
||||||
}
|
}
|
||||||
|
part, err := obj.PutObjectPart(t.Context(), bucket, key, mp.UploadID, 1,
|
||||||
|
mustGetPutObjReader(t, bytes.NewBufferString("part"), 4, "", ""), ObjectOptions{})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("continue known upload after restart: %v", err)
|
||||||
|
}
|
||||||
|
if _, err := obj.CompleteMultipartUpload(t.Context(), bucket, key, mp.UploadID,
|
||||||
|
[]CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{}); err != nil {
|
||||||
|
t.Fatalf("complete known upload after restart: %v", err)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMultipartListingConflictingIdentityCopies(t *testing.T) {
|
func TestMultipartDiagnosticConflictingIdentityCopies(t *testing.T) {
|
||||||
for _, oldCopies := range []int{1, 3} {
|
for _, oldCopies := range []int{1, 3} {
|
||||||
z, set, bucket := multipartListingFixture(t)
|
z, set, bucket := multipartListingFixture(t)
|
||||||
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
|
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
|
||||||
@@ -213,14 +213,14 @@ func TestMultipartListingConflictingIdentityCopies(t *testing.T) {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
|
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
|
||||||
if !errors.Is(err, errMultipartListingIdentity) || len(got.Uploads) != 0 {
|
if !errors.Is(err, errMultipartListingIdentity) || len(got.Uploads) != 0 {
|
||||||
t.Fatalf("%d contradictory copies returned a successful page: %+v %v", oldCopies, got, err)
|
t.Fatalf("%d contradictory copies returned a successful page: %+v %v", oldCopies, got, err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestMultipartListingUnsupportedStoredID(t *testing.T) {
|
func TestMultipartDiagnosticUnsupportedStoredID(t *testing.T) {
|
||||||
for _, historical := range []bool{false, true} {
|
for _, historical := range []bool{false, true} {
|
||||||
z, set, bucket := multipartListingFixture(t)
|
z, set, bucket := multipartListingFixture(t)
|
||||||
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
|
mp, err := z.NewMultipartUpload(t.Context(), bucket, "a", ObjectOptions{})
|
||||||
@@ -244,7 +244,7 @@ func TestMultipartListingUnsupportedStoredID(t *testing.T) {
|
|||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
got, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 10)
|
got, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 10)
|
||||||
if historical {
|
if historical {
|
||||||
if err != nil || len(got.Uploads) != 0 {
|
if err != nil || len(got.Uploads) != 0 {
|
||||||
t.Fatalf("unidentified historical ID: %+v %v", got, err)
|
t.Fatalf("unidentified historical ID: %+v %v", got, err)
|
||||||
|
|||||||
@@ -408,7 +408,7 @@ func paginateMultipartUploads(uploads []MultipartInfo, prefix, keyMarker, upload
|
|||||||
}
|
}
|
||||||
|
|
||||||
// listMultipartUploadsExact preserves the hashed exact-object lookup used by
|
// listMultipartUploadsExact preserves the hashed exact-object lookup used by
|
||||||
// multipart write placement. Public listing always uses the durable scan.
|
// multipart write placement and nonempty public listing prefixes.
|
||||||
func (er erasureObjects) listMultipartUploadsExact(ctx context.Context, bucket, object, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
|
func (er erasureObjects) listMultipartUploadsExact(ctx context.Context, bucket, object, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
|
||||||
auditObjectErasureSet(ctx, "ListMultipartUploads", object, &er)
|
auditObjectErasureSet(ctx, "ListMultipartUploads", object, &er)
|
||||||
|
|
||||||
@@ -516,16 +516,7 @@ func (er erasureObjects) ListMultipartUploads(ctx context.Context, bucket, prefi
|
|||||||
if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
|
if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
|
||||||
return ListMultipartsInfo{}, err
|
return ListMultipartsInfo{}, err
|
||||||
}
|
}
|
||||||
scan, err := startMultipartScan(ctx, false)
|
return er.listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
|
||||||
if err != nil {
|
|
||||||
return ListMultipartsInfo{}, err
|
|
||||||
}
|
|
||||||
defer scan.close()
|
|
||||||
uploads, err := er.scanMultipartUploads(scan, bucket, 0, 0)
|
|
||||||
if err != nil {
|
|
||||||
return ListMultipartsInfo{}, err
|
|
||||||
}
|
|
||||||
return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// newMultipartUpload - wrapper for initializing a new multipart
|
// newMultipartUpload - wrapper for initializing a new multipart
|
||||||
|
|||||||
+30
-12
@@ -1916,27 +1916,45 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
|
|||||||
if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
|
if _, err := z.GetBucketInfo(ctx, bucket, BucketOptions{}); err != nil {
|
||||||
return ListMultipartsInfo{}, toObjectErr(err, bucket)
|
return ListMultipartsInfo{}, toObjectErr(err, bucket)
|
||||||
}
|
}
|
||||||
scan, err := startMultipartScan(ctx, false)
|
|
||||||
if err != nil {
|
|
||||||
return ListMultipartsInfo{}, err
|
|
||||||
}
|
|
||||||
defer scan.close()
|
|
||||||
|
|
||||||
var uploads []MultipartInfo
|
poolResult := ListMultipartsInfo{}
|
||||||
|
poolResult.MaxUploads = maxUploads
|
||||||
|
poolResult.KeyMarker = keyMarker
|
||||||
|
poolResult.Prefix = prefix
|
||||||
|
poolResult.Delimiter = delimiter
|
||||||
|
|
||||||
|
// Preserve the legacy cache view, including its incomplete pagination.
|
||||||
|
// Persistent discovery remains an explicit admin diagnostic, not a
|
||||||
|
// prerequisite for public listing or known-upload operations.
|
||||||
|
if prefix == "" {
|
||||||
|
z.mpCache.Range(func(_ string, mp MultipartInfo) bool {
|
||||||
|
if mp.Bucket == bucket {
|
||||||
|
poolResult.Uploads = append(poolResult.Uploads, mp)
|
||||||
|
}
|
||||||
|
return true
|
||||||
|
})
|
||||||
|
sort.Slice(poolResult.Uploads, func(i int, j int) bool {
|
||||||
|
return poolResult.Uploads[i].Initiated.Before(poolResult.Uploads[j].Initiated)
|
||||||
|
})
|
||||||
|
return poolResult, nil
|
||||||
|
}
|
||||||
|
|
||||||
|
if z.SinglePool() {
|
||||||
|
return z.serverPools[0].getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
|
||||||
|
}
|
||||||
|
|
||||||
for idx, pool := range z.serverPools {
|
for idx, pool := range z.serverPools {
|
||||||
if z.IsSuspended(idx) {
|
if z.IsSuspended(idx) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
poolUploads, err := pool.scanMultipartUploads(scan, bucket, idx)
|
result, err := pool.getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return ListMultipartsInfo{}, err
|
return result, err
|
||||||
}
|
}
|
||||||
uploads = append(uploads, poolUploads...)
|
poolResult.Uploads = append(poolResult.Uploads, result.Uploads...)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Confirmed historical records lack a recoverable bucket/key and are
|
return poolResult, nil
|
||||||
// omitted by the scan. They do not block listing identifiable uploads.
|
|
||||||
return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Initiate a new multipart upload on a hashedSet based on object name.
|
// Initiate a new multipart upload on a hashedSet based on object name.
|
||||||
|
|||||||
+1
-25
@@ -886,31 +886,7 @@ func (s *erasureSets) CopyObject(ctx context.Context, srcBucket, srcObject, dstB
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (s *erasureSets) ListMultipartUploads(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
|
func (s *erasureSets) ListMultipartUploads(ctx context.Context, bucket, prefix, keyMarker, uploadIDMarker, delimiter string, maxUploads int) (result ListMultipartsInfo, err error) {
|
||||||
if err := checkListMultipartArgs(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter); err != nil {
|
return s.getHashedSet(prefix).ListMultipartUploads(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
|
||||||
return ListMultipartsInfo{}, err
|
|
||||||
}
|
|
||||||
scan, err := startMultipartScan(ctx, false)
|
|
||||||
if err != nil {
|
|
||||||
return ListMultipartsInfo{}, err
|
|
||||||
}
|
|
||||||
defer scan.close()
|
|
||||||
uploads, err := s.scanMultipartUploads(scan, bucket, 0)
|
|
||||||
if err != nil {
|
|
||||||
return ListMultipartsInfo{}, err
|
|
||||||
}
|
|
||||||
return paginateMultipartUploads(uploads, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads), nil
|
|
||||||
}
|
|
||||||
|
|
||||||
func (s *erasureSets) scanMultipartUploads(scan *multipartScan, bucket string, poolIdx int) ([]MultipartInfo, error) {
|
|
||||||
var uploads []MultipartInfo
|
|
||||||
for i, set := range s.sets {
|
|
||||||
setUploads, err := set.scanMultipartUploads(scan, bucket, poolIdx, i)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
uploads = append(uploads, setUploads...)
|
|
||||||
}
|
|
||||||
return uploads, nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (s *erasureSets) listMultipartUploadsExact(ctx context.Context, bucket, object string) (ListMultipartsInfo, error) {
|
func (s *erasureSets) listMultipartUploadsExact(ctx context.Context, bucket, object string) (ListMultipartsInfo, error) {
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ func requireMultipartUploadKeys(t *testing.T, got ListMultipartsInfo, want ...st
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestListMultipartUploadsS3Compatibility(t *testing.T) {
|
func TestMultipartDiagnosticScanPagination(t *testing.T) {
|
||||||
obj, dirs, err := prepareErasureSets32(t.Context())
|
obj, dirs, err := prepareErasureSets32(t.Context())
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
@@ -81,14 +81,14 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
|
|||||||
uploadIDs[object] = mp.UploadID
|
uploadIDs[object] = mp.UploadID
|
||||||
}
|
}
|
||||||
|
|
||||||
// Durable multipart metadata, rather than this node-local cache, must be
|
// The diagnostic scanner must discover metadata independently of the cache.
|
||||||
// authoritative after a restart or when another node handles the request.
|
// These checks do not claim that public listing offers durable pagination.
|
||||||
z.mpCache.Range(func(uploadID string, _ MultipartInfo) bool {
|
z.mpCache.Range(func(uploadID string, _ MultipartInfo) bool {
|
||||||
z.mpCache.Delete(uploadID)
|
z.mpCache.Delete(uploadID)
|
||||||
return true
|
return true
|
||||||
})
|
})
|
||||||
|
|
||||||
all, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
|
all, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 100)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -97,7 +97,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
|
|||||||
t.Fatalf("complete listing has truncation state: %+v", all)
|
t.Fatalf("complete listing has truncation state: %+v", all)
|
||||||
}
|
}
|
||||||
|
|
||||||
first, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 1)
|
first, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 1)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -108,31 +108,31 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
|
|||||||
objects[0], uploadIDs[objects[0]])
|
objects[0], uploadIDs[objects[0]])
|
||||||
}
|
}
|
||||||
|
|
||||||
rest, err := z.ListMultipartUploads(t.Context(), bucket, "", first.NextKeyMarker, first.NextUploadIDMarker, "", 100)
|
rest, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", first.NextKeyMarker, first.NextUploadIDMarker, "", 100)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
requireMultipartUploadKeys(t, rest, objects[1:]...)
|
requireMultipartUploadKeys(t, rest, objects[1:]...)
|
||||||
|
|
||||||
afterKey, err := z.ListMultipartUploads(t.Context(), bucket, "", objects[1], "", "", 100)
|
afterKey, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", objects[1], "", "", 100)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
requireMultipartUploadKeys(t, afterKey, objects[2:]...)
|
requireMultipartUploadKeys(t, afterKey, objects[2:]...)
|
||||||
|
|
||||||
prefixed, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", "", 100)
|
prefixed, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", "", "", "", 100)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
requireMultipartUploadKeys(t, prefixed, objects[:3]...)
|
requireMultipartUploadKeys(t, prefixed, objects[:3]...)
|
||||||
|
|
||||||
nested, err := z.ListMultipartUploads(t.Context(), bucket, "t/a_b/", "", "", "", 100)
|
nested, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/a_b/", "", "", "", 100)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
requireMultipartUploadKeys(t, nested, objects[:2]...)
|
requireMultipartUploadKeys(t, nested, objects[:2]...)
|
||||||
|
|
||||||
grouped, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", SlashSeparator, 100)
|
grouped, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", "", "", SlashSeparator, 100)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -141,7 +141,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
|
|||||||
t.Fatalf("common prefixes = %v, want %v", grouped.CommonPrefixes, want)
|
t.Fatalf("common prefixes = %v, want %v", grouped.CommonPrefixes, want)
|
||||||
}
|
}
|
||||||
|
|
||||||
groupPage, err := z.ListMultipartUploads(t.Context(), bucket, "t/", "", "", SlashSeparator, 1)
|
groupPage, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", "", "", SlashSeparator, 1)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -153,7 +153,7 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
|
|||||||
groupPage.NextKeyMarker, groupPage.NextUploadIDMarker, groupPage.IsTruncated)
|
groupPage.NextKeyMarker, groupPage.NextUploadIDMarker, groupPage.IsTruncated)
|
||||||
}
|
}
|
||||||
|
|
||||||
groupRest, err := z.ListMultipartUploads(t.Context(), bucket, "t/", groupPage.NextKeyMarker, "", SlashSeparator, 1)
|
groupRest, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "t/", groupPage.NextKeyMarker, "", SlashSeparator, 1)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -197,12 +197,12 @@ func TestListMultipartUploadsS3Compatibility(t *testing.T) {
|
|||||||
er.getUploadIDDir(bucket, legacyObject, uploadIDs[legacyObject]), metadata, fi.WriteQuorum(er.defaultWQuorum())); err != nil {
|
er.getUploadIDDir(bucket, legacyObject, uploadIDs[legacyObject]), metadata, fi.WriteQuorum(er.defaultWQuorum())); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
legacy, err := z.ListMultipartUploads(t.Context(), bucket, legacyObject, "", "", "", 100)
|
legacy, err := z.scanMultipartUploadsForTest(t.Context(), bucket, legacyObject, "", "", "", 100)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
requireMultipartUploadKeys(t, legacy)
|
requireMultipartUploadKeys(t, legacy)
|
||||||
remaining, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 100)
|
remaining, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 100)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -280,7 +280,7 @@ func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) {
|
|||||||
return true
|
return true
|
||||||
})
|
})
|
||||||
|
|
||||||
page, err := z.ListMultipartUploads(t.Context(), bucket, "", "", "", "", 2)
|
page, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", "", "", "", 2)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@@ -289,7 +289,7 @@ func TestListMultipartUploadsGlobalPageAcrossPools(t *testing.T) {
|
|||||||
t.Fatalf("first global page = %+v", page)
|
t.Fatalf("first global page = %+v", page)
|
||||||
}
|
}
|
||||||
|
|
||||||
rest, err := z.ListMultipartUploads(t.Context(), bucket, "", page.NextKeyMarker, page.NextUploadIDMarker, "", 2)
|
rest, err := z.scanMultipartUploadsForTest(t.Context(), bucket, "", page.NextKeyMarker, page.NextUploadIDMarker, "", 2)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -781,6 +781,15 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
|
|||||||
}
|
}
|
||||||
last := ""
|
last := ""
|
||||||
var toMerge []int
|
var toMerge []int
|
||||||
|
var directoryCollisions []int
|
||||||
|
discardDirectories := func() error {
|
||||||
|
for _, idx := range directoryCollisions {
|
||||||
|
if err := selectFrom(idx); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
// Choose the best to return.
|
// Choose the best to return.
|
||||||
for {
|
for {
|
||||||
@@ -790,6 +799,7 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
|
|||||||
best := top[0]
|
best := top[0]
|
||||||
bestIdx := 0
|
bestIdx := 0
|
||||||
toMerge = toMerge[:0]
|
toMerge = toMerge[:0]
|
||||||
|
directoryCollisions = directoryCollisions[:0]
|
||||||
for i, other := range top[1:] {
|
for i, other := range top[1:] {
|
||||||
otherIdx := i + 1
|
otherIdx := i + 1
|
||||||
if other == nil {
|
if other == nil {
|
||||||
@@ -819,10 +829,9 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
|
|||||||
if serverDebugLog {
|
if serverDebugLog {
|
||||||
console.Debugln("mergeEntryChannels: discarding directory", other.name, "for object", best.name)
|
console.Debugln("mergeEntryChannels: discarding directory", other.name, "for object", best.name)
|
||||||
}
|
}
|
||||||
// Discard the directory.
|
// A pending object can resolve to absence. Keep the
|
||||||
if err := selectFrom(otherIdx); err != nil {
|
// directory until the whole object group is confirmed.
|
||||||
return err
|
directoryCollisions = append(directoryCollisions, otherIdx)
|
||||||
}
|
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
// Replace directory with object.
|
// Replace directory with object.
|
||||||
@@ -838,6 +847,7 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
|
|||||||
}
|
}
|
||||||
if best.name > other.name {
|
if best.name > other.name {
|
||||||
toMerge = toMerge[:0]
|
toMerge = toMerge[:0]
|
||||||
|
directoryCollisions = directoryCollisions[:0]
|
||||||
best = other
|
best = other
|
||||||
bestIdx = otherIdx
|
bestIdx = otherIdx
|
||||||
}
|
}
|
||||||
@@ -866,6 +876,11 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
if resolved != nil {
|
||||||
|
if err := discardDirectories(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
}
|
||||||
if resolved != nil && resolved.name > last {
|
if resolved != nil && resolved.name > last {
|
||||||
select {
|
select {
|
||||||
case <-ctxDone:
|
case <-ctxDone:
|
||||||
@@ -876,6 +891,9 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
|
|||||||
}
|
}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
if err := discardDirectories(); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
// Merge any unmerged
|
// Merge any unmerged
|
||||||
if len(toMerge) > 0 {
|
if len(toMerge) > 0 {
|
||||||
|
|||||||
@@ -0,0 +1,243 @@
|
|||||||
|
// Copyright (c) 2026 Ruohang Feng
|
||||||
|
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||||
|
|
||||||
|
package cmd
|
||||||
|
|
||||||
|
import (
|
||||||
|
"bytes"
|
||||||
|
"context"
|
||||||
|
"encoding/xml"
|
||||||
|
"fmt"
|
||||||
|
"io"
|
||||||
|
"net/http"
|
||||||
|
"net/http/httptest"
|
||||||
|
"slices"
|
||||||
|
"testing"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
type listingDirectoryWalkSnapshot struct {
|
||||||
|
StorageAPI
|
||||||
|
bucket string
|
||||||
|
snapshot []byte
|
||||||
|
}
|
||||||
|
|
||||||
|
func (d listingDirectoryWalkSnapshot) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) error {
|
||||||
|
if opts.Bucket != d.bucket {
|
||||||
|
return d.StorageAPI.WalkDir(ctx, opts, out)
|
||||||
|
}
|
||||||
|
_, err := out.Write(d.snapshot)
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Capture two actual directory-walk generations for a directory object, then
|
||||||
|
// acknowledge its deletion. A different pool still contains dir/child. The
|
||||||
|
// stale source walkers disagree, while every current source drive is absent.
|
||||||
|
func TestListingRecoveryDeletedDirectoryKeepsCommonPrefix(t *testing.T) {
|
||||||
|
for _, marker := range []bool{false, true} {
|
||||||
|
for _, sourcePool := range []int{0, 1} {
|
||||||
|
t.Run(fmt.Sprintf("pool=%d/marker=%t", sourcePool, marker), func(t *testing.T) {
|
||||||
|
z, bucket, router := listingPoolsBackend(t)
|
||||||
|
ctx := t.Context()
|
||||||
|
source := z.serverPools[sourcePool].sets[0]
|
||||||
|
target := z.serverPools[1-sourcePool].sets[0]
|
||||||
|
childBody := []byte("surviving child")
|
||||||
|
put := func(set *erasureObjects, key string, tm time.Time) {
|
||||||
|
var body []byte
|
||||||
|
if key == "dir/child" {
|
||||||
|
body = childBody
|
||||||
|
}
|
||||||
|
_, err := set.PutObject(ctx, bucket, encodeDirObject(key), mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm, VersionSuspended: marker})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
tm := time.Now().UTC().Add(-time.Hour)
|
||||||
|
put(target, "dir/child", tm)
|
||||||
|
disks, switchDisks := installListingDiskSwitch(source)
|
||||||
|
snapshots := make([][]byte, len(disks))
|
||||||
|
for gen := range 3 {
|
||||||
|
if gen < 2 {
|
||||||
|
put(source, "dir/", tm.Add(time.Duration(gen)*time.Minute))
|
||||||
|
} else {
|
||||||
|
if _, err := source.DeleteObject(ctx, bucket, encodeDirObject("dir/"), ObjectOptions{VersionSuspended: marker, DeleteMarker: marker, MTime: tm.Add(2 * time.Minute)}); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for i, disk := range disks {
|
||||||
|
wanted := 0
|
||||||
|
if i >= 6 {
|
||||||
|
wanted = 1
|
||||||
|
}
|
||||||
|
if i >= 12 {
|
||||||
|
wanted = 2
|
||||||
|
}
|
||||||
|
if wanted != gen {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
var b bytes.Buffer
|
||||||
|
if err := disk.WalkDir(ctx, WalkDirOptions{Bucket: bucket, Recursive: false}, &b); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
snapshots[i] = append([]byte(nil), b.Bytes()...)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
for _, key := range []string{"dir/child", "dir/"} {
|
||||||
|
request, err := newTestSignedRequestV4(http.MethodGet, getGetObjectURL("", bucket, key), 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
response := httptest.NewRecorder()
|
||||||
|
router.ServeHTTP(response, request.WithContext(ctx))
|
||||||
|
t.Logf("GET %s HTTP %d", key, response.Code)
|
||||||
|
want := http.StatusOK
|
||||||
|
if key == "dir/" {
|
||||||
|
want = http.StatusNotFound
|
||||||
|
}
|
||||||
|
if response.Code != want {
|
||||||
|
t.Fatalf("GET %s: got %d want %d: %s", key, response.Code, want, response.Body.String())
|
||||||
|
}
|
||||||
|
if key == "dir/child" && !bytes.Equal(response.Body.Bytes(), childBody) {
|
||||||
|
t.Fatal("surviving child content changed")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
wrapped := make([]StorageAPI, len(disks))
|
||||||
|
for i, disk := range disks {
|
||||||
|
wrapped[i] = listingDirectoryWalkSnapshot{disk, bucket, snapshots[i]}
|
||||||
|
}
|
||||||
|
switchDisks(wrapped)
|
||||||
|
t.Cleanup(func() { switchDisks(disks) })
|
||||||
|
req, err := newTestSignedRequestV4(http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", "/"), 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
rec := httptest.NewRecorder()
|
||||||
|
router.ServeHTTP(rec, req.WithContext(ctx))
|
||||||
|
t.Logf("LIST HTTP %d body=%s", rec.Code, rec.Body.String())
|
||||||
|
var page ListObjectsV2Response
|
||||||
|
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil {
|
||||||
|
t.Fatalf("LIST failed: %d %s", rec.Code, rec.Body.String())
|
||||||
|
}
|
||||||
|
if len(page.CommonPrefixes) != 1 || page.CommonPrefixes[0].Prefix != "dir/" {
|
||||||
|
t.Fatalf("acknowledged child lost its CommonPrefix: %+v", page)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// The directory must remain in the sorted input when an earlier object name
|
||||||
|
// disappears. Emitting it immediately would skip an intervening name, and
|
||||||
|
// keeping only one colliding directory would leave input order significant.
|
||||||
|
func TestListingRecoveryDirectoryCollisionOrdering(t *testing.T) {
|
||||||
|
for _, object := range []string{"dir", "dir/"} {
|
||||||
|
for _, present := range []bool{false, true} {
|
||||||
|
if present && object == "dir" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
for source := range 3 {
|
||||||
|
t.Run(fmt.Sprintf("object=%s/present=%t/source=%d", object, present, source), func(t *testing.T) {
|
||||||
|
in := make([]chan metaCacheEntry, 6)
|
||||||
|
for i := range in {
|
||||||
|
in[i] = make(chan metaCacheEntry, 1)
|
||||||
|
entry := metaCacheEntry{name: "dir/"}
|
||||||
|
if i == source {
|
||||||
|
entry = metaCacheEntry{name: object, listingSet: &erasureObjects{}}
|
||||||
|
}
|
||||||
|
if i >= 3 {
|
||||||
|
entry = metaCacheEntry{name: []string{"dir-", "dir.ext", "dir0"}[i-3], metadata: []byte{1}}
|
||||||
|
}
|
||||||
|
in[i] <- entry
|
||||||
|
close(in[i])
|
||||||
|
}
|
||||||
|
out := make(chan metaCacheEntry, len(in))
|
||||||
|
err := mergeEntryChannelsWithResolver(t.Context(), in, out, 1, nil, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
|
||||||
|
if len(group) != 1 || group[0].listingSet == nil {
|
||||||
|
t.Fatalf("directory entered the object group: %+v", group)
|
||||||
|
}
|
||||||
|
if !present {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
return &metaCacheEntry{name: object, metadata: []byte{1}}, nil
|
||||||
|
})
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
var names []string
|
||||||
|
for entry := range out {
|
||||||
|
names = append(names, entry.name)
|
||||||
|
if entry.name == "dir/" && entry.isObject() != present {
|
||||||
|
t.Fatalf("directory collision returned the wrong entry type: %+v", entry)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
want := []string{"dir-", "dir.ext", "dir/", "dir0"}
|
||||||
|
if !slices.Equal(names, want) {
|
||||||
|
t.Fatalf("names=%v, want %v", names, want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A regular walk result may be stale when a pending peer forces current
|
||||||
|
// recovery. Keep sibling prefixes until the complete object group is resolved.
|
||||||
|
func TestListingRecoveryMixedDirectoryCollision(t *testing.T) {
|
||||||
|
orders := [][3]int{{0, 1, 2}, {0, 2, 1}, {1, 0, 2}, {1, 2, 0}, {2, 0, 1}, {2, 1, 0}}
|
||||||
|
for _, order := range orders {
|
||||||
|
for _, present := range []bool{false, true} {
|
||||||
|
t.Run(fmt.Sprintf("order=%v/present=%t", order, present), func(t *testing.T) {
|
||||||
|
entries := []metaCacheEntry{
|
||||||
|
{name: "dir/", metadata: []byte{1}},
|
||||||
|
{name: "dir/"},
|
||||||
|
{name: "dir/", listingSet: &erasureObjects{}},
|
||||||
|
}
|
||||||
|
in := make([]chan metaCacheEntry, len(order))
|
||||||
|
for i, index := range order {
|
||||||
|
in[i] = make(chan metaCacheEntry, 2)
|
||||||
|
in[i] <- entries[index]
|
||||||
|
in[i] <- metaCacheEntry{name: fmt.Sprintf("tail-%d", i), metadata: []byte{1}}
|
||||||
|
close(in[i])
|
||||||
|
}
|
||||||
|
out := make(chan metaCacheEntry, 6)
|
||||||
|
calls := 0
|
||||||
|
err := mergeEntryChannelsWithResolver(t.Context(), in, out, 1, nil, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
|
||||||
|
calls++
|
||||||
|
if len(group) != 2 {
|
||||||
|
t.Fatalf("incomplete object group: %+v", group)
|
||||||
|
}
|
||||||
|
pending := 0
|
||||||
|
for _, entry := range group {
|
||||||
|
if entry.name != "dir/" || entry.isDir() {
|
||||||
|
t.Fatalf("unexpected recovery entry: %+v", entry)
|
||||||
|
}
|
||||||
|
if entry.listingSet != nil {
|
||||||
|
pending++
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if pending != 1 {
|
||||||
|
t.Fatalf("pending entries=%d, want 1", pending)
|
||||||
|
}
|
||||||
|
if !present {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
|
return &metaCacheEntry{name: "dir/", metadata: []byte{1}}, nil
|
||||||
|
})
|
||||||
|
if err != nil || calls != 1 {
|
||||||
|
t.Fatalf("err=%v recovery calls=%d", err, calls)
|
||||||
|
}
|
||||||
|
var names []string
|
||||||
|
for entry := range out {
|
||||||
|
names = append(names, entry.name)
|
||||||
|
if entry.listingSet != nil || entry.name == "dir/" && entry.isObject() != present {
|
||||||
|
t.Fatalf("unexpected merged entry: %+v", entry)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
want := []string{"dir/", "tail-0", "tail-1", "tail-2"}
|
||||||
|
if !slices.Equal(names, want) {
|
||||||
|
t.Fatalf("names=%v, want %v", names, want)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -221,34 +221,51 @@ func TestListingRecoveryNormalPathDoesNotReadOrLock(t *testing.T) {
|
|||||||
|
|
||||||
func TestListingPendingDirectoryCollision(t *testing.T) {
|
func TestListingPendingDirectoryCollision(t *testing.T) {
|
||||||
for _, objectFirst := range []bool{false, true} {
|
for _, objectFirst := range []bool{false, true} {
|
||||||
t.Run(fmt.Sprint(objectFirst), func(t *testing.T) {
|
for _, present := range []bool{true, false} {
|
||||||
in := []chan metaCacheEntry{make(chan metaCacheEntry, 1), make(chan metaCacheEntry, 1)}
|
t.Run(fmt.Sprintf("objectFirst=%v/present=%v", objectFirst, present), func(t *testing.T) {
|
||||||
|
in := []chan metaCacheEntry{make(chan metaCacheEntry, 2), make(chan metaCacheEntry, 2), make(chan metaCacheEntry, 2)}
|
||||||
objectIndex := 1
|
objectIndex := 1
|
||||||
if objectFirst {
|
if objectFirst {
|
||||||
objectIndex = 0
|
objectIndex = 0
|
||||||
}
|
}
|
||||||
in[objectIndex] <- metaCacheEntry{name: "dir/", listingSet: &erasureObjects{}}
|
in[objectIndex] <- metaCacheEntry{name: "dir/", listingSet: &erasureObjects{}}
|
||||||
in[1-objectIndex] <- metaCacheEntry{name: "dir/"}
|
in[1-objectIndex] <- metaCacheEntry{name: "dir/"}
|
||||||
|
in[2] <- metaCacheEntry{name: "a", metadata: []byte{1}}
|
||||||
for _, c := range in {
|
for _, c := range in {
|
||||||
close(c)
|
close(c)
|
||||||
}
|
}
|
||||||
out := make(chan metaCacheEntry, 2)
|
out := make(chan metaCacheEntry, 4)
|
||||||
calls := 0
|
calls := 0
|
||||||
err := mergeEntryChannelsWithResolver(t.Context(), in, out, 1, nil, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
|
err := mergeEntryChannelsWithResolver(t.Context(), in, out, 1, nil, func(group []*metaCacheEntry) (*metaCacheEntry, error) {
|
||||||
calls++
|
calls++
|
||||||
if len(group) != 1 || group[0].listingSet == nil {
|
if len(group) != 1 || group[0].listingSet == nil {
|
||||||
t.Fatalf("wrong pending group: %+v", group)
|
t.Fatalf("wrong pending group: %+v", group)
|
||||||
}
|
}
|
||||||
|
if !present {
|
||||||
|
return nil, nil
|
||||||
|
}
|
||||||
return &metaCacheEntry{name: "dir/", metadata: []byte{1}}, nil
|
return &metaCacheEntry{name: "dir/", metadata: []byte{1}}, nil
|
||||||
})
|
})
|
||||||
if err != nil || calls != 1 || len(out) != 1 {
|
if err != nil || calls != 1 || len(out) != 2 {
|
||||||
t.Fatalf("err=%v recoveries=%d outputs=%d", err, calls, len(out))
|
t.Fatalf("err=%v recoveries=%d outputs=%d", err, calls, len(out))
|
||||||
}
|
}
|
||||||
if entry := <-out; entry.isDir() || !entry.isObjectDir() {
|
first := <-out
|
||||||
|
if first.name != "a" {
|
||||||
|
t.Fatalf("expected smaller key first, got %+v", first)
|
||||||
|
}
|
||||||
|
entry := <-out
|
||||||
|
if entry.name != "dir/" || entry.listingSet != nil {
|
||||||
|
t.Fatalf("unexpected recovered entry: %+v", entry)
|
||||||
|
}
|
||||||
|
if present && (entry.isDir() || !entry.isObjectDir()) {
|
||||||
t.Fatalf("object became directory: %+v", entry)
|
t.Fatalf("object became directory: %+v", entry)
|
||||||
}
|
}
|
||||||
|
if !present && (!entry.isDir() || entry.name != "dir/") {
|
||||||
|
t.Fatalf("absent pending object dropped live directory prefix: %+v", entry)
|
||||||
|
}
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestListingRecoveryRejectsUnexpectedSinglePoolSet(t *testing.T) {
|
func TestListingRecoveryRejectsUnexpectedSinglePoolSet(t *testing.T) {
|
||||||
|
|||||||
@@ -2,4 +2,4 @@ module github.com/minio/minio/docs/debugging/reorder-disks
|
|||||||
|
|
||||||
go 1.26.0
|
go 1.26.0
|
||||||
|
|
||||||
require github.com/pgsty/silo-pkg/v3 v3.13.2
|
require github.com/pgsty/silo-pkg/v3 v3.14.3
|
||||||
|
|||||||
@@ -1,2 +1,2 @@
|
|||||||
github.com/pgsty/silo-pkg/v3 v3.13.2 h1:Clw11c/J54Tx6pijNCWtXiC7e0fwP/f5Tgeb6fsXg2w=
|
github.com/pgsty/silo-pkg/v3 v3.14.3 h1:hMDxOo36F1whGRkYScrUGGwn98FvsmJ8hVM6pE4i8Bs=
|
||||||
github.com/pgsty/silo-pkg/v3 v3.13.2/go.mod h1:0GmaDA0ArQ8bkAI/obiSNTBQzdgBu6W0e0olDCbyXXo=
|
github.com/pgsty/silo-pkg/v3 v3.14.3/go.mod h1:zqIm3FMETmwPHRKuUExCDGfYkCNXt+vpfBnPM821RdI=
|
||||||
|
|||||||
@@ -81,7 +81,7 @@ require (
|
|||||||
github.com/nats-io/stan.go v0.10.4
|
github.com/nats-io/stan.go v0.10.4
|
||||||
github.com/ncw/directio v1.0.5
|
github.com/ncw/directio v1.0.5
|
||||||
github.com/nsqio/go-nsq v1.1.0
|
github.com/nsqio/go-nsq v1.1.0
|
||||||
github.com/pgsty/silo-pkg/v3 v3.14.2
|
github.com/pgsty/silo-pkg/v3 v3.14.3
|
||||||
github.com/philhofer/fwd v1.2.1
|
github.com/philhofer/fwd v1.2.1
|
||||||
github.com/pierrec/lz4/v4 v4.1.30
|
github.com/pierrec/lz4/v4 v4.1.30
|
||||||
github.com/pkg/errors v0.9.1
|
github.com/pkg/errors v0.9.1
|
||||||
|
|||||||
@@ -549,8 +549,8 @@ github.com/pgsty/mc v0.0.0-20261017101407-fa3b8bf72458 h1:f3kbkjtjj6D7PXmsIeKrv8
|
|||||||
github.com/pgsty/mc v0.0.0-20261017101407-fa3b8bf72458/go.mod h1:n8NDi0J3hPUGCIM9tOxlhYXkYS9xiTVIxlxa3Rwx1cw=
|
github.com/pgsty/mc v0.0.0-20261017101407-fa3b8bf72458/go.mod h1:n8NDi0J3hPUGCIM9tOxlhYXkYS9xiTVIxlxa3Rwx1cw=
|
||||||
github.com/pgsty/silo-console v0.0.0-20261017112417-ef2dd2c41d21 h1:PHxPswGiEwU9MVisu0/Ab7AcNX4IveiX1thQzUXOsM4=
|
github.com/pgsty/silo-console v0.0.0-20261017112417-ef2dd2c41d21 h1:PHxPswGiEwU9MVisu0/Ab7AcNX4IveiX1thQzUXOsM4=
|
||||||
github.com/pgsty/silo-console v0.0.0-20261017112417-ef2dd2c41d21/go.mod h1:HFNaafu5iCO9bWS86PIyTbgF+yPD4phFt+gxML/1jJU=
|
github.com/pgsty/silo-console v0.0.0-20261017112417-ef2dd2c41d21/go.mod h1:HFNaafu5iCO9bWS86PIyTbgF+yPD4phFt+gxML/1jJU=
|
||||||
github.com/pgsty/silo-pkg/v3 v3.14.2 h1:nU9gQ9LuDhN/Z6gZNvFj94Vb9xrjEMKAegX16v55s/I=
|
github.com/pgsty/silo-pkg/v3 v3.14.3 h1:hMDxOo36F1whGRkYScrUGGwn98FvsmJ8hVM6pE4i8Bs=
|
||||||
github.com/pgsty/silo-pkg/v3 v3.14.2/go.mod h1:zqIm3FMETmwPHRKuUExCDGfYkCNXt+vpfBnPM821RdI=
|
github.com/pgsty/silo-pkg/v3 v3.14.3/go.mod h1:zqIm3FMETmwPHRKuUExCDGfYkCNXt+vpfBnPM821RdI=
|
||||||
github.com/philhofer/fwd v1.2.1 h1:UtpS73q8bFgy/yH3gCq8LSNvj5gv/FowNOiWZa7uKwQ=
|
github.com/philhofer/fwd v1.2.1 h1:UtpS73q8bFgy/yH3gCq8LSNvj5gv/FowNOiWZa7uKwQ=
|
||||||
github.com/philhofer/fwd v1.2.1/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
|
github.com/philhofer/fwd v1.2.1/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
|
||||||
github.com/pierrec/lz4/v4 v4.1.30 h1:cchX8N2DVP668WkElI9QMwVyoNabLkq1LofDHFeIrdg=
|
github.com/pierrec/lz4/v4 v4.1.30 h1:cchX8N2DVP668WkElI9QMwVyoNabLkq1LofDHFeIrdg=
|
||||||
|
|||||||
Reference in New Issue
Block a user