Compare commits

..

4 Commits

Author SHA1 Message Date
Feng Ruohang 2a4db6a4f9 fix(listing): snapshot asynchronous cache cleanup inputs
Capture the metacache peer before launching cleanup and pass listing
options by value. The parent can clear its cache ID without changing the
background RPC lookup or racing with its reads of shared request state.

Publish complete immutable disk snapshots through atomic.Value in the
listing fixtures instead of replacing getDisks after requests start.
Add a real peer RPC regression for cleanup using the original cache ID;
reverting the options copy reproduces both the race and an empty RPC ID.

Verified the final 18-test listing race suite, 30 repetitions of the real
peer regression, formatting checks and make lint. Prior runtime evidence
retains its original e90cb206 source identity.

Refs: https://github.com/pgsty/silo/issues/218
Refs: https://github.com/pgsty/silo/pull/239
Signed-off-by: Feng Ruohang <rh@vonng.com>
2026-10-16 01:23:29 +08:00
Feng Ruohang e90cb2067d fix(listing): preserve readable keys across pool and metadata disagreements
Continue PR #221 while retaining its original commits and authorship.
Route latest-only fallback reads through the namespace lock used by normal
writes, propagate terminal scan errors before cache completion, and retain
delete markers until the cross-pool merge has resolved them.

Recount narrowly eligible latest null-version metadata when mixed erasure
layouts prevent selection. Keep each EC group separate and preserve the
requested quorum, strict resolution, and version-listing behavior.

Add regression coverage for locking, terminal errors, cache completion,
delete markers, mixed EC layouts, and HTTP behavior; run focused race tests
in CI. Verified with make test, focused race tests, and local distributed
single- and multi-pool rolling-restart workloads with exact final oracles.

Refs: https://github.com/pgsty/silo/issues/218
Refs: https://github.com/pgsty/silo/pull/221
Signed-off-by: Feng Ruohang <rh@vonng.com>
2026-10-16 00:58:46 +08:00
mr javad seydi 8c784f6133 fix(listing): narrow the list fallback per review
Signed-off-by: mr javad seydi <seydi.birjand@gmail.com>
2026-09-23 19:03:00 +03:30
mr javad seydi 258b167b1c fix(listing): resolve undecidable list entries against the locked read path
Refs: https://github.com/pgsty/silo/issues/218

Signed-off-by: mr javad seydi <seydi.birjand@gmail.com>
2026-09-23 18:23:15 +03:30
24 changed files with 1958 additions and 721 deletions
+3
View File
@@ -96,6 +96,9 @@ jobs:
- name: Run multipart listing and cancellation tests under race detector
run: go test -race ./cmd -run '^Test(MultipartListing|MultipartAbort|PaginateMultipartUploads|ListMultipartUploads)' -count=1 -timeout=5m
- name: Run listing metadata recovery, pagination and cancellation tests under race detector
run: go test -race ./cmd -run '^Test(ListFallback|ListingCache|ListingMerge|MergeXLV2LatestNullCrossParity|CountXLV2MixedECGroups|ListCrossParity)' -count=1 -timeout=5m
- name: Run CPU metrics tests under race detector
run: go test -race ./cmd -run '^TestLoadCPUMetrics' -count=1 -timeout=5m
+1 -3
View File
@@ -249,10 +249,8 @@ func (a adminAPIHandlers) AddServiceAccountLDAP(w http.ResponseWriter, r *http.R
// In case of LDAP/OIDC we need to set `opts.claims` to ensure
// it is associated with the LDAP/OIDC user properly.
// The new credential's policy comes from the request, not the caller.
for k, v := range cred.Claims {
switch k {
case expClaim, policy.SessionPolicyName, sessionPolicyNameExtracted, iamPolicyClaimNameSA():
if k == expClaim {
continue
}
opts.claims[k] = v
-507
View File
@@ -1,507 +0,0 @@
// Copyright (c) 2026 PGSTY
// SPDX-License-Identifier: AGPL-3.0-or-later
package cmd
import (
"bytes"
"context"
"encoding/base64"
"encoding/json"
"io"
"net/http"
"net/http/httptest"
"strings"
"sync/atomic"
"testing"
"time"
"github.com/minio/madmin-go/v3"
"github.com/minio/minio/internal/auth"
polplugin "github.com/minio/minio/internal/config/policy/plugin"
"github.com/minio/mux"
xnet "github.com/pgsty/silo-pkg/v3/net"
"github.com/pgsty/silo-pkg/v3/policy"
)
func iamAdminFixture(t *testing.T) (context.Context, *IAMSys) {
t.Helper()
ctx, sys, obj := prepareIAMRevisionFixture(t)
globalObjLayerMutex.Lock()
globalObjectAPI = obj
globalObjLayerMutex.Unlock()
return ctx, sys
}
func iamAdminUser(ctx context.Context, t *testing.T, sys *IAMSys, name, policyName string) auth.Credentials {
t.Helper()
cred := auth.Credentials{AccessKey: name, SecretKey: "review-user-valid-secret"}
_, err := sys.CreateUser(ctx, name, madmin.AddOrUpdateUserReq{SecretKey: cred.SecretKey, Status: madmin.AccountEnabled})
mustIAM(t, err)
if policyName != "" {
_, err = sys.PolicyDBSet(ctx, name, policyName, regUser, false)
mustIAM(t, err)
}
return cred
}
func iamAdminServiceAccount(ctx context.Context, t *testing.T, sys *IAMSys, parent, key, document string) auth.Credentials {
t.Helper()
var sp *policy.Policy
var err error
if document != "" {
sp, err = policy.ParseConfig(strings.NewReader(document))
mustIAM(t, err)
}
cred, _, err := sys.NewServiceAccount(ctx, parent, nil, newServiceAccountOpts{accessKey: key, secretKey: "review-service-valid-secret", sessionPolicy: sp})
mustIAM(t, err)
return cred
}
func iamAdminRequest(t *testing.T, cred auth.Credentials, method, target string, payload any) *http.Request {
t.Helper()
var body []byte
if payload != nil {
raw, err := json.Marshal(payload)
mustIAM(t, err)
body, err = madmin.EncryptData(cred.SecretKey, raw)
mustIAM(t, err)
}
headers := map[string]string{}
if cred.IsTemp() && !cred.IsServiceAccount() {
headers["X-Amz-Security-Token"] = cred.SessionToken
}
req, err := newTestSignedRequestV4(method, "http://localhost"+target, int64(len(body)), bytes.NewReader(body), cred.AccessKey, cred.SecretKey, headers)
mustIAM(t, err)
mustIAM(t, req.ParseForm())
return req
}
func iamAdminCall(t *testing.T, cred auth.Credentials, method, target string, payload any, vars map[string]string, handler http.HandlerFunc) *httptest.ResponseRecorder {
t.Helper()
req := iamAdminRequest(t, cred, method, target, payload)
req = mux.SetURLVars(req, vars)
rec := httptest.NewRecorder()
req = req.WithContext(newContext(req, rec, "IAMAdminAuthorization"))
handler(rec, req)
return rec
}
func assertIAMAdminStatus(t *testing.T, operation string, rec *httptest.ResponseRecorder, want int) {
t.Helper()
if rec.Code != want {
t.Errorf("%s status=%d want=%d body=%s", operation, rec.Code, want, rec.Body.String())
return
}
if want == http.StatusForbidden || want == http.StatusNotFound {
var response APIErrorResponse
mustIAM(t, json.Unmarshal(rec.Body.Bytes(), &response))
wantCode := errorCodes.ToAPIErr(ErrAccessDenied).Code
if want == http.StatusNotFound {
wantCode = errorCodes.ToAPIErr(ErrAdminServiceAccountNotFound).Code
}
if response.Code != wantCode {
t.Errorf("%s error code=%s want=%s", operation, response.Code, wantCode)
}
}
}
func iamAdminPlugin(t *testing.T, allow bool) {
t.Helper()
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
_ = json.NewEncoder(w).Encode(map[string]bool{"result": allow})
}))
t.Cleanup(server.Close)
u, err := xnet.ParseHTTPURL(server.URL)
mustIAM(t, err)
previous := newGlobalAuthZPluginFn()
setGlobalAuthZPlugin(polplugin.New(polplugin.Args{URL: u, CloseRespFn: func(r io.ReadCloser) { r.Close() }}))
t.Cleanup(func() { setGlobalAuthZPlugin(previous) })
}
func TestIAMAdminRootEmptyPolicy(t *testing.T) {
for _, doc := range []string{"", "null", "{}", `{"Statement":null}`} {
t.Run("policy="+doc, func(t *testing.T) {
ctx, sys := iamAdminFixture(t)
caller := iamAdminServiceAccount(ctx, t, sys, globalActiveCred.AccessKey, "review-root-empty", doc)
sibling := iamAdminServiceAccount(ctx, t, sys, globalActiveCred.AccessKey, "review-root-sibling", "")
req := iamAdminRequest(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil)
checked, owner, code := checkKeyValid(req, caller.AccessKey)
if code != ErrNone {
t.Fatalf("checkKeyValid: %v", code)
}
active, _ := isAllowedBySessionPolicyForServiceAccount(policy.Args{Claims: checked.Claims})
t.Logf("owner=%v parsed_active_session_policy=%v", owner, active)
api := adminAPIHandlers{}
assertIAMAdminStatus(t, "create", iamAdminCall(t, caller, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{AccessKey: "review-root-new", SecretKey: "review-service-valid-secret"}, nil, api.AddServiceAccount), 200)
assertIAMAdminStatus(t, "update", iamAdminCall(t, caller, "POST", "/minio/admin/v3/update-service-account?accessKey="+sibling.AccessKey, madmin.UpdateServiceAccountReq{NewName: "review"}, map[string]string{"accessKey": sibling.AccessKey}, api.UpdateServiceAccount), 204)
assertIAMAdminStatus(t, "delete", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+sibling.AccessKey, nil, map[string]string{"accessKey": sibling.AccessKey}, api.DeleteServiceAccount), 204)
assertIAMAdminStatus(t, "revoke", iamAdminCall(t, caller, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true", nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 204)
})
}
}
func TestIAMAdminEmptyPolicyDelegation(t *testing.T) {
for _, parentType := range []string{"root", "user"} {
for _, document := range []string{"null", "{}", `{"Statement":null}`} {
t.Run(parentType+"/"+document, func(t *testing.T) {
ctx, sys := iamAdminFixture(t)
parent := globalActiveCred
if parentType == "user" {
parent = iamAdminUser(ctx, t, sys, "empty-policy-parent", "readwrite")
}
sp, err := policy.ParseConfig(strings.NewReader(document))
mustIAM(t, err)
caller, _, err := sys.NewServiceAccount(ctx, parent.AccessKey, nil, newServiceAccountOpts{
accessKey: "empty-policy-caller", secretKey: "empty-policy-caller-secret", sessionPolicy: sp,
claims: map[string]any{"identity-attribute": "preserved"},
})
mustIAM(t, err)
const limitedPolicy = `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:ListServiceAccounts"]}]}`
for _, tc := range []struct {
policy string
listStatus int
allowObject bool
}{
{"", http.StatusOK, true},
{"null", http.StatusOK, true},
{"{}", http.StatusOK, true},
{`{"Statement":null}`, http.StatusOK, true},
{limitedPolicy, http.StatusOK, false},
{`{"Version":"2012-10-17"}`, http.StatusForbidden, false},
{`{"Statement":[]}`, http.StatusForbidden, false},
} {
expires := UTCNow().Add(time.Hour)
req := madmin.AddServiceAccountReq{Expiration: &expires}
if tc.policy != "" {
req.Policy = json.RawMessage(tc.policy)
}
api := adminAPIHandlers{}
rec := iamAdminCall(t, caller, "PUT", "/minio/admin/v3/add-service-account", req, nil, api.AddServiceAccount)
assertIAMAdminStatus(t, "create expiring child", rec, http.StatusOK)
data, err := madmin.DecryptData(caller.SecretKey, rec.Body)
mustIAM(t, err)
var response madmin.AddServiceAccountResp
mustIAM(t, json.Unmarshal(data, &response))
child, ok := sys.store.GetUser(response.Credentials.AccessKey)
if !ok {
t.Fatal("new service account missing")
}
// Exercise the issued key through authentication and authorization.
assertIAMAdminStatus(t, "use expiring child policy="+tc.policy, iamAdminCall(t, child.Credentials, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), tc.listStatus)
signed := iamAdminRequest(t, child.Credentials, "GET", "/example/object", nil)
checked, owner, code := checkKeyValid(signed, child.Credentials.AccessKey)
if code != ErrNone {
t.Fatalf("child authentication: %v", code)
}
if checked.Claims["identity-attribute"] != "preserved" {
t.Fatal("creation lost the caller's identity context")
}
allowed := sys.IsAllowed(policy.Args{AccountName: checked.AccessKey, Claims: checked.Claims, IsOwner: owner, Action: policy.GetObjectAction, BucketName: "example", ObjectName: "object"})
if allowed != tc.allowObject {
t.Errorf("GetObject allowed=%v with requested policy=%s", allowed, tc.policy)
}
}
})
}
}
}
func TestIAMAdminCorruptServiceCleanup(t *testing.T) {
for _, status := range []string{auth.AccountOn, auth.AccountOff, "expired"} {
t.Run(status, func(t *testing.T) {
ctx, sys := iamAdminFixture(t)
caller := iamAdminUser(ctx, t, sys, "review-admin", "consoleAdmin")
parent := iamAdminUser(ctx, t, sys, "review-parent", "readwrite")
svc := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "rv-corrupt", "")
if status == auth.AccountOff {
_, err := sys.UpdateServiceAccount(ctx, svc.AccessKey, updateServiceAccountOpts{status: status})
mustIAM(t, err)
}
cache := sys.store.lock()
u := cache.iamUsersMap[svc.AccessKey]
u.Credentials.SessionToken = "broken-session-token"
if status == "expired" {
u.Credentials.Expiration = UTCNow().Add(-time.Minute)
}
cache.iamUsersMap[svc.AccessKey] = u
sys.store.unlock()
_, _, err := sys.GetServiceAccount(ctx, svc.AccessKey)
if err == nil {
t.Fatal("fault injection failed")
}
raw, exists := sys.store.GetUser(svc.AccessKey)
_, loginOK := sys.GetUser(ctx, svc.AccessKey)
t.Logf("metadata_exists=%v parent_known=%v GetUser_ok=%v", exists, raw.Credentials.ParentUser == parent.AccessKey, loginOK)
api := adminAPIHandlers{}
outsider := iamAdminUser(ctx, t, sys, "cleanup-outsider", "readwrite")
assertIAMAdminStatus(t, "unauthorized cleanup", iamAdminCall(t, outsider, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.DeleteServiceAccount), 404)
assertIAMAdminStatus(t, "admin cleanup", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+svc.AccessKey, nil, map[string]string{"accessKey": svc.AccessKey}, api.DeleteServiceAccount), 204)
if _, ok := sys.store.GetUser(svc.AccessKey); ok {
t.Fatal("deleted service account remains in the IAM cache")
}
})
}
}
func TestIAMAdminPolicylessSelf(t *testing.T) {
ctx, sys := iamAdminFixture(t)
caller := iamAdminUser(ctx, t, sys, "review-policyless", "")
svc := iamAdminServiceAccount(ctx, t, sys, caller.AccessKey, "rv-policyless-child", "")
api := adminAPIHandlers{}
assertIAMAdminStatus(t, "list", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), 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, "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, "revoke", iamAdminCall(t, caller, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true", nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 204)
assertIAMAdminStatus(t, "list another user", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts?user="+globalActiveCred.AccessKey, nil, nil, api.ListServiceAccounts), 403)
assertIAMAdminStatus(t, "revoke another user", iamAdminCall(t, caller, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true&user="+globalActiveCred.AccessKey, nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 403)
}
func TestIAMAdminExpiringInheritedPolicy(t *testing.T) {
ctx, sys := iamAdminFixture(t)
parent := iamAdminUser(ctx, t, sys, "review-legacy-parent", "consoleAdmin")
document := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:CreateServiceAccount","admin:UpdateServiceAccount"]}]}`
// Model a credential persisted by the old handler: it copied its caller's
// policy claims but set the newly created service's type to inherited.
expires := UTCNow().Add(time.Hour)
caller, _, err := sys.NewServiceAccount(ctx, parent.AccessKey, nil, newServiceAccountOpts{
accessKey: "rv-exp-inherited", secretKey: "review-service-valid-secret", expiration: &expires,
claims: map[string]any{policy.SessionPolicyName: base64.StdEncoding.EncodeToString([]byte(document))},
})
mustIAM(t, err)
req := iamAdminRequest(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil)
checked, owner, code := checkKeyValid(req, caller.AccessKey)
if code != ErrNone {
t.Fatalf("checkKeyValid: %v", code)
}
args := policy.Args{AccountName: caller.AccessKey, Claims: checked.Claims, IsOwner: owner, Action: policy.CreateServiceAccountAdminAction, DenyOnly: true}
if !sys.IsAllowed(args) {
t.Fatal("fixture cannot create")
}
args.Action = policy.CreateUserAdminAction
args.DenyOnly = false
if sys.IsAllowed(args) {
t.Fatal("fixture not constrained")
}
t.Logf("service=%v temp=%v inherited=%v; create allowed, create-user denied", checked.IsServiceAccount(), checked.IsTemp(), checked.Claims[iamPolicyClaimNameSA()] == inheritedPolicyType)
api := adminAPIHandlers{}
rec := iamAdminCall(t, caller, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{AccessKey: "review-escaped", SecretKey: "review-service-valid-secret"}, nil, api.AddServiceAccount)
if rec.Code == 200 {
child, ok := sys.GetUser(ctx, "review-escaped")
if !ok {
t.Fatal("new key missing")
}
childReq := iamAdminRequest(t, child.Credentials, "GET", "/minio/admin/v3/list-service-accounts", nil)
childChecked, childOwner, childCode := checkKeyValid(childReq, child.Credentials.AccessKey)
if childCode != ErrNone {
t.Fatalf("child auth=%v", childCode)
}
allowed := sys.IsAllowed(policy.Args{AccountName: childChecked.AccessKey, Claims: childChecked.Claims, IsOwner: childOwner, Action: policy.CreateUserAdminAction})
t.Errorf("constrained inherited service created a new key; new key CreateUser allowed=%v", allowed)
} else {
assertIAMAdminStatus(t, "reject derived creation", rec, 403)
}
updated := iamAdminCall(t, caller, "POST", "/minio/admin/v3/update-service-account?accessKey="+caller.AccessKey, madmin.UpdateServiceAccountReq{NewName: "metadata-only"}, map[string]string{"accessKey": caller.AccessKey}, api.UpdateServiceAccount)
if updated.Code == 204 {
after, ok := sys.GetUser(ctx, caller.AccessKey)
if !ok {
t.Fatal("updated key missing")
}
afterReq := iamAdminRequest(t, after.Credentials, "GET", "/minio/admin/v3/list-service-accounts", nil)
afterChecked, afterOwner, afterCode := checkKeyValid(afterReq, caller.AccessKey)
if afterCode != ErrNone {
t.Fatalf("updated auth=%v", afterCode)
}
allowed := sys.IsAllowed(policy.Args{AccountName: caller.AccessKey, Claims: afterChecked.Claims, IsOwner: afterOwner, Action: policy.CreateUserAdminAction})
t.Errorf("constrained inherited service updated itself; CreateUser allowed after update=%v", allowed)
} else {
assertIAMAdminStatus(t, "reject inherited self update", updated, 403)
}
}
func TestIAMAdminSessionPolicyCannotDelegate(t *testing.T) {
for _, temp := range []bool{false, true} {
t.Run(map[bool]string{false: "service", true: "sts"}[temp], func(t *testing.T) {
ctx, sys := iamAdminFixture(t)
parent := iamAdminUser(ctx, t, sys, "rv-parent", "consoleAdmin")
doc := `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:CreateServiceAccount","admin:UpdateServiceAccount"]}]}`
var caller auth.Credentials
if !temp {
caller = iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "review-restricted", doc)
} else {
var err error
caller, err = auth.GetNewCredentialsWithMetadata(map[string]any{parentClaim: parent.AccessKey, "exp": UTCNow().Add(time.Hour).Unix(), policy.SessionPolicyName: base64.StdEncoding.EncodeToString([]byte(doc))}, globalActiveCred.SecretKey)
mustIAM(t, err)
caller.ParentUser = parent.AccessKey
_, err = sys.SetTempUser(ctx, caller.AccessKey, caller, "")
mustIAM(t, err)
}
sibling := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "rv-sibling", "")
api := adminAPIHandlers{}
assertIAMAdminStatus(t, "restrict create", iamAdminCall(t, caller, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{AccessKey: "review-bad-new", SecretKey: "review-service-valid-secret"}, nil, api.AddServiceAccount), 403)
assertIAMAdminStatus(t, "restrict LDAP create", iamAdminCall(t, caller, "PUT", "/minio/admin/v3/idp/ldap/add-service-account", madmin.AddServiceAccountReq{AccessKey: "review-bad-new", SecretKey: "review-service-valid-secret"}, nil, api.AddServiceAccountLDAP), 403)
assertIAMAdminStatus(t, "restrict update", iamAdminCall(t, caller, "POST", "/minio/admin/v3/update-service-account?accessKey="+sibling.AccessKey, madmin.UpdateServiceAccountReq{NewSecretKey: "review-known-new-secret"}, map[string]string{"accessKey": sibling.AccessKey}, api.UpdateServiceAccount), 403)
if _, ok := sys.store.GetUser("review-bad-new"); ok {
t.Fatal("denied creation persisted a credential")
}
u, ok := sys.store.GetUser(sibling.AccessKey)
if !ok || u.Credentials.SecretKey != sibling.SecretKey {
t.Fatal("denied update changed the target credential")
}
})
}
}
func TestIAMAdminSelfDenials(t *testing.T) {
for _, mode := range []string{"parent_deny", "group_deny", "session_implicit_deny", "session_explicit_deny", "plugin_deny"} {
t.Run(mode, func(t *testing.T) {
ctx, sys := iamAdminFixture(t)
parent := iamAdminUser(ctx, t, sys, "rv-self-parent", "readwrite")
target := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "rv-self-target", "")
caller := parent
doc := `{"Version":"2012-10-17","Statement":[{"Effect":"Deny","Action":["admin:ListServiceAccounts","admin:RemoveServiceAccount"]}]}`
switch mode {
case "parent_deny", "group_deny":
p, err := policy.ParseConfig(strings.NewReader(doc))
mustIAM(t, err)
_, err = sys.SetPolicy(ctx, "review-self-deny", *p)
mustIAM(t, err)
if mode == "group_deny" {
_, err = sys.AddUsersToGroup(ctx, "self-deny-group", []string{parent.AccessKey})
mustIAM(t, err)
_, err = sys.PolicyDBSet(ctx, "self-deny-group", "review-self-deny", regUser, true)
} else {
_, err = sys.PolicyDBSet(ctx, parent.AccessKey, "readwrite,review-self-deny", regUser, false)
}
mustIAM(t, err)
case "session_implicit_deny":
caller = iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "rv-self-caller", `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["s3:GetObject"],"Resource":["arn:aws:s3:::example/*"]}]}`)
case "session_explicit_deny":
caller = iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "rv-self-caller", doc)
case "plugin_deny":
iamAdminPlugin(t, false)
}
api := adminAPIHandlers{}
assertIAMAdminStatus(t, "list", iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts", nil, nil, api.ListServiceAccounts), 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, "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, "revoke", iamAdminCall(t, caller, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true", nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 403)
})
}
}
func TestIAMAdminNonRootCannotManageRoot(t *testing.T) {
ctx, sys := iamAdminFixture(t)
caller := iamAdminUser(ctx, t, sys, "rv-console-admin", "consoleAdmin")
target := iamAdminServiceAccount(ctx, t, sys, globalActiveCred.AccessKey, "rv-root-target", "")
api := adminAPIHandlers{}
assertIAMAdminStatus(t, "create root", iamAdminCall(t, caller, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{AccessKey: "rv-new-root", SecretKey: "review-service-valid-secret", TargetUser: globalActiveCred.AccessKey}, nil, api.AddServiceAccount), 403)
assertIAMAdminStatus(t, "update root", iamAdminCall(t, caller, "POST", "/minio/admin/v3/update-service-account?accessKey="+target.AccessKey, madmin.UpdateServiceAccountReq{NewSecretKey: "review-known-new-secret"}, map[string]string{"accessKey": target.AccessKey}, api.UpdateServiceAccount), 403)
assertIAMAdminStatus(t, "delete root", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.DeleteServiceAccount), 404)
assertIAMAdminStatus(t, "revoke root", iamAdminCall(t, caller, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true&user="+globalActiveCred.AccessKey, nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 403)
// Restricted root-derived credentials and the site replicator are not root
// administrators, even when their effective policy allows these actions.
limited := iamAdminServiceAccount(ctx, t, sys, globalActiveCred.AccessKey, "limited-root", `{"Version":"2012-10-17","Statement":[{"Effect":"Allow","Action":["admin:*"]}]}`)
assertIAMAdminStatus(t, "limited root delete", iamAdminCall(t, limited, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.DeleteServiceAccount), 404)
assertIAMAdminStatus(t, "limited root revoke", iamAdminCall(t, limited, "POST", "/minio/admin/v3/revoke-tokens/builtin?fullRevoke=true", nil, map[string]string{"userProvider": madmin.BuiltinProvider}, api.RevokeTokens), 403)
replicator, _, err := sys.NewServiceAccount(ctx, globalActiveCred.AccessKey, nil, newServiceAccountOpts{accessKey: siteReplicatorSvcAcc, secretKey: "replicator-valid-secret", allowSiteReplicatorAccount: true})
mustIAM(t, err)
assertIAMAdminStatus(t, "replicator create root", iamAdminCall(t, replicator, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{}, nil, api.AddServiceAccount), 403)
// An external plugin remains authoritative, including for root targets.
iamAdminPlugin(t, true)
assertIAMAdminStatus(t, "plugin update root", iamAdminCall(t, limited, "POST", "/minio/admin/v3/update-service-account?accessKey="+target.AccessKey, madmin.UpdateServiceAccountReq{NewName: "plugin-authorized"}, map[string]string{"accessKey": target.AccessKey}, api.UpdateServiceAccount), 204)
assertIAMAdminStatus(t, "plugin create root", iamAdminCall(t, limited, "PUT", "/minio/admin/v3/add-service-account", madmin.AddServiceAccountReq{}, nil, api.AddServiceAccount), 200)
}
func TestIAMAdminDeleteCorruptServiceReplicates(t *testing.T) {
ctx, sys := iamAdminFixture(t)
caller := iamAdminUser(ctx, t, sys, "cleanup-admin", "consoleAdmin")
parent := iamAdminUser(ctx, t, sys, "cleanup-parent", "readwrite")
target := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "cleanup-target", "")
cache := sys.store.lock()
u := cache.iamUsersMap[target.AccessKey]
u.Credentials.SessionToken = "broken-session-token"
u.Credentials.Status = auth.AccountOff
cache.iamUsersMap[target.AccessKey] = u
sys.store.unlock()
var deletions atomic.Int32
peer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.URL.Path == "/minio/health/live" {
return
}
var batch iamRevisionBatch
if err := json.NewDecoder(r.Body).Decode(&batch); err != nil {
t.Error(err)
w.WriteHeader(http.StatusBadRequest)
return
}
if len(batch.Items) != 1 || batch.Items[0].SvcAccChange == nil || batch.Items[0].SvcAccChange.Delete == nil || batch.Items[0].SvcAccChange.Delete.AccessKey != target.AccessKey {
t.Error("replication did not carry the service account deletion")
} else {
deletions.Add(1)
}
_ = json.NewEncoder(w).Encode(iamRevisionResponse{iamRevisionStatus: iamRevisionStatus{Version: iamRevisionProtocol, Node: "peer", Instance: "peer-boot", Digest: "deleted"}})
}))
t.Cleanup(peer.Close)
globalSiteReplicatorCred.RLock()
previousSecret := globalSiteReplicatorCred.secretKey
globalSiteReplicatorCred.RUnlock()
globalSiteReplicatorCred.Set(caller.SecretKey)
t.Cleanup(func() { globalSiteReplicatorCred.Set(previousSecret) })
globalSiteReplicationSys.Lock()
globalSiteReplicationSys.enabled = true
globalSiteReplicationSys.state = srState{ServiceAccountAccessKey: caller.AccessKey, Peers: map[string]madmin.PeerInfo{
globalDeploymentID(): {DeploymentID: globalDeploymentID(), Name: "local"},
"peer": {DeploymentID: "peer", Name: "peer", Endpoint: peer.URL},
}}
globalSiteReplicationSys.Unlock()
t.Cleanup(func() {
globalSiteReplicationSys.Lock()
globalSiteReplicationSys.enabled = false
globalSiteReplicationSys.Unlock()
})
api := adminAPIHandlers{}
assertIAMAdminStatus(t, "replicated cleanup", iamAdminCall(t, caller, "DELETE", "/minio/admin/v3/delete-service-account?accessKey="+target.AccessKey, nil, map[string]string{"accessKey": target.AccessKey}, api.DeleteServiceAccount), 204)
if deletions.Load() != 1 {
t.Fatalf("expected one immediate deletion notification, got %d", deletions.Load())
}
}
func TestIAMAdminListSelfAliases(t *testing.T) {
for _, kind := range []string{"service", "sts"} {
t.Run(kind, func(t *testing.T) {
ctx, sys := iamAdminFixture(t)
parent := iamAdminUser(ctx, t, sys, "alias-parent", "readwrite")
target := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-target", "")
caller := iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "alias-caller", "")
if kind == "sts" {
var err error
caller, err = auth.GetNewCredentialsWithMetadata(map[string]any{parentClaim: parent.AccessKey, "exp": UTCNow().Add(time.Hour).Unix()}, globalActiveCred.SecretKey)
mustIAM(t, err)
caller.ParentUser = parent.AccessKey
_, err = sys.SetTempUser(ctx, caller.AccessKey, caller, "")
mustIAM(t, err)
}
api := adminAPIHandlers{}
for _, alias := range []string{"", caller.AccessKey, parent.AccessKey} {
rec := iamAdminCall(t, caller, "GET", "/minio/admin/v3/list-service-accounts?user="+alias, nil, nil, api.ListServiceAccounts)
assertIAMAdminStatus(t, "list self alias", rec, 200)
data, err := madmin.DecryptData(caller.SecretKey, rec.Body)
mustIAM(t, err)
var response madmin.ListServiceAccountsResp
mustIAM(t, json.Unmarshal(data, &response))
found := false
for _, account := range response.Accounts {
found = found || account.AccessKey == target.AccessKey
}
if !found {
t.Fatalf("self alias %q did not list the parent's service accounts", alias)
}
}
})
}
}
+65 -122
View File
@@ -714,10 +714,8 @@ func (a adminAPIHandlers) AddServiceAccount(w http.ResponseWriter, r *http.Reque
// In case of LDAP/OIDC we need to set `opts.claims` to ensure
// it is associated with the LDAP/OIDC user properly.
// The new credential's policy comes from the request, not the caller.
for k, v := range cred.Claims {
switch k {
case expClaim, policy.SessionPolicyName, sessionPolicyNameExtracted, iamPolicyClaimNameSA():
if k == expClaim {
continue
}
opts.claims[k] = v
@@ -857,9 +855,10 @@ func (a adminAPIHandlers) UpdateServiceAccount(w http.ResponseWriter, r *http.Re
// Permission checks:
//
// 1. Updating requires policy.UpdateServiceAccountAdminAction. Under built-in
// IAM, session-limited callers cannot update credentials, and only unrestricted
// root credentials can manage root's service accounts.
// 1. Any type of account (i.e. access keys (previously/still called service
// accounts), STS accounts, internal IDP accounts, etc) with the
// policy.UpdateServiceAccountAdminAction permission can update any service
// account.
//
// 2. We would like to let a user update their own access keys, however it
// is currently blocked pending a re-design. Users are still able to delete
@@ -875,11 +874,6 @@ func (a adminAPIHandlers) UpdateServiceAccount(w http.ResponseWriter, r *http.Re
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
return
}
if newGlobalAuthZPluginFn() == nil && (hasActiveSessionPolicy(cred) ||
(svcAccount.ParentUser == globalActiveCred.AccessKey && !canManageRootCredentials(cred))) {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
return
}
var sp *policy.Policy
if len(updateReq.NewPolicy) > 0 {
@@ -957,22 +951,24 @@ func (a adminAPIHandlers) InfoServiceAccount(w http.ResponseWriter, r *http.Requ
return
}
requestUser := cred.AccessKey
if cred.ParentUser != "" {
requestUser = cred.ParentUser
}
if !isAllowedServiceAccountAction(cred, policy.Args{
if !globalIAMSys.IsAllowed(policy.Args{
AccountName: cred.AccessKey,
Groups: cred.Groups,
Action: policy.ListServiceAccountsAdminAction,
ConditionValues: getConditionValues(r, "", cred),
IsOwner: owner,
Claims: cred.Claims,
DenyOnly: requestUser == svcAccount.ParentUser,
}) {
requestUser := cred.AccessKey
if cred.ParentUser != "" {
requestUser = cred.ParentUser
}
if requestUser != svcAccount.ParentUser {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
return
}
}
// if session policy is nil or empty, then it is implied policy
impliedPolicy := sessionPolicy == nil || (sessionPolicy.Version == "" && len(sessionPolicy.Statements) == 0)
@@ -1043,27 +1039,29 @@ func (a adminAPIHandlers) ListServiceAccounts(w http.ResponseWriter, r *http.Req
return
}
selfUser := cred.AccessKey
if cred.ParentUser != "" {
selfUser = cred.ParentUser
}
var targetAccount string
// If listing is requested for a specific user (who is not the request
// sender), check that the user has permissions.
user := r.Form.Get("user")
selfOnly := user == "" || user == cred.AccessKey || user == selfUser
if !isAllowedServiceAccountAction(cred, policy.Args{
if user != "" && user != cred.AccessKey {
if !globalIAMSys.IsAllowed(policy.Args{
AccountName: cred.AccessKey,
Groups: cred.Groups,
Action: policy.ListServiceAccountsAdminAction,
ConditionValues: getConditionValues(r, "", cred),
IsOwner: owner,
Claims: cred.Claims,
DenyOnly: selfOnly,
}) {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
return
}
targetAccount := user
if selfOnly {
targetAccount = selfUser
targetAccount = user
} else {
targetAccount = cred.AccessKey
if cred.ParentUser != "" {
targetAccount = cred.ParentUser
}
}
serviceAccounts, err := globalIAMSys.ListServiceAccounts(ctx, targetAccount)
@@ -1112,7 +1110,7 @@ func (a adminAPIHandlers) DeleteServiceAccount(w http.ResponseWriter, r *http.Re
// Get current object layer instance.
objectAPI := newObjectLayerFn()
if objectAPI == nil || globalNotificationSys == nil || !globalIAMSys.Initialized() {
if objectAPI == nil || globalNotificationSys == nil {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrServerNotInitialized), r.URL)
return
}
@@ -1133,41 +1131,35 @@ func (a adminAPIHandlers) DeleteServiceAccount(w http.ResponseWriter, r *http.Re
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrInvalidArgument), r.URL)
return
}
// Deletion needs the stored identity, not a valid token or session policy.
// Keep disabled, expired and damaged service accounts removable, and retain
// their parent for authorization and the site replication hook below.
u, ok := globalIAMSys.store.GetUser(serviceAccount)
if !ok || !u.Credentials.IsServiceAccount() {
// We do not care if service account is readable or not at this point,
// since this is a delete call we shall allow it to be deleted if possible.
svcAccount, _, err := globalIAMSys.GetServiceAccount(ctx, serviceAccount)
if errors.Is(err, errNoSuchServiceAccount) {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminServiceAccountNotFound), r.URL)
return
}
svcAccount := u.Credentials
parentUser := cred.AccessKey
if cred.ParentUser != "" {
parentUser = cred.ParentUser
}
isSelf := parentUser == svcAccount.ParentUser
if !isAllowedServiceAccountAction(cred, policy.Args{
adminPrivilege := globalIAMSys.IsAllowed(policy.Args{
AccountName: cred.AccessKey,
Groups: cred.Groups,
Action: policy.RemoveServiceAccountAdminAction,
ConditionValues: getConditionValues(r, "", cred),
IsOwner: owner,
Claims: cred.Claims,
DenyOnly: isSelf,
}) {
if !isSelf {
// Hide the existence of another user's service account.
})
if !adminPrivilege {
parentUser := cred.AccessKey
if cred.ParentUser != "" {
parentUser = cred.ParentUser
}
if svcAccount.ParentUser != "" && parentUser != svcAccount.ParentUser {
// The service account belongs to another user but return not
// found error to mitigate brute force attacks. or the
// serviceAccount doesn't exist.
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminServiceAccountNotFound), r.URL)
return
}
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
return
}
if newGlobalAuthZPluginFn() == nil && svcAccount.ParentUser == globalActiveCred.AccessKey && !canManageRootCredentials(cred) {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminServiceAccountNotFound), r.URL)
return
}
if err := globalIAMSys.DeleteServiceAccount(ctx, serviceAccount, true); err != nil {
@@ -2054,30 +2046,26 @@ func (a adminAPIHandlers) RevokeTokens(w http.ResponseWriter, r *http.Request) {
return
}
parentUser := cred.AccessKey
if cred.ParentUser != "" {
parentUser = cred.ParentUser
}
isSelfTarget := isTokenSelfRevoke || user == parentUser
if !isAllowedServiceAccountAction(cred, policy.Args{
adminPrivilege := globalIAMSys.IsAllowed(policy.Args{
AccountName: cred.AccessKey,
Groups: cred.Groups,
Action: policy.RemoveServiceAccountAdminAction,
ConditionValues: getConditionValues(r, "", cred),
IsOwner: owner,
Claims: cred.Claims,
DenyOnly: isSelfTarget,
}) {
})
if !adminPrivilege || isTokenSelfRevoke {
parentUser := cred.AccessKey
if cred.ParentUser != "" {
parentUser = cred.ParentUser
}
if !isTokenSelfRevoke && user != parentUser {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
return
}
if isSelfTarget {
user = parentUser
}
if newGlobalAuthZPluginFn() == nil && user == globalActiveCred.AccessKey && !canManageRootCredentials(cred) {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
return
}
// Infer token revoke type from the request if requestor is STS.
if isTokenSelfRevoke && tokenRevokeType == "" && !fullRevoke {
@@ -2124,24 +2112,31 @@ func (a adminAPIHandlers) InfoAccessKey(w http.ResponseWriter, r *http.Request)
u, ok := globalIAMSys.GetUser(ctx, accessKey)
targetCred := u.Credentials
requestUser := cred.AccessKey
if cred.ParentUser != "" {
requestUser = cred.ParentUser
}
isSelf := ok && targetCred.ParentUser != "" && requestUser == targetCred.ParentUser
if !isAllowedServiceAccountAction(cred, policy.Args{
if !globalIAMSys.IsAllowed(policy.Args{
AccountName: cred.AccessKey,
Groups: cred.Groups,
Action: policy.ListServiceAccountsAdminAction,
ConditionValues: getConditionValues(r, "", cred),
IsOwner: owner,
Claims: cred.Claims,
DenyOnly: isSelf,
}) {
// If requested user does not exist and requestor is not allowed to list service accounts, return access denied.
if !ok {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
return
}
requestUser := cred.AccessKey
if cred.ParentUser != "" {
requestUser = cred.ParentUser
}
if requestUser != targetCred.ParentUser {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAccessDenied), r.URL)
return
}
}
if !ok {
writeErrorResponseJSON(ctx, w, errorCodes.ToAPIErr(ErrAdminNoSuchAccessKey), r.URL)
return
@@ -2913,44 +2908,6 @@ func addExpirationToCondValues(exp *time.Time, condValues map[string][]string) e
return nil
}
// hasActiveSessionPolicy follows the same credential dispatch as IAMSys.IsAllowed.
// Expiring service accounts also satisfy IsTemp and are evaluated as STS there.
// Callers use this only after authorization under built-in IAM.
func hasActiveSessionPolicy(cred auth.Credentials) bool {
args := policy.Args{Claims: cred.Claims}
if cred.IsTemp() {
hasSP, _ := isAllowedBySessionPolicy(args)
return hasSP
}
if cred.IsServiceAccount() {
if cred.Claims[iamPolicyClaimNameSA()] == inheritedPolicyType {
return false
}
hasSP, _ := isAllowedBySessionPolicyForServiceAccount(args)
return hasSP
}
return false
}
// canManageRootCredentials is an additional guard after built-in IAM authorization.
// The signature's owner flag is insufficient: historical empty session policies
// clear it even though the service account still inherits root's permissions.
func canManageRootCredentials(cred auth.Credentials) bool {
return cred.AccessKey == globalActiveCred.AccessKey ||
(cred.ParentUser == globalActiveCred.AccessKey && cred.AccessKey != siteReplicatorSvcAcc && !hasActiveSessionPolicy(cred))
}
// isAllowedServiceAccountAction preserves regular users' self inspection and
// cleanup even without a policy attachment, while enforcing explicit denies.
// Derived credentials and external authorization plugins use full IAM evaluation.
func isAllowedServiceAccountAction(cred auth.Credentials, args policy.Args) bool {
if args.DenyOnly && !args.IsOwner && !cred.IsTemp() && !cred.IsServiceAccount() && newGlobalAuthZPluginFn() == nil {
policies, err := globalIAMSys.PolicyDBGet(args.AccountName, args.Groups...)
return err == nil && globalIAMSys.GetCombinedPolicy(policies...).IsAllowed(args)
}
return globalIAMSys.IsAllowed(args)
}
func commonAddServiceAccount(r *http.Request, ldap bool) (context.Context, auth.Credentials, newServiceAccountOpts, madmin.AddServiceAccountReq, string, APIError) {
ctx := r.Context()
@@ -3039,14 +2996,6 @@ func commonAddServiceAccount(r *http.Request, ldap bool) (context.Context, auth.
}) {
return ctx, auth.Credentials{}, newServiceAccountOpts{}, madmin.AddServiceAccountReq{}, "", errorCodes.ToAPIErr(ErrAccessDenied)
}
effectiveTarget := targetUser
if (targetUser == cred.AccessKey || targetUser == cred.ParentUser) && cred.ParentUser != "" {
effectiveTarget = cred.ParentUser
}
if newGlobalAuthZPluginFn() == nil && (hasActiveSessionPolicy(cred) ||
(effectiveTarget == globalActiveCred.AccessKey && !canManageRootCredentials(cred))) {
return ctx, auth.Credentials{}, newServiceAccountOpts{}, madmin.AddServiceAccountReq{}, "", errorCodes.ToAPIErr(ErrAccessDenied)
}
var sp *policy.Policy
if len(createReq.Policy) > 0 {
@@ -3054,12 +3003,6 @@ func commonAddServiceAccount(r *http.Request, ldap bool) (context.Context, auth.
if err != nil {
return ctx, auth.Credentials{}, newServiceAccountOpts{}, madmin.AddServiceAccountReq{}, "", toAdminAPIErr(ctx, err)
}
// Empty service policies mean inheritance, including for expiring keys.
// Normalize the replication request as well as the local credential.
if sp.Version == "" && sp.Statements == nil && sp.ID == "" {
sp = nil
createReq.Policy = nil
}
}
opts.sessionPolicy = sp
+5 -4
View File
@@ -744,10 +744,10 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
}
},
// Some disks have data for this.
partial: func(entries metaCacheEntries, errs []error) {
partial: func(entries metaCacheEntries, errs []error) error {
if !f.shouldHeal() {
cancel()
return
return nil
}
entry, ok := entries.resolve(&resolver)
if !ok {
@@ -772,7 +772,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
}
if entry.isDir() {
return
return nil
}
// We got an entry which we should be able to heal.
@@ -787,7 +787,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
scannerLogIf(ctx, err)
}
foundObjs = foundObjs || err == nil
return
return nil
}
custom["versions"] = fmt.Sprint(len(fiv.Versions))
@@ -814,6 +814,7 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
}
custom["success_versions"] = fmt.Sprint(successVersions)
custom["failed_versions"] = fmt.Sprint(failVersions)
return nil
},
// Too many disks failed.
finished: func(errs []error) {
+3 -3
View File
@@ -86,7 +86,7 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string,
cancel()
}
},
partial: func(entries metaCacheEntries, _ []error) {
partial: func(entries metaCacheEntries, _ []error) error {
entry, ok := entries.resolve(&resolver)
if !ok {
// check if we can get one entry at least
@@ -94,12 +94,12 @@ func (er erasureObjects) listAndHeal(ctx context.Context, bucket, prefix string,
entry, _ = entries.firstFound()
}
if !recursive && prefix != entry.name {
return
return nil
}
if err := healEntry(bucket, *entry, scanMode); err != nil {
cancel()
return
}
return nil
},
finished: nil,
}
+2 -1
View File
@@ -730,11 +730,12 @@ func (set *erasureObjects) listObjectsToDecommission(ctx context.Context, bi dec
minDisks: listingQuorum,
reportNotFound: false,
agreed: fn,
partial: func(entries metaCacheEntries, _ []error) {
partial: func(entries metaCacheEntries, _ []error) error {
entry, ok := entries.resolve(&resolver)
if ok {
fn(*entry)
}
return nil
},
finished: nil,
})
+2 -1
View File
@@ -525,11 +525,12 @@ func (set *erasureObjects) listObjectsToRebalance(ctx context.Context, bucketNam
minDisks: listingQuorum,
reportNotFound: false,
agreed: fn,
partial: func(entries metaCacheEntries, _ []error) {
partial: func(entries metaCacheEntries, _ []error) error {
entry, ok := entries.resolve(&resolver)
if ok {
fn(*entry)
}
return nil
},
finished: nil,
})
+7 -3
View File
@@ -2651,11 +2651,15 @@ func (z *erasureServerPools) Walk(ctx context.Context, bucket, prefix string, re
minDisks: listingQuorum,
reportNotFound: false,
agreed: send,
partial: func(entries metaCacheEntries, _ []error) {
entry, ok := entries.resolve(&resolver)
if ok {
partial: func(entries metaCacheEntries, errs []error) error {
entry, err := set.resolveListEntry(ctx, bucket, entries, errs, &resolver, z.listingNSLock(set))
if err != nil {
return err
}
if entry != nil {
send(*entry)
}
return nil
},
finished: nil,
}
+2 -1
View File
@@ -529,7 +529,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
jt.Take()
go healEntry(bucket, entry)
},
partial: func(entries metaCacheEntries, _ []error) {
partial: func(entries metaCacheEntries, _ []error) error {
entry, ok := entries.resolve(&resolver)
if !ok {
// check if we can get one entry at least
@@ -538,6 +538,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
}
jt.Take()
go healEntry(bucket, *entry)
return nil
},
finished: func(errs []error) {
success := countErrs(errs, nil)
-2
View File
@@ -24,8 +24,6 @@ func TestIAMPeerDeleteNotificationReloadsCommittedState(t *testing.T) {
resetTestGlobals()
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
// Both caches use object storage, independent of earlier etcd suites.
globalEtcdClient = nil
obj, disk, err := prepareFS(ctx)
if err != nil {
t.Fatal(err)
+14
View File
@@ -670,6 +670,12 @@ func (m *metaCacheEntriesSorted) forwardPast(s string) {
// If the context is canceled the function will return the error,
// otherwise the function will return nil.
func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<- metaCacheEntry, readQuorum int) error {
return mergeEntryChannelsWithErrors(ctx, in, out, readQuorum, nil)
}
// terminalErr is inspected after each input closes and before the merger
// advances beyond it. Callers must publish the error before closing the input.
func mergeEntryChannelsWithErrors(ctx context.Context, in []chan metaCacheEntry, out chan<- metaCacheEntry, readQuorum int, terminalErr func(int) error) error {
defer xioutil.SafeClose(out)
top := make([]*metaCacheEntry, len(in))
nDone := 0
@@ -683,6 +689,9 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
return ctx.Err()
case v, ok := <-in[0]:
if !ok {
if terminalErr != nil {
return terminalErr(0)
}
return nil
}
select {
@@ -700,6 +709,11 @@ func mergeEntryChannels(ctx context.Context, in []chan metaCacheEntry, out chan<
return ctx.Err()
case entry, ok := <-in[idx]:
if !ok {
if terminalErr != nil {
if err := terminalErr(idx); err != nil {
return err
}
}
top[idx] = nil
nDone++
} else {
+326
View File
@@ -0,0 +1,326 @@
// Copyright (c) 2026 Feng Ruohang
//
// This file is part of Silo Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"bytes"
"context"
"encoding/xml"
"net/http"
"net/http/httptest"
"net/url"
"slices"
"sync"
"sync/atomic"
"testing"
"time"
"github.com/minio/minio/internal/grid"
)
type listingReadFailDisk struct {
StorageAPI
object string
reads *atomic.Int64
}
func (d *listingReadFailDisk) ReadXL(ctx context.Context, bucket, object string, readData bool) (RawFileInfo, error) {
if object == d.object {
d.reads.Add(1)
return RawFileInfo{}, errDiskNotFound
}
return d.StorageAPI.ReadXL(ctx, bucket, object, readData)
}
// Install before issuing requests: cache writers can outlive their HTTP
// response, so disk faults must be switched without replacing getDisks.
// Each published slice must remain immutable.
func installListingDiskSwitch(set *erasureObjects) ([]StorageAPI, func([]StorageAPI)) {
disks := set.getDisks()
var current atomic.Value
current.Store(disks)
set.getDisks = func() []StorageAPI { return current.Load().([]StorageAPI) }
return disks, func(disks []StorageAPI) { current.Store(disks) }
}
func TestListingCachePreservesFallbackError(t *testing.T) {
for _, continuation := range []bool{false, true} {
name := "cold"
if continuation {
name = "continuation"
}
t.Run(name, func(t *testing.T) {
testListingCacheFallbackError(t, continuation, false)
})
}
}
func TestListingCacheCleanupKeepsOriginalID(t *testing.T) {
testListingCacheFallbackError(t, true, true)
}
// Use actual peer RPCs to verify the abandoned cache is updated after the
// caller clears its own listing ID. The local-only path cannot observe this.
func recordListingPeerCleanup(t *testing.T) func() {
t.Helper()
tg, err := grid.SetupTestGrid(2)
if err != nil {
t.Fatal(err)
}
previous := globalNotificationSys
t.Cleanup(func() {
tg.Cleanup()
globalNotificationSys = previous
})
getIDs := make(chan string, 8)
type cleanupResult struct {
id string
err *grid.RemoteErr
}
updated := make(chan cleanupResult, 1)
server := &peerRESTServer{}
if err := getMetacacheListingRPC.Register(tg.Managers[1], func(o *listPathOptions) (*metacache, *grid.RemoteErr) {
getIDs <- o.ID
return server.GetMetacacheListingHandler(o)
}); err != nil {
t.Fatal(err)
}
if err := updateMetacacheListingRPC.Register(tg.Managers[1], func(m *metacache) (*metacache, *grid.RemoteErr) {
result, err := server.UpdateMetacacheListingHandler(m)
if m.error == "no longer used" {
updated <- cleanupResult{id: m.id, err: err}
}
return result, err
}); err != nil {
t.Fatal(err)
}
client := &peerRESTClient{gridConn: func() *grid.Connection { return tg.Managers[0].Connection(tg.Hosts[1]) }}
globalNotificationSys = &NotificationSys{peerClients: []*peerRESTClient{client}, allPeerClients: []*peerRESTClient{client}}
return func() {
ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second)
defer cancel()
var ids []string
for range 2 {
select {
case id := <-getIDs:
ids = append(ids, id)
case <-ctx.Done():
t.Fatal("peer did not receive the failed cache cleanup lookup")
}
}
if ids[0] == "" || ids[1] != ids[0] {
t.Fatalf("cleanup lookup lost the original cache ID: %q", ids)
}
select {
case result := <-updated:
if result.err != nil || result.id != ids[0] {
t.Fatalf("cleanup update: id=%q err=%v, want id=%q", result.id, result.err, ids[0])
}
t.Logf("peer cleanup Get/Update retained cache ID %q", result.id)
case <-ctx.Done():
t.Fatal("peer did not finish the failed cache cleanup update")
}
}
}
func testListingCacheFallbackError(t *testing.T, continuation, peer bool) {
t.Helper()
z, bucket, router := nullQuorumBackend(t)
var waitPeerCleanup func()
if peer {
waitPeerCleanup = recordListingPeerCleanup(t)
}
set := z.serverPools[0].sets[0]
disks, switchDisks := installListingDiskSwitch(set)
body := bytes.Repeat([]byte{'a'}, 8192)
mtime := time.Now().UTC().Add(-time.Hour)
for _, object := range []string{"a", "b", "z"} {
nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime})
}
old := make([][]byte, 3)
for i := range old {
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, "z")
}
nullQuorumPut(t, z, bucket, "z", body, ObjectOptions{MTime: mtime.Add(time.Minute)})
var token string
if continuation {
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1", "", "", ""))
var list ListObjectsV2Response
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil || !list.IsTruncated || list.NextContinuationToken == "" {
t.Fatalf("first page: %d %s", rec.Code, rec.Body.String())
}
token = list.NextContinuationToken
}
for i := range old {
if err := disks[i].WriteAll(t.Context(), bucket, "z/"+xlStorageFormatFile, old[i]); err != nil {
t.Fatal(err)
}
}
var reads atomic.Int64
walkers := slices.Clone(disks)
for i := range walkers {
walkers[i] = &listingReadFailDisk{StorageAPI: disks[i], object: "z", reads: &reads}
if i >= 8 {
walkers[i] = &failedMetadataWalkDisk{StorageAPI: walkers[i]}
}
}
switchDisks(walkers)
target := getListObjectsV2URL("", bucket, "", "1000", "", "", "")
if token != "" {
target += "&continuation-token=" + url.QueryEscape(token)
}
rec := nullQuorumRequest(t, router, http.MethodGet, target)
t.Logf("LIST=%d failed metadata reads=%d body=%s", rec.Code, reads.Load(), rec.Body.String())
if reads.Load() == 0 {
t.Fatal("fallback not exercised")
}
var apiErr APIErrorResponse
if rec.Code != http.StatusServiceUnavailable || xml.Unmarshal(rec.Body.Bytes(), &apiErr) != nil || apiErr.Code != "SlowDownRead" {
t.Fatalf("fallback failure lost its read-quorum error: %d %s", rec.Code, rec.Body.String())
}
if waitPeerCleanup != nil {
waitPeerCleanup()
}
if continuation {
caches, _ := localMetacacheMgr.getBucket(t.Context(), bucket).cloneCaches()
if len(caches) == 0 {
t.Fatal("continuation did not create a cache")
}
for _, cache := range caches {
if cache.status != scanStateError {
t.Fatalf("incomplete cache was not marked failed: %+v", cache)
}
}
}
switchDisks(disks)
retry := nullQuorumRequest(t, router, http.MethodGet, target)
var recovered ListObjectsV2Response
if retry.Code != http.StatusOK || xml.Unmarshal(retry.Body.Bytes(), &recovered) != nil {
t.Fatalf("retry after metadata recovers: %d %s", retry.Code, retry.Body.String())
}
var keys []string
for _, entry := range recovered.Contents {
keys = append(keys, entry.Key)
}
want := []string{"a", "b", "z"}
if continuation {
want = []string{"b", "z"}
}
if !slices.Equal(keys, want) || recovered.IsTruncated {
t.Fatalf("retry returned incomplete result: %+v", recovered)
}
}
func TestListingCachePagination(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
for _, key := range []string{"a", "b", "c", "d"} {
nullQuorumPut(t, z, bucket, key, []byte(key), ObjectOptions{})
}
var keys []string
var token string
var finished bool
for page := 0; page < 8; page++ {
target := getListObjectsV2URL("", bucket, "", "1", "", "", "")
if token != "" {
target += "&continuation-token=" + url.QueryEscape(token)
}
rec := nullQuorumRequest(t, router, http.MethodGet, target)
var list ListObjectsV2Response
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
t.Fatalf("page %d: %d %s", page, rec.Code, rec.Body.String())
}
for _, entry := range list.Contents {
keys = append(keys, entry.Key)
}
if !list.IsTruncated {
finished = true
break
}
if list.NextContinuationToken == "" || list.NextContinuationToken == token {
t.Fatal("pagination did not advance")
}
token = list.NextContinuationToken
}
if !finished || !slices.Equal(keys, []string{"a", "b", "c", "d"}) {
t.Fatalf("pagination returned %v, finished=%v", keys, finished)
}
}
// A cached full page must remain streamable even if a later key is slow. The
// completion barrier belongs at EOF, not before every page response.
func TestListingCacheFullPageDoesNotWaitForTail(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
disks, switchDisks := installListingDiskSwitch(z.serverPools[0].sets[0])
for _, key := range []string{"a", "b", "c", "d", "e", "zz"} {
nullQuorumPut(t, z, bucket, key, []byte(key), ObjectOptions{})
}
first := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1", "", "", ""))
var page ListObjectsV2Response
if first.Code != http.StatusOK || xml.Unmarshal(first.Body.Bytes(), &page) != nil || !page.IsTruncated {
t.Fatalf("first page: %d %s", first.Code, first.Body.String())
}
resume := make(chan struct{})
release := sync.OnceFunc(func() { close(resume) })
defer release()
var blocked, active atomic.Int64
walkers := slices.Clone(disks)
for i, disk := range walkers {
walkers[i] = &listingPausedTailDisk{StorageAPI: disk, bucket: bucket, tail: "zz", resume: resume, blocked: &blocked, active: &active}
}
switchDisks(walkers)
ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
defer cancel()
target := getListObjectsV2URL("", bucket, "", "1", "", "", "") + "&continuation-token=" + url.QueryEscape(page.NextContinuationToken)
req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
if err != nil {
t.Fatal(err)
}
response := make(chan *httptest.ResponseRecorder, 1)
go func() {
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req.WithContext(ctx))
response <- rec
}()
select {
case rec := <-response:
page = ListObjectsV2Response{}
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil || !page.IsTruncated || len(page.Contents) != 1 || page.Contents[0].Key != "b" || page.NextContinuationToken == "" {
t.Fatalf("full page before tail: %d %s", rec.Code, rec.Body.String())
}
if blocked.Load() == 0 || active.Load() == 0 {
t.Fatal("full page did not overlap the paused tail")
}
case <-ctx.Done():
t.Fatal("full page waited for the paused tail")
}
release()
waitForListingCondition(t, "disk walkers did not finish", func() bool { return active.Load() == 0 })
target = getListObjectsV2URL("", bucket, "", "1000", "", "", "") + "&continuation-token=" + url.QueryEscape(page.NextContinuationToken)
rec := nullQuorumRequest(t, router, http.MethodGet, target)
page = ListObjectsV2Response{}
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil || page.IsTruncated {
t.Fatalf("page after tail resumes: %d %s", rec.Code, rec.Body.String())
}
var keys []string
for _, entry := range page.Contents {
keys = append(keys, entry.Key)
}
if !slices.Equal(keys, []string{"c", "d", "e", "zz"}) {
t.Fatalf("cached remainder lost keys: %v", keys)
}
}
+133
View File
@@ -0,0 +1,133 @@
// Copyright (c) 2026 Feng Ruohang
//
// This file is part of Silo Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"bytes"
"encoding/xml"
"fmt"
"net/http"
"slices"
"testing"
"time"
)
func TestListFallbackPreservesPoolDeleteMarker(t *testing.T) {
for _, markerPool := range []int{0, 1} {
t.Run(fmt.Sprint(markerPool), func(t *testing.T) {
z, bucket, router := listingPoolsBackend(t)
ctx := t.Context()
live, deleted := z.serverPools[1-markerPool].sets[0], z.serverPools[markerPool].sets[0]
disks, switchDisks := installListingDiskSwitch(deleted)
const object = "gone"
body := bytes.Repeat([]byte{'a'}, 8192)
oldTime := time.Now().UTC().Add(-time.Hour)
for _, set := range []*erasureObjects{live, deleted} {
_, err := set.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: oldTime, VersionSuspended: true})
if err != nil {
t.Fatal(err)
}
}
old := make([][]byte, 3)
for i := range old {
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
}
_, err := deleted.DeleteObject(ctx, bucket, object, ObjectOptions{VersionSuspended: true, DeleteMarker: true, MTime: oldTime.Add(time.Minute)})
if err != nil {
t.Fatal(err)
}
get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object))
if get.Code != http.StatusNotFound {
t.Fatalf("expected deleted GET, got %d %s", get.Code, get.Body.String())
}
first := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
var page ListObjectsV2Response
if first.Code != http.StatusOK || xml.Unmarshal(first.Body.Bytes(), &page) != nil || len(page.Contents) != 0 {
t.Fatalf("normal LIST marker: %d %s", first.Code, first.Body.String())
}
for i := range old {
if err := disks[i].WriteAll(ctx, bucket, object+"/"+xlStorageFormatFile, old[i]); err != nil {
t.Fatal(err)
}
}
walkers := slices.Clone(disks)
for i := 8; i < len(walkers); i++ {
walkers[i] = &failedMetadataWalkDisk{StorageAPI: disks[i]}
}
switchDisks(walkers)
fi, _, _, err := deleted.getObjectFileInfo(ctx, bucket, object, ObjectOptions{}, false)
if err != nil || !fi.Deleted {
t.Fatalf("read fallback marker: %+v %v", fi, err)
}
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
page = ListObjectsV2Response{}
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &page) != nil || len(page.Contents) != 0 {
t.Fatalf("fallback resurrected old pool copy: %d %s", rec.Code, rec.Body.String())
}
})
}
}
// The regular metadata reader can translate a persistent generation split to
// ObjectNotFound even though every drive still has valid metadata. LIST must
// preserve that uncertainty instead of treating the key as deleted.
func TestListFallbackRejectsSplitGenerations(t *testing.T) {
z, bucket, _ := nullQuorumBackend(t)
set := z.serverPools[0].sets[0]
disks := set.getDisks()
const object = "object"
mtime := time.Now().UTC().Add(-time.Hour)
var generations [3][][]byte
for generation := range generations {
body := bytes.Repeat([]byte{byte('a' + generation)}, 8192)
nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime.Add(time.Duration(generation) * time.Minute)})
generations[generation] = make([][]byte, len(disks))
for i, disk := range disks {
generations[generation][i] = mustReadNullQuorumMeta(t, disk, bucket, object)
}
}
// All sixteen drives are online, but the 7/5/4 split leaves no generation
// with the eight matching snapshots required by the listing resolver.
for i, disk := range disks {
generation := 0
if i >= 7 {
generation = 1
}
if i >= 12 {
generation = 2
}
if err := disk.WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, generations[generation][i]); err != nil {
t.Fatal(err)
}
}
if _, _, _, err := set.getObjectFileInfo(t.Context(), bucket, object, ObjectOptions{}, false); !isErrObjectNotFound(err) {
t.Fatalf("split metadata did not exercise ambiguous not-found: %v", err)
}
entries := make(metaCacheEntries, len(disks))
for i, disk := range disks {
entries[i] = metaCacheEntry{name: object, metadata: mustReadNullQuorumMeta(t, disk, bucket, object)}
if _, err := entries[i].xlmeta(); err != nil {
t.Fatalf("drive %d no longer has valid metadata: %v", i, err)
}
}
resolver := metadataResolutionParams{dirQuorum: 8, objQuorum: 8, bucket: bucket, requestedVersions: 1}
entry, err := set.resolveListEntry(t.Context(), bucket, entries, nil, &resolver, z.listingNSLock(set))
if entry != nil || !isErrReadQuorum(err) {
t.Fatalf("split live metadata became absence: entry=%v err=%v", entry, err)
}
}
+101
View File
@@ -0,0 +1,101 @@
// Copyright (c) 2026 Feng Ruohang
//
// This file is part of Silo Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"bytes"
"encoding/xml"
"net/http"
"net/http/httptest"
"slices"
"testing"
"time"
)
// Exercise the actual LIST pipeline with the disputed entry in pool 1.
func TestListFallbackMultiPoolHTTPReadLock(t *testing.T) {
for _, object := range []string{"object", "dir/"} {
t.Run(object, func(t *testing.T) {
z, bucket, router := listingPoolsBackend(t)
ctx := t.Context()
set := z.serverPools[1].sets[0]
disks := set.getDisks()
encoded := encodeDirObject(object)
body := bytes.Repeat([]byte{'a'}, 8192)
if object == "dir/" {
body = nil
}
mtime := time.Now().UTC().Add(-time.Hour)
put := func(tm time.Time) {
_, err := set.PutObject(ctx, bucket, encoded, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm})
if err != nil {
t.Fatal(err)
}
}
put(mtime)
old := make([][]byte, 3)
for i := range old {
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, encoded)
}
put(mtime.Add(time.Minute))
for i := range old {
if err := disks[i].WriteAll(ctx, bucket, encoded+"/"+xlStorageFormatFile, old[i]); err != nil {
t.Fatal(err)
}
}
walkers := slices.Clone(disks)
for i := 8; i < len(walkers); i++ {
walkers[i] = &failedMetadataWalkDisk{StorageAPI: disks[i]}
}
set.getDisks = func() []StorageAPI { return walkers }
lock := z.NewNSLock(bucket, encoded)
lkctx, err := lock.GetLock(ctx, globalOperationTimeout)
if err != nil {
t.Fatal(err)
}
locked := true
defer func() {
if locked {
lock.Unlock(lkctx)
}
}()
target := getListObjectsV2URL("", bucket, "", "1000", "", "", "")
req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
if err != nil {
t.Fatal(err)
}
rec := httptest.NewRecorder()
done := make(chan struct{})
go func() {
defer close(done)
router.ServeHTTP(rec, req.WithContext(ctx))
}()
waitForListingReadLock(t, z.serverPools[0].sets[0].nsMutex, bucket, encoded, done)
lock.Unlock(lkctx)
locked = false
<-done
var list ListObjectsV2Response
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
t.Fatalf("LIST HTTP=%d body=%s", rec.Code, rec.Body.String())
}
if len(list.Contents) != 1 || list.Contents[0].Key != object {
t.Fatalf("LIST lost key: %+v", list)
}
})
}
}
+84
View File
@@ -0,0 +1,84 @@
// Copyright (c) 2026 Feng Ruohang
//
// This file is part of Silo Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"context"
"testing"
"time"
"github.com/minio/minio/internal/dsync"
)
// Use the real distributed-lock algorithm with in-process lock servers.
// This is a protocol-level check, not a network or rolling-restart test.
func TestListFallbackDistributedLockRoute(t *testing.T) {
for _, tc := range []struct {
name string
multiPool, sharedLockers bool
}{
{"multi-pool-shared-lockers", true, true},
{"multi-pool-disjoint-lockers", true, false},
{"single-pool-target-set-lockers", false, false},
} {
t.Run(tc.name, func(t *testing.T) {
a := []dsync.NetLocker{newLocker(), newLocker()}
b := []dsync.NetLocker{newLocker(), newLocker()}
if tc.sharedLockers {
b = a
}
set0 := &erasureObjects{nsMutex: newNSLock(true), getLockers: func() ([]dsync.NetLocker, string) { return a, "listing-test" }}
target := &erasureObjects{nsMutex: newNSLock(true), getLockers: func() ([]dsync.NetLocker, string) { return b, "listing-test" }}
z := &erasureServerPools{serverPools: []*erasureSets{{sets: []*erasureObjects{set0, target}}}}
if tc.multiPool {
z.serverPools = []*erasureSets{{sets: []*erasureObjects{set0}}, {sets: []*erasureObjects{target}}}
}
writer := target.NewNSLock("bucket", "object")
if tc.multiPool {
writer = z.NewNSLock("bucket", "object")
}
wctx, err := writer.GetLock(t.Context(), globalOperationTimeout)
if err != nil {
t.Fatal(err)
}
locked := true
defer func() {
if locked {
writer.Unlock(wctx)
}
}()
reader := z.listingNSLock(target)("bucket", "object")
ctx, cancel := context.WithTimeout(t.Context(), 80*time.Millisecond)
defer cancel()
rctx, err := reader.GetRLock(ctx, globalOperationTimeout)
if err == nil {
reader.RUnlock(rctx)
t.Fatal("fallback read lock acquired while ordinary writer lock is held")
}
writer.Unlock(wctx)
locked = false
ctx, cancel = context.WithTimeout(t.Context(), 5*time.Second)
defer cancel()
rctx, err = reader.GetRLock(ctx, globalOperationTimeout)
if err != nil {
t.Fatalf("fallback read lock failed after writer unlock: %v", err)
}
reader.RUnlock(rctx)
})
}
}
+137
View File
@@ -0,0 +1,137 @@
// Copyright (c) 2026 Feng Ruohang
//
// This file is part of Silo Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"bytes"
"context"
"net/http"
"testing"
"time"
)
// The fallback on pool 1 must contend with the
// pools-layer lock used by ordinary multi-pool writers (anchored at pool 0).
func TestListFallbackMultiPoolReadLock(t *testing.T) {
z, bucket, _ := listingPoolsBackend(t)
ctx := t.Context()
set := z.serverPools[1].sets[0]
disks := set.getDisks()
const object = "object"
body := bytes.Repeat([]byte{'a'}, 8192)
mtime := time.Now().UTC().Add(-time.Hour)
put := func(tm time.Time) {
_, err := set.PutObject(ctx, bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm})
if err != nil {
t.Fatal(err)
}
}
put(mtime)
old := make([][]byte, len(disks))
for i := range old {
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
}
put(mtime.Add(time.Minute))
entries := make(metaCacheEntries, len(disks))
errs := make([]error, len(disks))
for i := range 3 {
entries[i] = metaCacheEntry{name: object, metadata: old[i]}
}
for i := 3; i < 8; i++ {
entries[i] = metaCacheEntry{name: object, metadata: mustReadNullQuorumMeta(t, disks[i], bucket, object)}
}
for i := 8; i < len(disks); i++ {
errs[i] = errDiskNotFound
}
resolver := metadataResolutionParams{dirQuorum: 8, objQuorum: 8, bucket: bucket, requestedVersions: 1}
lock := z.NewNSLock(bucket, object)
lkctx, err := lock.GetLock(ctx, globalOperationTimeout)
if err != nil {
t.Fatal(err)
}
locked := true
defer func() {
if locked {
lock.Unlock(lkctx)
}
}()
done := make(chan struct{})
var entry *metaCacheEntry
var readErr error
go func() {
defer close(done)
entry, readErr = set.resolveListEntry(ctx, bucket, entries, errs, &resolver, z.listingNSLock(set))
}()
waitForListingReadLock(t, z.serverPools[0].sets[0].nsMutex, bucket, object, done)
lock.Unlock(lkctx)
locked = false
<-done
if readErr != nil || entry == nil || entry.name != object {
t.Fatalf("fallback after writer unlock: entry=%v err=%v", entry, readErr)
}
}
// Observe the reader registered on the writer's actual namespace lock before
// releasing the writer. This tests mutual exclusion without a timed sleep.
func waitForListingReadLock(t *testing.T, mutex *nsLockMap, bucket, object string, done <-chan struct{}) {
t.Helper()
timer := time.NewTimer(5 * time.Second)
defer timer.Stop()
tick := time.NewTicker(time.Millisecond)
defer tick.Stop()
for {
mutex.lockMapMutex.Lock()
lock := mutex.lockMap[pathJoin(bucket, object)]
waiting := lock != nil && lock.ref > 1
mutex.lockMapMutex.Unlock()
if waiting {
return
}
select {
case <-done:
t.Fatal("listing completed while the writer held its namespace lock")
case <-timer.C:
t.Fatal("listing did not wait on the writer's namespace lock")
case <-tick.C:
}
}
}
func listingPoolsBackend(t *testing.T) (*erasureServerPools, string, http.Handler) {
t.Helper()
ctx, cancel := context.WithCancel(t.Context())
obj, dirs, err := prepareErasurePoolsWithContext(ctx)
if err != nil {
cancel()
t.Fatal(err)
}
z := obj.(*erasureServerPools)
previous := newObjectLayerFn()
setObjectLayer(z)
t.Cleanup(func() {
cancel()
z.Shutdown(context.Background())
removeRoots(dirs)
setObjectLayer(previous)
})
bucket, router, err := initAPIHandlerTest(ctx, z, nil, MakeBucketOptions{})
if err != nil {
t.Fatal(err)
}
return z, bucket, router
}
+244
View File
@@ -0,0 +1,244 @@
// Copyright (c) 2026 Feng Ruohang
//
// This file is part of Silo Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"bytes"
"context"
"encoding/xml"
"errors"
"io"
"net/http"
"net/http/httptest"
"net/url"
"slices"
"sync/atomic"
"testing"
"time"
"github.com/klauspost/compress/s2"
"github.com/tinylib/msgp/msgp"
)
// Replay real disk-walk metadata in its existing wire format, but pause before
// the final key. This keeps the healthy pool's listing alive after the other
// pool has already failed, without forging entries or listing errors.
type listingPausedTailDisk struct {
StorageAPI
bucket string
tail string
resume <-chan struct{}
blocked *atomic.Int64
active *atomic.Int64
}
func (d *listingPausedTailDisk) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) (result error) {
if opts.Bucket != d.bucket {
return d.StorageAPI.WalkDir(ctx, opts, out)
}
if d.active != nil {
d.active.Add(1)
defer d.active.Add(-1)
}
var source bytes.Buffer
if err := d.StorageAPI.WalkDir(ctx, opts, &source); err != nil {
return err
}
r := newMetacacheReader(bytes.NewReader(source.Bytes()))
defer r.Close()
compressed := s2.NewWriter(out)
defer func() {
if err := compressed.Close(); result == nil {
result = err
}
}()
w := msgp.NewWriter(compressed)
if err := w.WriteByte(metacacheStreamVersion); err != nil {
return err
}
for {
entry, err := r.next()
if err == io.EOF {
if err = w.WriteBool(false); err != nil {
return err
}
if err = w.Flush(); err != nil {
return err
}
return nil
}
if err != nil {
return err
}
if entry.name == d.tail {
if err = w.Flush(); err != nil {
return err
}
if err = compressed.Flush(); err != nil {
return err
}
d.blocked.Add(1)
select {
case <-d.resume:
case <-ctx.Done():
return ctx.Err()
}
}
if err = w.WriteBool(true); err != nil {
return err
}
if err = w.WriteString(entry.name); err != nil {
return err
}
if err = w.WriteBytes(entry.metadata); err != nil {
return err
}
}
}
func TestListingMergeDoesNotSkipFailedPool(t *testing.T) {
ctx, cancel := context.WithTimeout(t.Context(), 20*time.Second)
defer cancel()
z, bucket, router := listingPoolsBackend(t)
broken, healthy := z.serverPools[0].sets[0], z.serverPools[1].sets[0]
brokenDisks, switchBrokenDisks := installListingDiskSwitch(broken)
healthyDisks, switchHealthyDisks := installListingDiskSwitch(healthy)
body := bytes.Repeat([]byte{'a'}, 8192)
mtime := time.Now().UTC().Add(-time.Hour)
put := func(set *erasureObjects, name string, tm time.Time) {
_, err := set.PutObject(ctx, bucket, name, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{MTime: tm})
if err != nil {
t.Fatal(err)
}
}
put(broken, "a", mtime)
put(broken, "e", mtime)
for _, name := range []string{"f", "g", "h", "i", "j", "k", "zz"} {
put(healthy, name, mtime)
}
first := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1", "", "", ""))
var page ListObjectsV2Response
if first.Code != http.StatusOK || xml.Unmarshal(first.Body.Bytes(), &page) != nil || !page.IsTruncated || page.NextContinuationToken == "" || len(page.Contents) != 1 || page.Contents[0].Key != "a" {
t.Fatalf("first page HTTP=%d body=%s", first.Code, first.Body.String())
}
firstToken := page.NextContinuationToken
old := make([][]byte, 3)
for i := range old {
old[i] = mustReadNullQuorumMeta(t, brokenDisks[i], bucket, "e")
}
put(broken, "e", mtime.Add(time.Minute))
for i := range old {
if err := brokenDisks[i].WriteAll(ctx, bucket, "e/"+xlStorageFormatFile, old[i]); err != nil {
t.Fatal(err)
}
}
var reads, blocked, active atomic.Int64
brokenWalkers := slices.Clone(brokenDisks)
for i := range brokenWalkers {
brokenWalkers[i] = &listingReadFailDisk{StorageAPI: brokenDisks[i], object: "e", reads: &reads}
if i >= 8 {
brokenWalkers[i] = &failedMetadataWalkDisk{StorageAPI: brokenWalkers[i]}
}
}
switchBrokenDisks(brokenWalkers)
resume := make(chan struct{})
defer close(resume)
healthyWalkers := slices.Clone(healthyDisks)
for i, disk := range healthyWalkers {
healthyWalkers[i] = &listingPausedTailDisk{StorageAPI: disk, bucket: bucket, tail: "zz", resume: resume, blocked: &blocked, active: &active}
}
switchHealthyDisks(healthyWalkers)
target := getListObjectsV2URL("", bucket, "", "2", "", "", "") + "&continuation-token=" + url.QueryEscape(firstToken)
req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
if err != nil {
t.Fatal(err)
}
response := make(chan *httptest.ResponseRecorder, 1)
go func() {
rec := httptest.NewRecorder()
router.ServeHTTP(rec, req.WithContext(ctx))
response <- rec
}()
select {
case rec := <-response:
t.Logf("continuation LIST=%d failed fallback ReadXL=%d paused healthy walkers=%d body=%s", rec.Code, reads.Load(), blocked.Load(), rec.Body.String())
if reads.Load() == 0 || blocked.Load() == 0 {
t.Fatal("schedule did not exercise fallback failure and unfinished healthy pool")
}
var apiErr APIErrorResponse
if rec.Code != http.StatusServiceUnavailable || xml.Unmarshal(rec.Body.Bytes(), &apiErr) != nil || apiErr.Code != "SlowDownRead" {
t.Fatalf("failed pool advanced the page instead of reporting read quorum: %d %s", rec.Code, rec.Body.String())
}
waitForListingCondition(t, "canceled disk walkers did not exit", func() bool { return active.Load() == 0 })
case <-ctx.Done():
t.Fatalf("no early response: %v reads=%d blocked=%d", ctx.Err(), reads.Load(), blocked.Load())
}
}
func TestListingMergeCancellation(t *testing.T) {
z, bucket, _ := nullQuorumBackend(t)
for _, key := range []string{"a", "b", "zz"} {
nullQuorumPut(t, z, bucket, key, []byte(key), ObjectOptions{})
}
set := z.serverPools[0].sets[0]
var blocked, active atomic.Int64
resume := make(chan struct{})
defer close(resume)
walkers := slices.Clone(set.getDisks())
for i, disk := range walkers {
walkers[i] = &listingPausedTailDisk{StorageAPI: disk, bucket: bucket, tail: "zz", resume: resume, blocked: &blocked, active: &active}
}
set.getDisks = func() []StorageAPI { return walkers }
ctx, cancel := context.WithCancel(t.Context())
defer cancel()
entries := make(chan metaCacheEntry, 10)
done := make(chan error, 1)
go func() {
done <- z.listMerged(ctx, listPathOptions{Bucket: bucket, Recursive: true, Limit: 10}, entries)
}()
waitForListingCondition(t, "disk walks did not reach the paused tail", func() bool { return blocked.Load() > 0 })
cancel()
select {
case err := <-done:
if !errors.Is(err, context.Canceled) {
t.Fatalf("canceled merge returned %v", err)
}
case <-time.After(5 * time.Second):
t.Fatal("canceled merge did not finish")
}
// listMerged must close its output and release every paused disk walk.
for range entries {
}
waitForListingCondition(t, "canceled disk walkers did not exit", func() bool { return active.Load() == 0 })
}
func waitForListingCondition(t *testing.T, failure string, ready func() bool) {
t.Helper()
timer := time.NewTimer(5 * time.Second)
defer timer.Stop()
tick := time.NewTicker(time.Millisecond)
defer tick.Stop()
for !ready() {
select {
case <-timer.C:
t.Fatal(failure)
case <-tick.C:
}
}
}
+406
View File
@@ -52,6 +52,37 @@ type nullQuorumWalkDisk struct {
index int
}
type failedMetadataWalkDisk struct {
StorageAPI
}
func (d *failedMetadataWalkDisk) WalkDir(context.Context, WalkDirOptions, io.Writer) error {
return errDiskNotFound
}
// offlineNullQuorumDisk models a drive of a node that is away.
type offlineNullQuorumDisk struct {
StorageAPI
}
func (d *offlineNullQuorumDisk) IsOnline() bool { return false }
func (d *offlineNullQuorumDisk) DiskInfo(context.Context, DiskInfoOptions) (DiskInfo, error) {
return DiskInfo{}, errDiskNotFound
}
func (d *offlineNullQuorumDisk) WalkDir(context.Context, WalkDirOptions, io.Writer) error {
return errDiskNotFound
}
func (d *offlineNullQuorumDisk) ReadXL(context.Context, string, string, bool) (RawFileInfo, error) {
return RawFileInfo{}, errDiskNotFound
}
func (d *offlineNullQuorumDisk) ReadVersion(context.Context, string, string, string, string, ReadOptions) (FileInfo, error) {
return FileInfo{}, errDiskNotFound
}
func (d *nullQuorumWalkDisk) WalkDir(ctx context.Context, opts WalkDirOptions, out io.Writer) error {
if opts.Bucket != d.schedule.bucket {
return d.StorageAPI.WalkDir(ctx, opts, out)
@@ -119,6 +150,30 @@ func nullQuorumRequest(t *testing.T, router http.Handler, method, target string)
return rec
}
func nullQuorumPut(t *testing.T, z *erasureServerPools, bucket, object string, body []byte, opts ObjectOptions) {
t.Helper()
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), opts); err != nil {
t.Fatal(err)
}
}
func nullQuorumListV2Keys(t *testing.T, router http.Handler, bucket string) (int, []string) {
t.Helper()
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
if rec.Code != http.StatusOK {
return rec.Code, nil
}
var list ListObjectsV2Response
if err := xml.Unmarshal(rec.Body.Bytes(), &list); err != nil {
t.Fatal(err)
}
keys := make([]string, 0, len(list.Contents))
for _, object := range list.Contents {
keys = append(keys, object.Key)
}
return rec.Code, keys
}
func TestListObjectsSingleNullQuorumHTTP(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
oldBody := bytes.Repeat([]byte{'a'}, 8192)
@@ -227,6 +282,357 @@ func TestListObjectsSingleNullQuorumHTTP(t *testing.T) {
}
}
func TestListObjectsFallsBackToReadableMetadata(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
set := z.serverPools[0].sets[0]
disks := set.getDisks()
const object = "object"
oldBody := bytes.Repeat([]byte{'a'}, 8192)
newBody := bytes.Repeat([]byte{'b'}, 8192)
oldTime := time.Now().UTC().Add(-time.Hour)
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(oldBody), int64(len(oldBody)), "", ""), ObjectOptions{MTime: oldTime}); err != nil {
t.Fatal(err)
}
oldMeta := make([][]byte, len(disks))
for i, disk := range disks {
oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
}
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(newBody), int64(len(newBody)), "", ""), ObjectOptions{MTime: oldTime.Add(time.Minute)}); err != nil {
t.Fatal(err)
}
// The eight readable walk streams are split 3/5 across two generations;
// the other eight streams fail. Neither generation is proven readable by
// the listing inputs, so returning HTTP 200 would silently omit the key.
for i := range 3 {
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil {
t.Fatal(err)
}
}
wrapped := make([]StorageAPI, len(disks))
for i, disk := range disks {
wrapped[i] = disk
if i >= 8 {
wrapped[i] = &failedMetadataWalkDisk{StorageAPI: disk}
}
}
set.getDisks = func() []StorageAPI { return wrapped }
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
var list ListObjectsV2Response
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
t.Fatalf("LIST failed to resolve readable metadata: %d %s", rec.Code, rec.Body.String())
}
if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object {
t.Fatalf("LIST omitted the readable key: %+v", list)
}
// The object itself still has a read quorum outside the failed walk
// streams, demonstrating why the LIST must report uncertainty, not absence.
get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object))
if get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), newBody) {
t.Fatalf("GET lost the readable generation: %d %q", get.Code, get.Body.String())
}
}
// The fallback must read under the object read lock. A lock-free read of an
// object being overwritten stably reports a live object as absent, which is
// the omission this repair exists to remove. Hold the write lock and assert
// the listing waits for it rather than resolving straight through.
func TestListFallbackTakesObjectReadLock(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
set := z.serverPools[0].sets[0]
disks := set.getDisks()
const object = "object"
oldBody, newBody := bytes.Repeat([]byte{'a'}, 8192), bytes.Repeat([]byte{'b'}, 8192)
oldTime := time.Now().UTC().Add(-time.Hour)
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(oldBody), int64(len(oldBody)), "", ""), ObjectOptions{MTime: oldTime}); err != nil {
t.Fatal(err)
}
oldMeta := make([][]byte, len(disks))
for i, disk := range disks {
oldMeta[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
}
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(newBody), int64(len(newBody)), "", ""), ObjectOptions{MTime: oldTime.Add(time.Minute)}); err != nil {
t.Fatal(err)
}
// Split the readable walk streams so the entry cannot be resolved from
// the snapshots and the fallback has to run.
for i := range 3 {
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, oldMeta[i]); err != nil {
t.Fatal(err)
}
}
wrapped := make([]StorageAPI, len(disks))
for i, disk := range disks {
wrapped[i] = disk
if i >= 8 {
wrapped[i] = &failedMetadataWalkDisk{StorageAPI: disk}
}
}
set.getDisks = func() []StorageAPI { return wrapped }
lock := set.NewNSLock(bucket, object)
lkctx, err := lock.GetLock(t.Context(), globalOperationTimeout)
if err != nil {
t.Fatal(err)
}
locked := true
defer func() {
if locked {
lock.Unlock(lkctx)
}
}()
target := getListObjectsV2URL("", bucket, "", "1000", "", "", "")
req, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
if err != nil {
t.Fatal(err)
}
rec := httptest.NewRecorder()
done := make(chan struct{})
go func() {
defer close(done)
router.ServeHTTP(rec, req)
}()
waitForListingReadLock(t, set.nsMutex, bucket, object, done)
lock.Unlock(lkctx)
locked = false
<-done
var list ListObjectsV2Response
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
t.Fatalf("LIST failed after the lock was released: %d %s", rec.Code, rec.Body.String())
}
if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object {
t.Fatalf("LIST omitted the readable key: %+v", list)
}
}
// A truncated xl.meta is what a concurrent overwrite looks like on one drive,
// and the walker skips that entry. Skipping is safe and deliberate: the key is
// only omitted when no generation reaches quorum, which listPath now resolves
// against the read path. Pin that, so the walker is not "hardened" into
// failing a whole listing over a minority of unreadable drives.
func TestListObjectsSurvivesTruncatedMetadataMinority(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
set := z.serverPools[0].sets[0]
disks := set.getDisks()
const object = "object"
body := bytes.Repeat([]byte{'a'}, 8192)
if _, err := z.PutObject(t.Context(), bucket, object, mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil {
t.Fatal(err)
}
// Truncate xl.meta on a minority of drives, as an overwrite in flight does.
for i := range 4 {
meta := mustReadNullQuorumMeta(t, disks[i], bucket, object)
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, meta[:len(meta)/2]); err != nil {
t.Fatal(err)
}
}
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectsV2URL("", bucket, "", "1000", "", "", ""))
var list ListObjectsV2Response
if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &list) != nil {
t.Fatalf("a truncated minority failed the listing: %d %s", rec.Code, rec.Body.String())
}
if list.KeyCount != 1 || len(list.Contents) != 1 || list.Contents[0].Key != object {
t.Fatalf("a truncated minority omitted the key: %+v", list)
}
}
func TestListStaleMinorityIgnored(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
disks := z.serverPools[0].sets[0].getDisks()
body := bytes.Repeat([]byte{'a'}, 8192)
nullQuorumPut(t, z, bucket, "gone", body, ObjectOptions{})
nullQuorumPut(t, z, bucket, "keep", body, ObjectOptions{})
for _, disk := range disks[4:] {
if err := disk.Delete(t.Context(), bucket, "gone", DeleteOptions{Recursive: true, Immediate: true}); err != nil {
t.Fatal(err)
}
}
if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{"keep"}) {
t.Errorf("ListObjectsV2: got %d %v; want 200 [keep]", code, keys)
}
results := make(chan itemOrErr[ObjectInfo], 16)
if err := z.Walk(t.Context(), bucket, "", results, WalkOptions{}); err != nil {
t.Fatal(err)
}
var names []string
for result := range results {
if result.Err != nil {
t.Fatal(result.Err)
}
names = append(names, result.Item.Name)
}
if !slices.Equal(names, []string{"keep"}) {
t.Errorf("Walk: got %v; want [keep]", names)
}
rec := nullQuorumRequest(t, router, http.MethodGet, getListObjectVersionsURL("", bucket, "", "1000", ""))
var list struct {
Versions []struct {
Key string `xml:"Key"`
} `xml:"Version"`
}
if rec.Code != http.StatusOK {
t.Errorf("ListObjectVersions: got %d; want 200", rec.Code)
} else if err := xml.Unmarshal(rec.Body.Bytes(), &list); err != nil {
t.Fatal(err)
} else if len(list.Versions) != 1 || list.Versions[0].Key != "keep" {
t.Errorf("ListObjectVersions: got %+v; want [keep]", list.Versions)
}
if _, err := z.DeleteObject(t.Context(), bucket, "keep", ObjectOptions{}); err != nil {
t.Fatal(err)
}
if err := z.DeleteBucket(t.Context(), bucket, DeleteBucketOptions{}); err != nil {
t.Errorf("DeleteBucket with only a stale minority: %v", err)
}
}
func TestIAMListingIgnoresStaleMinority(t *testing.T) {
z, _, _ := nullQuorumBackend(t)
body := []byte(`{"version":1}`)
alive := iamConfigUsersPrefix + "alive/identity.json"
ghost := iamConfigUsersPrefix + "ghost/identity.json"
nullQuorumPut(t, z, minioMetaBucket, alive, body, ObjectOptions{})
nullQuorumPut(t, z, minioMetaBucket, ghost, body, ObjectOptions{})
for _, disk := range z.serverPools[0].getHashedSet(ghost).getDisks()[4:] {
if err := disk.Delete(t.Context(), minioMetaBucket, ghost, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
t.Fatal(err)
}
}
items, err := newIAMObjectStore(z, MinIOUsersSysType).listAllIAMConfigItems(t.Context())
if err != nil {
t.Fatalf("IAM listing failed on a stale minority (retriable=%v): %v", configRetriableErrors(err), err)
}
if got := items["users/"]; !slices.Equal(got, []string{"alive/identity.json"}) {
t.Errorf("IAM users: got %v; want [alive/identity.json]", got)
}
}
func TestListSplitDirectoryObject(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
set := z.serverPools[0].sets[0]
disks := set.getDisks()
const object = "dir/"
encoded := encodeDirObject(object)
mtime := time.Now().UTC().Add(-time.Hour)
nullQuorumPut(t, z, bucket, object, nil, ObjectOptions{MTime: mtime})
old := make([][]byte, 3)
for i := range old {
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, encoded)
}
nullQuorumPut(t, z, bucket, object, nil, ObjectOptions{MTime: mtime.Add(time.Minute)})
for i := range old {
if err := disks[i].WriteAll(t.Context(), bucket, encoded+"/"+xlStorageFormatFile, old[i]); err != nil {
t.Fatal(err)
}
}
walkers := slices.Clone(disks)
for i := 8; i < len(walkers); i++ {
walkers[i] = &failedMetadataWalkDisk{StorageAPI: disks[i]}
}
set.getDisks = func() []StorageAPI { return walkers }
if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) {
t.Errorf("ListObjectsV2: got %d %v; want 200 [%s]", code, keys, object)
}
}
func TestListFallbackDoesNotTrustNotFound(t *testing.T) {
z, bucket, _ := nullQuorumBackend(t)
set := z.serverPools[0].sets[0]
disks := set.getDisks()
const object = "object"
body := bytes.Repeat([]byte{'a'}, 8192)
mtime := time.Now().UTC().Add(-time.Hour)
nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime})
old := make([][]byte, len(disks))
for i := range disks {
old[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
}
nullQuorumPut(t, z, bucket, object, body, ObjectOptions{MTime: mtime.Add(time.Minute)})
newer := make([][]byte, len(disks))
for i := range disks {
newer[i] = mustReadNullQuorumMeta(t, disks[i], bucket, object)
if err := disks[i].Delete(t.Context(), bucket, object, DeleteOptions{Recursive: true, Immediate: true}); err != nil {
t.Fatal(err)
}
}
entries := make(metaCacheEntries, len(disks))
errs := make([]error, len(disks))
for i := range 3 {
entries[i] = metaCacheEntry{name: object, metadata: old[i]}
}
for i := 3; i < 8; i++ {
entries[i] = metaCacheEntry{name: object, metadata: newer[i]}
}
for i := 8; i < len(errs); i++ {
errs[i] = errDiskNotFound
}
resolver := metadataResolutionParams{dirQuorum: 8, objQuorum: 8, bucket: bucket, requestedVersions: 1}
entry, err := set.resolveListEntry(t.Context(), bucket, entries, errs, &resolver, z.listingNSLock(set))
if entry != nil || !isErrReadQuorum(err) {
t.Fatalf("ambiguous not-found: got entry=%v err=%v; want read quorum error", entry, err)
}
}
// G0 was written 8+8 while a node was away. G1 was later written 12+4 but
// missed drive 0, which still holds G0. Then a node holding four G1 drives
// goes away: eleven G1 and one G0 remain online. G1 exists but lacks one of
// its twelve data shards until the node returns, so HEAD must not report 404
// and LIST must still show the key.
func TestListCrossParityMinorityStaysVisible(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
set := z.serverPools[0].sets[0]
disks, switchDisks := installListingDiskSwitch(set)
const object = "object"
mtime := time.Now().UTC().Add(-time.Hour)
nullQuorumPut(t, z, bucket, object, bytes.Repeat([]byte{'a'}, 256<<10), ObjectOptions{MTime: mtime, MaxParity: true})
g0 := mustReadNullQuorumMeta(t, disks[0], bucket, object)
var x xlMetaV2
if err := x.LoadOrConvert(g0); err != nil {
t.Fatal(err)
}
if h := x.versions[0].header; h.EcM != 8 || h.EcN != 8 {
t.Fatalf("G0 is %d+%d, want 8+8", h.EcM, h.EcN)
}
g1Body := bytes.Repeat([]byte{'b'}, 256<<10)
nullQuorumPut(t, z, bucket, object, g1Body, ObjectOptions{MTime: mtime.Add(time.Minute)})
if err := disks[0].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, g0); err != nil {
t.Fatal(err)
}
online := slices.Clone(disks)
for i := 12; i < len(online); i++ {
online[i] = &offlineNullQuorumDisk{StorageAPI: disks[i]}
}
switchDisks(online)
if code := nullQuorumRequest(t, router, http.MethodHead, getHeadObjectURL("", bucket, object)).Code; code == http.StatusNotFound {
t.Errorf("HEAD with a node away: 404 for an existing object")
}
if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) {
t.Errorf("ListObjectsV2 with a node away: got %d %v; want 200 [%s]", code, keys, object)
}
switchDisks(disks)
if get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object)); get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), g1Body) {
t.Fatalf("GET after the node returns: %d", get.Code)
}
}
func TestSingleNullQuorumReadAndConsumers(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
set := z.serverPools[0].sets[0]
+54 -14
View File
@@ -191,19 +191,19 @@ func (z *erasureServerPools) listPath(ctx context.Context, o *listPathOptions) (
return entries, err
}
entries.truncate(0)
go func() {
rpc := globalNotificationSys.restClientFromHash(pathJoin(o.Bucket, o.Prefix))
go func(o listPathOptions) {
if rpc != nil {
ctx, cancel := context.WithTimeout(GlobalContext, 5*time.Second)
defer cancel()
c, err := rpc.GetMetacacheListing(ctx, *o)
c, err := rpc.GetMetacacheListing(ctx, o)
if err == nil {
c.error = "no longer used"
c.status = scanStateError
rpc.UpdateMetacacheListing(ctx, *c)
}
}
}()
}(*o)
o.ID = ""
}
@@ -264,6 +264,16 @@ func (z *erasureServerPools) listPathShortcutEOF(ctx context.Context, bucket str
return io.EOF
}
// listingNSLock chooses the namespace used by regular object writes.
// Multiple pools serialize at the pools layer; single-pool PUT and GET use
// the destination set. This keeps the existing single-pool lock routing.
func (z *erasureServerPools) listingNSLock(er *erasureObjects) func(string, ...string) RWLocker {
if z.SinglePool() {
return er.NewNSLock
}
return z.NewNSLock
}
// listMerged will list across all sets and return a merged results stream.
// The result channel is closed when no more results are expected.
func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry) error {
@@ -283,13 +293,16 @@ func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions,
inputs = append(inputs, innerResults)
go func(i int, set *erasureObjects) {
defer wg.Done()
err := set.listPath(listCtx, o, innerResults)
err := set.listPath(listCtx, o, innerResults, z.listingNSLock(set))
mu.Lock()
defer mu.Unlock()
if err == nil {
allAtEOF = false
}
errs[i] = err
mu.Unlock()
// The merger must observe the terminal error before treating
// this stream as exhausted and advancing to another pool.
xioutil.SafeClose(innerResults)
}(len(errs), set)
errs = append(errs, nil)
}
@@ -298,7 +311,16 @@ func (z *erasureServerPools) listMerged(ctx context.Context, o listPathOptions,
// Gather results to a single channel.
// Quorum is one since we are merging across sets.
err := mergeEntryChannels(ctx, inputs, results, 1)
err := mergeEntryChannelsWithErrors(ctx, inputs, results, 1, func(i int) error {
mu.Lock()
err := errs[i]
mu.Unlock()
// Keep the existing aggregate handling of absent paths and buckets.
if errors.Is(err, io.EOF) || isAllNotFound([]error{err}) {
return nil
}
return err
})
cancelList()
wg.Wait()
@@ -415,13 +437,18 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
cancel()
}()
// Do listing...
// Publish the typed terminal result before either consumer observes EOF.
// mergeEntryChannels closes inCh before listMerged has inspected set errors.
listingDone := make(chan struct{})
var listingErr error
go func(o listPathOptions) {
err := z.listMerged(listCtx, o, inCh)
if err != nil {
meta.setErr(err.Error())
listingErr = z.listMerged(listCtx, o, inCh)
if listingErr != nil && !errors.Is(listingErr, io.EOF) {
meta.setErr(listingErr.Error())
cancel()
}
o.debugln("listAndSave: listing", o.ID, "finished with ", err)
close(listingDone)
o.debugln("listAndSave: listing", o.ID, "finished with ", listingErr)
}(*o)
// Keep track of when we return since we no longer have to send entries to output.
@@ -451,6 +478,9 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
entry.reusable = returned
saveCh <- entry
}
// Do not publish an end-of-stream cache block or a final response
// before the scan has recorded its terminal error.
<-listingDone
if !returned {
xioutil.SafeClose(outCh)
}
@@ -458,9 +488,19 @@ func (z *erasureServerPools) listAndSave(ctx context.Context, o *listPathOptions
}()
entries, err = filteredResults()
if err == nil {
// Check if listing recorded an error.
err = meta.getErr()
if err == nil || errors.Is(err, io.EOF) {
// Full pages may return before the scan ends. A final page has
// crossed the completion barrier above and must preserve scan errors.
select {
case <-listingDone:
if listingErr != nil && !errors.Is(listingErr, io.EOF) {
return entries, listingErr
}
default:
}
if cacheErr := meta.getErr(); cacheErr != nil {
return entries, cacheErr
}
}
return entries, err
}
+92 -8
View File
@@ -39,7 +39,6 @@ import (
"github.com/minio/minio/internal/bucket/versioning"
"github.com/minio/minio/internal/color"
"github.com/minio/minio/internal/hash"
xioutil "github.com/minio/minio/internal/ioutil"
"github.com/pgsty/silo-pkg/v3/console"
)
@@ -700,9 +699,88 @@ func getQuorumDisks(disks []StorageAPI, infos []DiskInfo, readQuorum int) (newDi
return newDisks
}
// resolveListEntry resolves walk snapshots and, for latest-only listings,
// falls back to the regular object read path when those snapshots disagree.
//
// Only unresolved entries pay for the additional namespace read lock and
// metadata read. The caller supplies the lock domain used by object writes.
func (er *erasureObjects) resolveListEntry(ctx context.Context, bucket string, entries metaCacheEntries, errs []error, resolver *metadataResolutionParams, newNSLock func(string, ...string) RWLocker) (*metaCacheEntry, error) {
entry, ok := entries.resolve(resolver)
if ok {
return entry, nil
}
if resolver.requestedVersions != 1 {
return nil, nil
}
valid := 0
for i := range entries {
if !entries[i].isObject() {
continue
}
if _, err := entries[i].xlmeta(); err != nil {
continue
}
valid++
if entry == nil {
entry = &entries[i]
}
}
for _, err := range errs {
if err != nil {
valid++
}
}
if entry == nil || valid < resolver.objQuorum {
return nil, nil
}
readQuorumErr := func() error {
return InsufficientReadQuorum{
Bucket: bucket,
Object: entry.name,
Err: errErasureReadQuorum,
Type: RQInconsistentMeta,
}
}
// Read under the object read lock, exactly as the GET path does. A
// lock-free read of an object being overwritten stably reports a live
// object as absent, so an unlocked fallback only trades one silent
// omission for another.
object := encodeDirObject(entry.name)
lock := newNSLock(bucket, object)
lockCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
defer cancel()
lkctx, err := lock.GetRLock(lockCtx, globalOperationTimeout)
if err != nil {
return nil, err
}
defer lock.RUnlock(lkctx)
fi, _, _, err := er.getObjectFileInfo(lkctx.Context(), bucket, object, ObjectOptions{}, false)
if err != nil {
// getObjectFileInfo maps inconsistent metadata to not-found, so the
// walk evidence means not-found is still undecidable here.
if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
return nil, readQuorumErr()
}
return nil, err
}
// Preserve delete markers until the cross-pool merge has discarded any
// older live copy. The normal listing filter removes the merged marker.
xl := &xlMetaV2{}
if err = xl.AddVersion(fi); err != nil {
return nil, readQuorumErr()
}
resolved := &metaCacheEntry{name: entry.name, cached: xl, reusable: true}
resolved.metadata, err = xl.AppendTo(metaDataPoolGet())
if err != nil {
return nil, err
}
return resolved, nil
}
// Will return io.EOF if continuing would not yield more results.
func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry) (err error) {
defer xioutil.SafeClose(results)
func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, results chan<- metaCacheEntry, newNSLock func(string, ...string) RWLocker) (err error) {
// The caller records the terminal error before closing results.
o.debugf(color.Green("listPath:")+" with options: %#v", o)
// get prioritized non-healing disks for listing
@@ -776,15 +854,19 @@ func (er *erasureObjects) listPath(ctx context.Context, o listPathOptions, resul
case results <- entry:
}
},
partial: func(entries metaCacheEntries, errs []error) {
partial: func(entries metaCacheEntries, errs []error) error {
// Results Disagree :-(
entry, ok := entries.resolve(&resolver)
if ok {
entry, err := er.resolveListEntry(ctx, o.Bucket, entries, errs, &resolver, newNSLock)
if err != nil {
return err
}
if entry != nil {
select {
case <-ctxDone:
case results <- *entry:
}
}
return nil
},
})
}
@@ -980,7 +1062,7 @@ type listPathRawOptions struct {
// partial will be called when there is disagreement between disks.
// if disk did not return any result, but also haven't errored
// the entry will be empty and errs will
partial func(entries metaCacheEntries, errs []error)
partial func(entries metaCacheEntries, errs []error) error
// finished will be called when all streams have finished and
// more than one disk returned an error.
@@ -1191,7 +1273,9 @@ func listPathRaw(ctx context.Context, opts listPathRawOptions) (err error) {
continue
}
if opts.partial != nil {
opts.partial(topEntries, errs)
if err := opts.partial(topEntries, errs); err != nil {
return err
}
}
// Skip the inputs we used.
for i, r := range readers {
+2 -20
View File
@@ -44,7 +44,6 @@ import (
"github.com/minio/minio-go/v7/pkg/tags"
xhttp "github.com/minio/minio/internal/http"
"github.com/pgsty/silo-pkg/v3/ldap"
"github.com/pgsty/silo-pkg/v3/policy"
)
func runAllIAMSTSTests(suite *TestSuiteIAM, c *check) {
@@ -2385,10 +2384,8 @@ func (s *TestSuiteIAM) TestIAMImport(c *check, exportedContent []byte, caseNum i
if err != nil {
c.Fatalf("import %d: Unable to list policies: %v", caseNum, err)
}
defaultCannedPolicies := set.CreateStringSet()
for _, p := range policy.DefaultPolicies {
defaultCannedPolicies.Add(p.Name)
}
defaultCannedPolicies := set.CreateStringSet("consoleAdmin", "readwrite", "readonly",
"diagnostics", "writeonly")
for policy, policyBytes := range policyContentMap {
if defaultCannedPolicies.Contains(policy) {
continue
@@ -2994,21 +2991,6 @@ func (s *TestSuiteIAM) TestLDAPSTSServiceAccounts(c *check) {
// 6. Check that service account cannot be created for some other user.
c.mustNotCreateSvcAccount(ctx, globalActiveCred.AccessKey, userAdmClient)
// An empty policy must not contaminate an expiring LDAP child credential.
emptyPolicyCred, err := userAdmClient.AddServiceAccountLDAP(ctx, madmin.AddServiceAccountReq{Policy: []byte("null")})
if err != nil {
c.Fatalf("Unable to create empty-policy LDAP service account: %v", err)
}
emptyPolicyAdmin := s.getAdminClient(c, emptyPolicyCred.AccessKey, emptyPolicyCred.SecretKey, "")
expires := UTCNow().Add(time.Hour)
for _, childPolicy := range [][]byte{nil, []byte("null")} {
child, err := emptyPolicyAdmin.AddServiceAccountLDAP(ctx, madmin.AddServiceAccountReq{Expiration: &expires, Policy: childPolicy})
if err != nil {
c.Fatalf("Unable to create expiring LDAP child: %v", err)
}
c.assertSvcAccS3Access(ctx, s, child, bucket)
}
// Detach the policy from the user
if _, err = s.adm.DetachPolicyLDAP(ctx, userReq); err != nil {
c.Fatalf("Unable to detach user policy: %v", err)
@@ -0,0 +1,222 @@
// Copyright (c) 2026 Feng Ruohang
//
// This file is part of Silo Object Storage stack
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.
package cmd
import (
"bytes"
"fmt"
"net/http"
"slices"
"testing"
"time"
)
func crossParityNullVersion(t *testing.T, generation int64, data, parity int) xlMetaV2ShallowVersion {
t.Helper()
fi := newFileInfo("bucket/object", data, parity)
fi.Erasure.Index = 1
fi.DataDir = "11111111-1111-1111-1111-111111111111"
fi.ModTime = time.Unix(generation, 0).UTC()
fi.Size = 1024
fi.Parts = []ObjectPartInfo{{Number: 1, Size: fi.Size, ActualSize: fi.Size}}
fi.Metadata = map[string]string{"etag": fmt.Sprintf("%032d", generation)}
var xl xlMetaV2
if err := xl.AddVersion(fi); err != nil {
t.Fatal(err)
}
return xl.versions[0]
}
// A restarted node can retain four 12+4 shards while a successful degraded
// write leaves twelve newer 8+8 shards. Neither disk order nor the minority's
// EC must hide the independently sufficient newer group.
func TestMergeXLV2LatestNullCrossParity(t *testing.T) {
versions := []xlMetaV2ShallowVersion{
crossParityNullVersion(t, 100, 12, 4),
crossParityNullVersion(t, 200, 8, 8),
crossParityNullVersion(t, 300, 10, 6),
}
for _, tt := range []struct {
name string
counts [3]int
empty int
quorum int
want int
}{
{"new-twelve-old-four", [3]int{4, 12}, 0, 8, 1},
{"old-twelve-new-four", [3]int{12, 4}, 0, 8, 0},
{"new-nine-old-seven", [3]int{7, 9}, 0, 8, 1},
{"eight-meets-eight", [3]int{4, 8}, 4, 8, 1},
{"eight-does-not-meet-nine", [3]int{8, 8}, 0, 9, -1},
{"three-subquorums", [3]int{7, 5, 4}, 0, 8, -1},
} {
t.Run(tt.name, func(t *testing.T) {
var input [][]xlMetaV2ShallowVersion
for group, count := range tt.counts {
for range count {
input = append(input, []xlMetaV2ShallowVersion{versions[group]})
}
}
input = append(input, make([][]xlMetaV2ShallowVersion, tt.empty)...)
for order, streams := range quorumVersionOrders(input) {
got := mergeXLV2Versions(tt.quorum, false, 1, streams...)
if tt.want < 0 {
if len(got) != 0 {
t.Fatalf("order %d: subquorum groups were combined: %v", order, got)
}
continue
}
if len(got) != 1 || got[0].header != versions[tt.want].header {
t.Fatalf("order %d: got %v, want group %d", order, got, tt.want)
}
xl := xlMetaV2{versions: got}
fi, err := xl.ToFileInfo("bucket", "object", "", false, true)
if err != nil || !fi.IsValid() || fi.Erasure.DataBlocks != int(versions[tt.want].header.EcM) {
t.Fatalf("order %d: selected metadata is invalid: %+v, %v", order, fi, err)
}
}
})
}
}
func TestMergeXLV2LatestNullCrossParityScope(t *testing.T) {
old := crossParityNullVersion(t, 100, 12, 4)
newer := crossParityNullVersion(t, 200, 8, 8)
for _, kind := range []string{"strict", "all-versions", "uuid", "delete", "free", "history", "legacy-modern-ec", "partial-ec"} {
t.Run(kind, func(t *testing.T) {
a, b := old, newer
strict, requested := false, 1
switch kind {
case "strict":
strict = true
case "all-versions":
requested = 0
case "uuid":
a.header.VersionID, b.header.VersionID = [16]byte{1}, [16]byte{1}
case "delete":
b.header.Type = DeleteType
case "free":
b.header.Flags |= xlFlagFreeVersion
case "legacy-modern-ec":
b.header.EcN, b.header.EcM = 0, 0
case "partial-ec":
b.header.EcN = 0
}
// These previously unresolved inputs remain outside the repair.
// Strict and legacy-header cases use an older majority followed by
// a newer minority; the other cases use the opposite grouping.
input := make([][]xlMetaV2ShallowVersion, 16)
for i := range input {
input[i] = []xlMetaV2ShallowVersion{b}
if i >= 12 {
input[i] = []xlMetaV2ShallowVersion{a}
}
}
if kind == "strict" || kind == "legacy-modern-ec" || kind == "partial-ec" {
for i := range input {
input[i] = []xlMetaV2ShallowVersion{a}
if i >= 12 {
input[i] = []xlMetaV2ShallowVersion{b}
}
}
}
if kind == "history" {
history := old
history.header.VersionID = [16]byte{1}
input[0] = append(input[0], history)
}
if got := mergeXLV2Versions(8, strict, requested, input...); len(got) != 0 {
t.Fatalf("excluded input changed: %v", got)
}
})
}
}
func TestCountXLV2MixedECGroups(t *testing.T) {
a := crossParityNullVersion(t, 100, 12, 4)
b := crossParityNullVersion(t, 100, 8, 8)
c := crossParityNullVersion(t, 200, 10, 6)
// Same-time groups remain separate, and equal counts have a deterministic
// final EC tiebreak even though sortsBefore does not compare EC itself.
input := [][]xlMetaV2ShallowVersion{{a}, {a}, {a}, {b}, {b}, {b}, {c}}
for order, streams := range quorumVersionOrders(input) {
tops := make([]xlMetaV2ShallowVersion, len(streams))
for i := range streams {
tops[i] = streams[i][0]
}
for range 20 {
got, count := countXLV2Versions(tops, c.header, c, false, true)
if count != 3 || got.header != a.header {
t.Fatalf("order %d: got %v, count %d; want distinct group of three", order, got.header, count)
}
}
}
differentSig := a
differentSig.header.Signature[0]++
if _, count := countXLV2Versions([]xlMetaV2ShallowVersion{a, a, differentSig}, a.header, a, false, true); count != 3 {
t.Fatalf("non-strict signature groups did not combine: count %d", count)
}
// Same EC with different time or flags must not accumulate votes either.
flags := a
flags.header.Flags ^= xlFlagInlineData
for _, other := range []xlMetaV2ShallowVersion{crossParityNullVersion(t, 200, 12, 4), flags} {
got, count := countXLV2Versions([]xlMetaV2ShallowVersion{a, a, other, other}, a.header, a, false, true)
if count != 2 {
t.Fatalf("incompatible header groups combined: %v, count %d", got.header, count)
}
}
}
// Exercise the regular full-metadata read and HTTP LIST with actual on-disk
// object metadata, not merely the metadata selection helper.
func TestListCrossParityReadableMajority(t *testing.T) {
for _, size := range []int{1024, 256 << 10} {
t.Run(fmt.Sprintf("bytes-%d", size), func(t *testing.T) {
z, bucket, router := nullQuorumBackend(t)
set := z.serverPools[0].sets[0]
disks := set.getDisks()
const object = "object"
mtime := time.Now().UTC().Add(-time.Hour)
oldBody := bytes.Repeat([]byte{'a'}, size)
newBody := bytes.Repeat([]byte{'b'}, size)
nullQuorumPut(t, z, bucket, object, oldBody, ObjectOptions{MTime: mtime})
old := make([][]byte, len(disks))
for i, disk := range disks {
old[i] = mustReadNullQuorumMeta(t, disk, bucket, object)
}
nullQuorumPut(t, z, bucket, object, newBody, ObjectOptions{MTime: mtime.Add(time.Minute), MaxParity: true})
for i := 12; i < len(disks); i++ {
if err := disks[i].WriteAll(t.Context(), bucket, object+"/"+xlStorageFormatFile, old[i]); err != nil {
t.Fatal(err)
}
}
for _, readData := range []bool{false, true} {
fi, _, _, err := set.getObjectFileInfo(t.Context(), bucket, object, ObjectOptions{}, readData)
if err != nil || !fi.ModTime.Equal(mtime.Add(time.Minute)) || fi.Erasure.DataBlocks != 8 {
t.Fatalf("readData=%v: full metadata read lost readable majority: %+v, %v", readData, fi, err)
}
}
if get := nullQuorumRequest(t, router, http.MethodGet, getGetObjectURL("", bucket, object)); get.Code != http.StatusOK || !bytes.Equal(get.Body.Bytes(), newBody) {
t.Fatalf("GET lost readable majority: status %d", get.Code)
}
if code, keys := nullQuorumListV2Keys(t, router, bucket); code != http.StatusOK || !slices.Equal(keys, []string{object}) {
t.Fatalf("LIST lost readable majority: %d %v", code, keys)
}
})
}
}
+34 -13
View File
@@ -1944,7 +1944,7 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions
// Keep the original stream shapes: pruning must not make a versioned object
// eligible for the single null-version recount below.
originalVersions := versions
var checkedSingleNull, singleNull bool
var checkedSingleNull, singleNull, mixedECNull bool
// Shallow copy input
versions = append(make([][]xlMetaV2ShallowVersion, 0, len(versions)), versions...)
@@ -2019,19 +2019,24 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions
// Version IDs match, but otherwise unable to resolve.
// We are either strict, or don't have enough information to match.
// Switch to a pure counting algo.
latest, latestCount = countXLV2Versions(tops, ver.header, latest, strict)
latest, latestCount = countXLV2Versions(tops, ver.header, latest, strict, false)
break
}
}
if latestCount < quorum {
if !checkedSingleNull {
singleNull = singleNullVersionStreams(originalVersions)
singleNull = singleNullVersionStreams(originalVersions, false)
// Availability-optimized writes can change EC while a drive
// is offline. Only latest-only non-strict reads may recount
// these ordinary null versions across separate modern EC groups.
mixedECNull = !singleNull && !strict && requestedVersions == 1 &&
singleNullVersionStreams(originalVersions, true)
checkedSingleNull = true
}
if singleNull {
if singleNull || mixedECNull {
// A newer minority at the end can hide an older quorum from
// the selection loop. Recount before discarding the null ID.
if candidate, count := countXLV2Versions(tops, latest.header, latest, strict); count >= quorum {
if candidate, count := countXLV2Versions(tops, latest.header, latest, strict, mixedECNull); count >= quorum {
latest, latestCount = candidate, count
}
}
@@ -2098,7 +2103,8 @@ func mergeXLV2Versions(quorum int, strict bool, requestedVersions int, versions
// singleNullVersionStreams excludes histories and other version types from the
// additional recount. Check the original inputs, before any stream is pruned.
func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool {
// Mixed EC groups require known modern EC headers on every non-empty stream.
func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion, allowMixedEC bool) bool {
var ec xlMetaV2VersionHeader
var haveEC bool
for _, stream := range versions {
@@ -2112,7 +2118,10 @@ func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool {
if h.VersionID != [16]byte{} || h.Type != ObjectType || h.FreeVersion() {
return false
}
if haveEC && (h.EcN != ec.EcN || h.EcM != ec.EcM) {
if allowMixedEC && !h.hasEC() {
return false
}
if !allowMixedEC && haveEC && (h.EcN != ec.EcN || h.EcM != ec.EcM) {
return false
}
ec, haveEC = h, true
@@ -2121,16 +2130,18 @@ func singleNullVersionStreams(versions [][]xlMetaV2ShallowVersion) bool {
}
// countXLV2Versions selects the most frequent compatible header for reference's
// VersionID, retaining the existing sort tiebreak and last matching entry.
func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionHeader, latest xlMetaV2ShallowVersion, strict bool) (xlMetaV2ShallowVersion, int) {
// VersionID and the last matching entry. The mixed-EC null-version recount adds
// a stable EC tiebreak without changing the ordinary counting path.
func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionHeader, latest xlMetaV2ShallowVersion, strict, allowMixedEC bool) (xlMetaV2ShallowVersion, int) {
x := make(map[xlMetaV2VersionHeader]int, len(tops))
for _, a := range tops {
if a.header.VersionID != reference.VersionID {
continue
}
if !strict {
// we must match EC, when we are not strict.
if !a.header.matchesEC(reference) {
// The narrow null-version recount can consider other EC groups,
// but EcN/EcM stay in the map key: their votes never combine.
if !allowMixedEC && !a.header.matchesEC(reference) {
continue
}
a.header.Signature = [4]byte{}
@@ -2142,10 +2153,20 @@ func countXLV2Versions(tops []xlMetaV2ShallowVersion, reference xlMetaV2VersionH
if v < latestCount {
continue
}
if v == latestCount && latest.header.sortsBefore(k) {
// Tiebreak, use sort.
if v == latestCount {
if allowMixedEC {
current := latest.header
current.Signature = [4]byte{}
// Signatures are not part of these groups. Use EC counts as
// the final tiebreak when the existing sort fields are equal.
if current.sortsBefore(k) || !k.sortsBefore(current) &&
(current.EcM > k.EcM || current.EcM == k.EcM && current.EcN > k.EcN) {
continue
}
} else if latest.header.sortsBefore(k) {
continue
}
}
for _, a := range tops {
hdr := a.header
if !strict {