From 6ef3a52fc314f07c3d8450a976a7d0639b8b211c Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Mon, 9 Nov 2026 23:53:43 +0800 Subject: [PATCH] Enforce Object Lock during S3 bulk version deletion Signed-off-by: Feng Ruohang --- .github/workflows/go.yml | 2 +- .github/workflows/release.yml | 2 +- CHANGELOG.md | 19 +- cmd/bucket-handlers.go | 133 ++++++--- cmd/bulk-delete-object-lock-race_test.go | 346 +++++++++++++++++++++++ docs/metrics/healthcheck/README.md | 6 +- 6 files changed, 459 insertions(+), 49 deletions(-) create mode 100644 cmd/bulk-delete-object-lock-race_test.go diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 0d06cf211..26bddb978 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -127,7 +127,7 @@ jobs: - name: Run Object Lock authorization and overwrite tests under race detector env: MINIO_API_REQUESTS_MAX: "10000" - run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|ReplicaDeleteObjectLock|PoolsReplicaDeleteObjectLock|BulkDeleteDirectoryObjectLock|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m + run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|ReplicaDeleteObjectLock|PoolsReplicaDeleteObjectLock|BulkDeleteDirectoryObjectLock|BulkDeleteObjectLock|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m - name: Run maintenance storage timeout and compatibility tests under race detector env: diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 05510776c..bbc2d1067 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -74,7 +74,7 @@ jobs: - name: Gate release on Object Lock authorization and overwrite tests under race detector env: MINIO_API_REQUESTS_MAX: "10000" - run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|ReplicaDeleteObjectLock|PoolsReplicaDeleteObjectLock|BulkDeleteDirectoryObjectLock|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m + run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|ReplicaDeleteObjectLock|PoolsReplicaDeleteObjectLock|BulkDeleteDirectoryObjectLock|BulkDeleteObjectLock|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m - name: Compute release variables env: diff --git a/CHANGELOG.md b/CHANGELOG.md index 274135c7f..eec03251d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -76,7 +76,12 @@ Object Lock, IAM and batch jobs: - Keep the target's legal hold and retention on a replicated delete; a source purge that bypassed GOVERNANCE retention is refused by a target copy that still carries it, and the source retries. Check a directory object's lock - before a bulk delete removes its null version. + before a bulk delete removes its null version. S3 `DeleteObjects` checks + physical version removals under the mutation lock and merges protection + across pools, including concurrent legal-hold and retention updates. These + removals in Object Lock buckets use the single-object path sequentially; + large version-purge batches can take longer. Marker creation and ordinary + buckets retain the bulk path. - Batch jobs enforce the submitter's data permissions throughout execution and leave Object Lock protected versions alone ([#310](https://github.com/pgsty/silo/issues/310), [#311](https://github.com/pgsty/silo/issues/311)). Key rotation keeps other @@ -142,6 +147,18 @@ Upgrade notes: while any peer is old or unreachable. Pause new restores until every node is upgraded. Multi-pool lifecycle protection is complete only after a full upgrade. +- Stop batch submissions and drain or cancel unfinished jobs before changing + versions. Unfinished legacy jobs require resubmission after all nodes upgrade; + completed jobs stay completed. Submitter-bound manifests and progress reports + use a new format that 20261026 cannot execute or resume. Do not downgrade + with pending or paused new jobs. An old node can still execute old-format + jobs, so mixed-version execution is not protected by the new authorization + rules. Preserve job definitions and reconcile completed work before resubmitting. +- `healthcheck ready` remains a process/dependency probe and can succeed before + storage or IAM load. `cluster` reports quorum, which can precede a restarted + node reconnecting its remote drives. Before resuming applications, run bounded + signed write/read checks through every node; neither HTTP probe alone proves + data-path readiness. Preserve peer discovery during bootstrap. - `helm upgrade --reuse-values` keeps the previously stored `image.tag`; set it explicitly when upgrading such a release. diff --git a/cmd/bucket-handlers.go b/cmd/bucket-handlers.go index 4bb7cec0f..23dd5b9cd 100644 --- a/cmd/bucket-handlers.go +++ b/cmd/bucket-handlers.go @@ -494,6 +494,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, } deleteResults := make([]deleteResult, len(deleteObjectsReq.Objects)) + duplicateOf := make(map[int]int) vc, _ := globalBucketVersioningSys.Get(bucket) oss := make([]*objSweeper, len(deleteObjectsReq.Objects)) @@ -536,7 +537,19 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, VersionSuspended: vc.Suspended(), } - if replicateDeletes || object.VersionID != "" && hasLockEnabled || !globalTierConfigMgr.Empty() { + // Physical version deletes in a locked bucket are checked by the + // single-object path below, while its mutation lock is held. A lookup + // here could miss a later protection update or a protected pool copy. + if hasLockEnabled && object.VersionID != "" { + if first, ok := objectsToDelete[object]; ok { + duplicateOf[index] = first + } else { + objectsToDelete[object] = index + } + continue + } + + if replicateDeletes || !globalTierConfigMgr.Empty() { if !globalTierConfigMgr.Empty() && object.VersionID == "" && opts.VersionSuspended { opts.VersionID = nullVersionID } @@ -565,21 +578,10 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, object.ReplicateDecisionStr = dsc.String() } } - if object.VersionID != "" && hasLockEnabled { - if err := enforceRetentionBypassForDelete(ctx, r, bucket, object, goi, gerr); err != nil { - apiErr := toAPIError(ctx, err) - deleteResults[index].errInfo = DeleteError{ - Code: apiErr.Code, - Message: apiErr.Description, - Key: object.ObjectName, - VersionID: object.VersionID, - } - continue - } - } - // Avoid duplicate objects, we use map to filter them out. - if _, ok := objectsToDelete[object]; !ok { + if first, ok := objectsToDelete[object]; ok { + duplicateOf[index] = first + } else { objectsToDelete[object] = index } } @@ -598,10 +600,57 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, ctx = bgContext(ctx) deleteList := toNames(objectsToDelete) - dObjects, errs := deleteObjectsFn(ctx, bucket, deleteList, ObjectOptions{ - PrefixEnabledFn: vc.PrefixEnabled, - VersionSuspended: vc.Suspended(), - }) + dObjects := make([]DeletedObject, len(deleteList)) + errs := make([]error, len(deleteList)) + bulk := make([]ObjectToDelete, 0, len(deleteList)) + bulkIndex := make([]int, 0, len(deleteList)) + for i, object := range deleteList { + if !hasLockEnabled || object.VersionID == "" { + bulk = append(bulk, object) + bulkIndex = append(bulkIndex, i) + continue + } + reqInfo.ObjectName, reqInfo.VersionID = object.ObjectName, object.VersionID + opts := ObjectOptions{ + VersionID: object.VersionID, + Versioned: vc.PrefixEnabled(object.ObjectName), + VersionSuspended: vc.Suspended(), + } + index := objectsToDelete[object] + if !globalTierConfigMgr.Empty() { + oss[index] = newObjSweeper(bucket, object.ObjectName).WithVersion(object.VersionID).WithVersioning(opts.Versioned, opts.VersionSuspended) + } + opts.SetEvalRetentionBypassFn(func(oi ObjectInfo, err error) error { + return enforceRetentionBypassForDelete(ctx, r, bucket, object, oi, err) + }) + opts.SetEvalMetadataFn(func(oi *ObjectInfo, err error) (ReplicateDecision, error) { + if err == nil && oss[index] != nil { + oss[index].SetTransitionState(oi.TransitionedObject) + } + if replicateDeletes { + return checkReplicateDelete(ctx, bucket, object, *oi, opts, err), nil + } + return ReplicateDecision{}, nil + }) + // DeleteObject merges protection across pools before evaluating these + // callbacks and holds the same lock as retention and legal-hold writes. + oi, err := objectAPI.DeleteObject(ctx, bucket, object.ObjectName, opts) + dObjects[i] = DeletedObject{ + ObjectName: object.ObjectName, + VersionID: object.VersionID, + ReplicationState: oi.ReplicationState(), + } + errs[i] = err + } + if len(bulk) > 0 { + deleted, bulkErrs := deleteObjectsFn(ctx, bucket, bulk, ObjectOptions{ + PrefixEnabledFn: vc.PrefixEnabled, + VersionSuspended: vc.Suspended(), + }) + for i, index := range bulkIndex { + dObjects[index], errs[index] = deleted[i], bulkErrs[i] + } + } // Are all objects saying bucket not found? if isAllBucketsNotFound(errs) { @@ -610,23 +659,11 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, } for i := range errs { - // DeleteMarkerVersionID is not used specifically to avoid - // lookup errors, since DeleteMarkerVersionID is only - // created during DeleteMarker creation when client didn't - // specify a versionID. - objToDel := ObjectToDelete{ - ObjectV: ObjectV{ - ObjectName: decodeDirObject(dObjects[i].ObjectName), - VersionID: dObjects[i].VersionID, - }, - VersionPurgeStatus: dObjects[i].VersionPurgeStatus(), - VersionPurgeStatuses: dObjects[i].ReplicationState.VersionPurgeStatusInternal, - DeleteMarkerReplicationStatus: dObjects[i].ReplicationState.ReplicationStatusInternal, - ReplicateDecisionStr: dObjects[i].ReplicationState.ReplicateDecisionStr, - } - dindex := objectsToDelete[objToDel] + // Match by the input identity: replication decisions may have changed + // under the mutation lock, and a missing version has no stored identity. + dindex := objectsToDelete[deleteList[i]] if errs[i] == nil || isErrObjectNotFound(errs[i]) || isErrVersionNotFound(errs[i]) { - if replicateDeletes { + if replicateDeletes && (!hasLockEnabled || deleteList[i].VersionID == "") { dObjects[i].ReplicationState = deleteList[i].ReplicationState() } deleteResults[dindex].delInfo = dObjects[i] @@ -640,20 +677,28 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, VersionID: deleteList[i].VersionID, } } + for index, first := range duplicateOf { + deleteResults[index] = deleteResults[first] + } // Generate response deleteErrors := make([]DeleteError, 0, len(deleteObjectsReq.Objects)) deletedObjects := make([]DeletedObject, 0, len(deleteObjectsReq.Objects)) - for _, deleteResult := range deleteResults { + eventObjects := make([]DeletedObject, 0, len(deleteObjectsReq.Objects)) + for index, deleteResult := range deleteResults { + // Storage normalizes a directory to the null version. The response + // retains the caller's version identity, including an explicit null. + if isDirObject(deleteObjectsReq.Objects[index].ObjectName) { + deleteResult.delInfo.VersionID = deleteObjectsReq.Objects[index].VersionID + deleteResult.errInfo.VersionID = deleteObjectsReq.Objects[index].VersionID + } if deleteResult.errInfo.Code != "" { deleteErrors = append(deleteErrors, deleteResult.errInfo) } else { - // All deletes on directory objects was with `nullVersionID`. - // Remove it from response. - if isDirObject(deleteResult.delInfo.ObjectName) && deleteResult.delInfo.VersionID == nullVersionID { - deleteResult.delInfo.VersionID = "" - } deletedObjects = append(deletedObjects, deleteResult.delInfo) + if _, duplicate := duplicateOf[index]; !duplicate { + eventObjects = append(eventObjects, deleteResult.delInfo) + } } } @@ -662,7 +707,7 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, // Write success response. writeSuccessResponseXML(w, encodedSuccessResponse) - for _, dobj := range deletedObjects { + for _, dobj := range eventObjects { if dobj.ObjectName == "" { continue } @@ -705,8 +750,8 @@ func (api objectAPIHandlers) DeleteMultipleObjectsHandler(w http.ResponseWriter, } // Clean up transitioned objects from remote tier - for _, os := range oss { - if os == nil { // skip objects that weren't deleted due to invalid versionID etc. + for index, os := range oss { + if os == nil || deleteResults[index].errInfo.Code != "" { // Sweep only successful deletions. continue } os.Sweep() diff --git a/cmd/bulk-delete-object-lock-race_test.go b/cmd/bulk-delete-object-lock-race_test.go new file mode 100644 index 000000000..4db1fb068 --- /dev/null +++ b/cmd/bulk-delete-object-lock-race_test.go @@ -0,0 +1,346 @@ +// Copyright (c) 2026 PGSTY +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "bytes" + "context" + "encoding/xml" + "fmt" + "net/http" + "net/http/httptest" + "net/url" + "strings" + "sync" + "testing" + "time" + + minio "github.com/minio/minio-go/v7" + "github.com/minio/minio/internal/bucket/replication" + xhttp "github.com/minio/minio/internal/http" + "github.com/minio/mux" +) + +// Schedule a protection update immediately before the storage mutation. Both +// public delete entry points are gated so the test is independent of whether +// the handler batches a version or uses the single-object deletion path. +type bulkDeleteLockBarrier struct { + ObjectLayer + entered chan struct{} + release chan struct{} + once sync.Once +} + +func (g *bulkDeleteLockBarrier) DeleteObjects(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error) { + g.wait() + return g.ObjectLayer.DeleteObjects(ctx, bucket, objects, opts) +} + +func (g *bulkDeleteLockBarrier) wait() { + g.once.Do(func() { close(g.entered); <-g.release }) +} + +func (g *bulkDeleteLockBarrier) DeleteObject(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) { + g.wait() + return g.ObjectLayer.DeleteObject(ctx, bucket, object, opts) +} + +func bulkDeleteLockRequest(t *testing.T, bucket, key, version string) *http.Request { + t.Helper() + body := encodeResponse(DeleteObjectsRequest{Objects: []ObjectToDelete{{ObjectV: ObjectV{ObjectName: key, VersionID: version}}}}) + req, err := newTestSignedRequestV4(http.MethodPost, getDeleteMultipleObjectsURL("", bucket), int64(len(body)), bytes.NewReader(body), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + return mux.SetURLVars(req, map[string]string{"bucket": bucket}) +} + +func TestBulkDeleteObjectLockInterleaving(t *testing.T) { + for _, pools := range []int{1, 2} { + for _, directory := range []bool{false, true} { + for _, kind := range []string{"legal-hold", "retention"} { + t.Run(fmt.Sprintf("pools=%d/directory=%v/%s", pools, directory, kind), func(t *testing.T) { + z, _ := bulkDeletePools(t, pools) + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true}) + if err != nil { + t.Fatal(err) + } + key := "protection-update" + if directory { + key += "/" + } + var oi ObjectInfo + if directory { + oi, err = z.PutObject(t.Context(), bucket, key, mustGetPutObjReader(t, bytes.NewReader(nil), 0, "", ""), ObjectOptions{Versioned: true, VersionID: nullVersionID}) + if err != nil { + t.Fatal(err) + } + } else { + oi = batchPutVersion(t, z, bucket, key, nil) + } + gate := &bulkDeleteLockBarrier{ObjectLayer: z, entered: make(chan struct{}), release: make(chan struct{})} + api := objectAPIHandlers{ObjectAPI: func() ObjectLayer { return gate }} + requestVersion := oi.VersionID + if directory { + requestVersion = "" + } + req := bulkDeleteLockRequest(t, bucket, key, requestVersion) + rec := httptest.NewRecorder() + done := make(chan struct{}) + go func() { defer close(done); api.DeleteMultipleObjectsHandler(rec, req) }() + var release sync.Once + unblock := func() { release.Do(func() { close(gate.release) }) } + defer func() { + unblock() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Error("delete did not finish") + } + }() + select { + case <-gate.entered: + case <-time.After(10 * time.Second): + t.Fatal("bulk handler did not reach deletion after validation") + } + body := []byte(`ON`) + if kind == "retention" { + body = retentionBody("COMPLIANCE", UTCNow().Add(24*time.Hour)) + } + holdReq, err := newTestSignedRequestV4(http.MethodPut, "/"+bucket+"/"+key+"?"+kind+"&versionId="+url.QueryEscape(oi.VersionID), int64(len(body)), bytes.NewReader(body), globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + holdRec := httptest.NewRecorder() + router.ServeHTTP(holdRec, holdReq) + protected, getErr := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: oi.VersionID}) + unblock() + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("delete did not finish") + } + if holdRec.Code != http.StatusOK || getErr != nil { + t.Fatalf("protection setup failed: status=%d body=%s err=%v", holdRec.Code, holdRec.Body.String(), getErr) + } + if kind == "legal-hold" && protected.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] != "ON" { + t.Fatal("legal hold was not stored") + } + if kind == "retention" && protected.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] != "COMPLIANCE" { + t.Fatal("retention was not stored") + } + var response DeleteObjectsResponse + if err := xml.Unmarshal(rec.Body.Bytes(), &response); err != nil { + t.Fatal(err) + } + exists := batchVersionExists(t, z, bucket, key, oi.VersionID) + t.Logf("protection PUT=%d; delete POST=%d response=%s; protected_version_exists=%v", holdRec.Code, rec.Code, rec.Body.String(), exists) + if rec.Code != http.StatusOK || !exists || len(response.Errors) != 1 || len(response.DeletedObjects) != 0 { + t.Errorf("bulk deletion removed a version after its protection update succeeded") + } + }) + } + } + } +} + +func TestBulkDeleteObjectLockPoolCopies(t *testing.T) { + for _, kind := range []string{"COMPLIANCE", "LEGALHOLD"} { + t.Run(kind, func(t *testing.T) { + z, _ := bulkDeletePools(t, 2) + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true}) + if err != nil { + t.Fatal(err) + } + for _, bulk := range []bool{false, true} { + key := "single-control" + if bulk { + key = "bulk" + } + id := mustGetUUID() + mtime := UTCNow().Add(-time.Hour) + putConsistencyObject(t, z, bucket, key, 0, "same-data", ObjectOptions{Versioned: true, VersionID: id, MTime: mtime, UserDefined: nil}) + putConsistencyObject(t, z, bucket, key, 1, "same-data", ObjectOptions{Versioned: true, VersionID: id, MTime: mtime, UserDefined: batchLockMeta(kind)}) + selected, err := z.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: id}) + if err != nil { + t.Fatal(err) + } + if selected.UserDefined[strings.ToLower(xhttp.AmzObjectLockMode)] != "" || selected.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] != "" { + t.Fatal("fixture did not select the unprotected pool copy") + } + req := bulkDeleteLockRequest(t, bucket, key, id) + if !bulk { + req, err = newTestSignedRequestV4(http.MethodDelete, getDeleteObjectURL("", bucket, key)+"?versionId="+id, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil) + if err != nil { + t.Fatal(err) + } + } + rec := httptest.NewRecorder() + router.ServeHTTP(rec, req) + if !strings.Contains(rec.Body.String(), "Object is WORM protected") { + t.Fatalf("expected Object Lock refusal: HTTP %d: %s", rec.Code, rec.Body.String()) + } + remaining := 0 + for _, pool := range z.serverPools { + _, err := pool.GetObjectInfo(t.Context(), bucket, key, ObjectOptions{VersionID: id}) + if err == nil { + remaining++ + } else if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) { + t.Fatal(err) + } + } + t.Logf("bulk=%v HTTP=%d response=%s remaining_pool_copies=%d", bulk, rec.Code, rec.Body.String(), remaining) + if remaining != 2 { + t.Errorf("protected version copies removed by bulk=%v", bulk) + } + } + }) + } +} + +// One response must retain the identity and outcome of every requested item, +// even when physical version removals and marker creation take different paths. +func TestBulkDeleteObjectLockMixedResponse(t *testing.T) { + for _, pools := range []int{1, 2} { + for _, quiet := range []bool{false, true} { + t.Run(fmt.Sprintf("pools=%d/quiet=%v", pools, quiet), func(t *testing.T) { + z, _ := bulkDeletePools(t, pools) + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true}) + if err != nil { + t.Fatal(err) + } + locked := batchPutVersion(t, z, bucket, "locked", batchLockMeta("COMPLIANCE")) + free := batchPutVersion(t, z, bucket, "free", nil) + marker := batchPutVersion(t, z, bucket, "marker", batchLockMeta("LEGALHOLD")) + if _, err := z.PutObject(t.Context(), bucket, "directory/", mustGetPutObjReader(t, bytes.NewReader(nil), 0, "", ""), ObjectOptions{Versioned: true, VersionID: nullVersionID}); err != nil { + t.Fatal(err) + } + missing := mustGetUUID() + objects := []ObjectToDelete{ + {ObjectV: ObjectV{ObjectName: "locked", VersionID: locked.VersionID}}, + {ObjectV: ObjectV{ObjectName: "free", VersionID: free.VersionID}}, + {ObjectV: ObjectV{ObjectName: "missing", VersionID: missing}}, + {ObjectV: ObjectV{ObjectName: "marker"}}, + {ObjectV: ObjectV{ObjectName: "directory/"}}, + {ObjectV: ObjectV{ObjectName: "directory/", VersionID: nullVersionID}}, + {ObjectV: ObjectV{ObjectName: "invalid", VersionID: "invalid"}}, + } + objects = append(objects, objects[0], objects[1]) + body := encodeResponse(DeleteObjectsRequest{Quiet: quiet, Objects: objects}) + rec := r5Request(t, router, globalActiveCred, http.MethodPost, getDeleteMultipleObjectsURL("", bucket), string(body), nil) + var response DeleteObjectsResponse + if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &response) != nil { + t.Fatalf("HTTP %d: %s", rec.Code, rec.Body.String()) + } + wantErrors := []ObjectV{objects[0].ObjectV, objects[6].ObjectV, objects[0].ObjectV} + if len(response.Errors) != len(wantErrors) { + t.Fatalf("errors: %+v", response.Errors) + } + for i, e := range response.Errors { + if e.Key != wantErrors[i].ObjectName || e.VersionID != wantErrors[i].VersionID { + t.Errorf("error identity %d: %+v", i, e) + } + } + if quiet { + if len(response.DeletedObjects) != 0 { + t.Errorf("quiet successes: %+v", response.DeletedObjects) + } + } else { + want := []ObjectV{objects[1].ObjectV, objects[2].ObjectV, objects[3].ObjectV, objects[4].ObjectV, objects[5].ObjectV, objects[1].ObjectV} + if len(response.DeletedObjects) != len(want) { + t.Fatalf("successes: %+v", response.DeletedObjects) + } + for i, d := range response.DeletedObjects { + if d.ObjectName != want[i].ObjectName || d.VersionID != want[i].VersionID { + t.Errorf("success identity %d: %+v", i, d) + } + if i == 2 && (!d.DeleteMarker || d.DeleteMarkerVersionID == "") { + t.Errorf("missing delete marker: %+v", d) + } + } + } + if !batchVersionExists(t, z, bucket, "locked", locked.VersionID) || !batchVersionExists(t, z, bucket, "marker", marker.VersionID) || batchVersionExists(t, z, bucket, "free", free.VersionID) || batchVersionExists(t, z, bucket, "directory/", nullVersionID) { + t.Fatal("unexpected stored versions after mixed deletion") + } + }) + } + } +} + +func TestBulkDeleteObjectLockGovernance(t *testing.T) { + for _, pools := range []int{1, 2} { + t.Run(fmt.Sprintf("pools=%d", pools), func(t *testing.T) { + z, _ := bulkDeletePools(t, pools) + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true}) + if err != nil { + t.Fatal(err) + } + deleteOnly := newObjectAttributesAuthzUser(t, "bulk-governance", bucket, `"s3:DeleteObjectVersion"`) + for _, allowed := range []bool{false, true} { + for _, bypass := range []bool{false, true} { + key := fmt.Sprintf("allowed-%v-bypass-%v", allowed, bypass) + oi := batchPutVersion(t, z, bucket, key, batchLockMeta("GOVERNANCE")) + cred := deleteOnly + if allowed { + cred = globalActiveCred + } + body := encodeResponse(DeleteObjectsRequest{Objects: []ObjectToDelete{{ObjectV: ObjectV{ObjectName: key, VersionID: oi.VersionID}}}}) + header := map[string]string{} + if bypass { + header[xhttp.AmzObjectLockBypassGovernance] = "true" + } + rec := r5Request(t, router, cred, http.MethodPost, getDeleteMultipleObjectsURL("", bucket), string(body), header) + var response DeleteObjectsResponse + if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &response) != nil { + t.Fatalf("HTTP %d: %s", rec.Code, rec.Body.String()) + } + deleted := allowed && bypass + if batchVersionExists(t, z, bucket, key, oi.VersionID) == deleted || (len(response.Errors) == 0) != deleted { + t.Fatalf("%s: %s", key, rec.Body.String()) + } + } + } + }) + } +} + +// The locked path computes replication decisions while holding the mutation +// lock. Keep pending purges and their queue entries, including duplicate input. +func TestBulkDeleteObjectLockReplication(t *testing.T) { + for _, pools := range []int{1, 2} { + t.Run(fmt.Sprintf("pools=%d", pools), func(t *testing.T) { + z, _ := bulkDeletePools(t, pools) + bucket, router, err := initAPIHandlerTest(t.Context(), z, nil, MakeBucketOptions{VersioningEnabled: true, LockEnabled: true}) + if err != nil { + t.Fatal(err) + } + client, err := minio.New("localhost:1", &minio.Options{Region: "us-east-1", MaxRetries: 1}) + if err != nil { + t.Fatal(err) + } + worker, cleanup := r5ReplicationFixture(t, z, bucket, client) + defer cleanup() + free := batchPutVersion(t, z, bucket, "free", nil) + locked := batchPutVersion(t, z, bucket, "locked", batchLockMeta("LEGALHOLD")) + objects := []ObjectToDelete{{ObjectV: ObjectV{ObjectName: "free", VersionID: free.VersionID}}, {ObjectV: ObjectV{ObjectName: "locked", VersionID: locked.VersionID}}} + objects = append(objects, objects[0]) + body := encodeResponse(DeleteObjectsRequest{Objects: objects}) + rec := r5Request(t, router, globalActiveCred, http.MethodPost, getDeleteMultipleObjectsURL("", bucket), string(body), nil) + var response DeleteObjectsResponse + if rec.Code != http.StatusOK || xml.Unmarshal(rec.Body.Bytes(), &response) != nil || len(response.Errors) != 1 || len(response.DeletedObjects) != 2 { + t.Fatalf("HTTP %d: %s", rec.Code, rec.Body.String()) + } + if len(worker) != 1 { + t.Fatalf("want one purge queued, got %d", len(worker)) + } + op, ok := (<-worker).(DeletedObjectReplicationInfo) + if !ok || op.ObjectName != "free" || op.VersionID != free.VersionID || op.VersionPurgeStatus() != replication.VersionPurgePending || op.ReplicationState.ReplicateDecisionStr == "" { + t.Fatalf("lost purge identity/state: %+v", op) + } + if !batchVersionExists(t, z, bucket, "locked", locked.VersionID) { + t.Fatal("protected version was deleted") + } + }) + } +} diff --git a/docs/metrics/healthcheck/README.md b/docs/metrics/healthcheck/README.md index ef72ee15a..88fb19136 100644 --- a/docs/metrics/healthcheck/README.md +++ b/docs/metrics/healthcheck/README.md @@ -29,13 +29,15 @@ healthcheck: retries: 5 ``` -`ready` turns healthy as soon as the process accepts connections, before storage and IAM finish loading, so that distributed nodes can reach each other while they start. To hold back a dependent service until the server answers S3 requests (for example Compose `depends_on: condition: service_healthy`), probe `cluster` instead. +`ready` can return success before storage and IAM finish loading. It checks the process and, when configured, KMS and etcd; it does not certify S3 data access. Keep this bootstrap behavior so distributed peers can discover and connect to one another. Kubernetes headless peer services must retain `publishNotReadyAddresses: true`. + +`cluster` checks initialization and reported write quorum, but a restarted node can still be reconnecting its remote drives after that check succeeds. Neither endpoint is a sufficient gate for an application that immediately reads or writes S3. Compose `depends_on: condition: service_healthy` inherits the image probe's limited meaning. Keep the node probe, and gate the dependent application with a bounded signed S3 check: write a uniquely named small object through each node, then read every acknowledged object through every node and verify its bytes and version. Retry within a fixed overall deadline with per-request timeouts, and clean up the probe objects. Failure should stop the rollout or dependent startup. This samples the tested keys and erasure sets; it does not prove spare failure capacity or that every drive is online. See the [startup window](#startup-readiness-window) below. `silo healthcheck --maintenance cluster` answers the pre-drain question documented below: exit `0` when the node can be taken down safely, exit `1` (HTTP 412) when doing so would lose HA. Keep the `cluster` checks out of per-container liveness probes — they reflect cluster-wide quorum, not this process. ## Liveness probe -This probe always responds with '200 OK'. Only fails if 'etcd' is configured and unreachable. When liveness probe fails, Kubernetes like platforms restart the container. +This probe returns `200 OK` while the process accepts requests, including before the object layer initializes. It can return `503` under request-queue saturation and does not check KMS or etcd. Kubernetes uses repeated liveness failures to restart the container, so configure its startup budget and failure threshold accordingly. ``` livenessProbe: