diff --git a/cmd/batch-expire-purge-authz_test.go b/cmd/batch-expire-purge-authz_test.go index 6e3dfdedd..92ab295c3 100644 --- a/cmd/batch-expire-purge-authz_test.go +++ b/cmd/batch-expire-purge-authz_test.go @@ -15,6 +15,7 @@ import ( "time" "github.com/minio/minio/internal/auth" + "github.com/minio/minio/internal/bucket/lifecycle" "github.com/minio/minio/internal/config/storageclass" xhttp "github.com/minio/minio/internal/http" "github.com/pgsty/silo-pkg/v3/policy" @@ -190,6 +191,41 @@ func TestBatchExpirePurgeRefusesLockedAfterWalk(t *testing.T) { } } +// transition marks version versionID of key as transitioned to a remote +// tier, as a lifecycle transition leaves it. +func (f *batchAuthzFixture) transition(t *testing.T, key, versionID string) { + t.Helper() + set := f.obj.(*erasureServerPools).serverPools[0] + infos, errs := readAllFileInfo(f.ctx, set.getHashedSet(key).getDisks(), "", f.bucket, key, versionID, false, false) + mustIAM(t, errs[0]) + fi := infos[0] + fi.TransitionStatus, fi.TransitionTier, fi.TransitionedObjName = lifecycle.TransitionComplete, "WARM", "remote/"+key+"/"+versionID + mustIAM(t, set.DecomTieredObject(f.ctx, f.bucket, key, fi, ObjectOptions{})) +} + +// A key with a transitioned version is refused as a whole, as a key with a +// locked version is: the job completes with the key counted as refused, not +// failed, and the key keeps every version. +func TestBatchExpirePurgeRefusesTieredKey(t *testing.T) { + f := newBatchAuthzFixture(t) + batchAuthzPolicy(f.ctx, t, f.sys, "batch-full", batchAuthzFullAccess) + cred := iamAdminUser(f.ctx, t, f.sys, "batch-user", "batch-full") + const key = "purge/tiered" + older := f.putVersions(t, key) + f.transition(t, key, older) + want := batchPurgeVersionIDs(t, f, key) + ri := f.runChanged(t, cred, "purge/", func() {}) + for _, id := range want { + if !f.versionExists(t, key, id) { + t.Errorf("version %s of a key with a transitioned version was purged", id) + } + } + if !ri.Complete || ri.Failed || ri.KeysRefused != 1 || ri.ObjectsFailed != 0 || ri.Objects != 0 { + t.Errorf("job complete=%v failed=%v refused=%d failed objects=%d objects=%d, want complete with 1 refused key", + ri.Complete, ri.Failed, ri.KeysRefused, ri.ObjectsFailed, ri.Objects) + } +} + // Controls: a purge by root and by an authorized user removes every version, // including one added after the walk. func TestBatchExpirePurgeAuthorizedControls(t *testing.T) { diff --git a/cmd/batch-expire.go b/cmd/batch-expire.go index 542bd39ad..cf1a607bf 100644 --- a/cmd/batch-expire.go +++ b/cmd/batch-expire.go @@ -474,10 +474,10 @@ func (r *BatchJobExpire) expireVersions(ctx context.Context, api ObjectLayer, vc // expireAllVersions purges every version of a key whose rule retains none, // with the purge of exactly that key (erasureServerPools.deleteAllVersions), // while the version the walk matched is still its latest. Under its lock, -// before it deletes anything, the purge refuses the key when Object Lock -// protects a version and has every write it would delete, each copy and -// each revision a drive holds, authorized for the submitter, with the -// version ID and tags it has then. +// before it deletes anything, the purge refuses the key when a version is +// transitioned or Object Lock protects one, and has every write it would +// delete, each copy and each revision a drive holds, authorized for the +// submitter, with the version ID and tags it has then. // // What the purge removed is counted at once; a refused key is counted in res // as refused, and the versions a failed purge left in res as failures. A @@ -520,9 +520,9 @@ func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer, case errors.Is(err, errDeleteAllChanged): // Changed since the walk matched it: not expired. return nil - case isErrObjectLocked(err): + case isErrObjectLocked(err), errors.Is(err, errDeleteAllTiered): // Refused as a whole: nothing removed, not retried and not a - // failure. + // failure. A later run of the job checks the key again. res.keysRefused++ return nil case errors.Is(err, errDeleteAllBudget) && purge.Removed > 0: @@ -533,9 +533,6 @@ func (r *BatchJobExpire) expireAllVersions(ctx context.Context, api ObjectLayer, return err } batchLogIf(ctx, fmt.Errorf("Failed to expire %s/%s due to %v (attempts=%d)", exp.Bucket, exp.Name, err, attempts)) - if errors.Is(err, errDeleteAllTiered) { - break // refused; a later run of the job checks again - } } res.expireAllFailed(exp, objects, markers) return nil diff --git a/cmd/erasure-server-pool-delete-all_test.go b/cmd/erasure-server-pool-delete-all_test.go index 1e7c24e27..60e7e253f 100644 --- a/cmd/erasure-server-pool-delete-all_test.go +++ b/cmd/erasure-server-pool-delete-all_test.go @@ -816,8 +816,8 @@ func TestPurgeAllVersionsRefusesTieredKeys(t *testing.T) { if got := p.remotes(); got != remotes { t.Errorf("tier holds %d remote objects, want %d", got, remotes) } - if ri.Objects != 0 || ri.ObjectsFailed == 0 { - t.Errorf("refused batch purges counted objects=%d failed=%d", ri.Objects, ri.ObjectsFailed) + if ri.Objects != 0 || ri.ObjectsFailed != 0 || ri.KeysRefused != 2 { + t.Errorf("refused batch purges counted objects=%d failed=%d refused=%d, want 2 refused keys", ri.Objects, ri.ObjectsFailed, ri.KeysRefused) } }