From dcc2f20ba81c51a11dab047bb65c86e0dc6a6038 Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Mon, 26 Oct 2026 22:43:51 +0800 Subject: [PATCH] Preserve maintenance timeout behavior and prepare SILO 20261026 Signed-off-by: Feng Ruohang --- .github/workflows/go.yml | 7 +- CHANGELOG.md | 46 +-- CONTRIBUTING.md | 22 +- Dockerfile.goreleaser | 2 +- README.md | 10 +- README_ZH.md | 10 +- buildscripts/install-mcli.sh | 2 +- cmd/abandoned-mutation_test.go | 2 + cmd/erasure-object.go | 30 +- cmd/erasure-write-generation_test.go | 2 + cmd/maintenance-read-compatibility_test.go | 3 +- cmd/object-api-utils.go | 5 +- cmd/object-mutation-consistency_test.go | 4 +- cmd/object-mutation-helpers-strict_test.go | 57 +++ cmd/object-mutation-isolation_test.go | 439 --------------------- cmd/object-mutation-lease_test.go | 2 + cmd/object-mutation.go | 218 ---------- cmd/rename-cancellation_test.go | 2 + cmd/storage-rest-client.go | 30 +- cmd/storage-rollback_test.go | 139 +++++++ cmd/storage-timeout_test.go | 228 +++++++++++ cmd/xl-storage-disk-id-check.go | 20 +- cmd/xl-storage-format-v2.go | 4 - cmd/xl-storage.go | 56 +-- go.mod | 8 +- go.sum | 16 +- helm/silo/values.yaml | 2 +- internal/grid/connection.go | 6 +- internal/grid/handlers.go | 3 +- internal/grid/muxclient.go | 16 +- internal/grid/muxclient_timeout_test.go | 82 ++-- internal/ioutil/deadline_test.go | 62 +++ internal/ioutil/ioutil.go | 15 - internal/ioutil/ioutil_test.go | 53 --- 34 files changed, 634 insertions(+), 969 deletions(-) create mode 100644 cmd/object-mutation-helpers-strict_test.go delete mode 100644 cmd/object-mutation-isolation_test.go delete mode 100644 cmd/object-mutation.go create mode 100644 cmd/storage-rollback_test.go create mode 100644 cmd/storage-timeout_test.go create mode 100644 internal/ioutil/deadline_test.go diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml index 466542492..6b7824a24 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/go.yml @@ -118,10 +118,13 @@ jobs: MINIO_API_REQUESTS_MAX: "10000" run: go test -race -tags kqueue,dev ./cmd -run '^Test(APIPutObjectRetention|APIVersionAddressedWrite|PutObjectRetention|DirectoryObjectLock|ObjectLockOverwrite|APICopyObjectPreservesVersionAfterSuspension|MultipartNullObjectLock|PoolsObjectLockOverwrite|MultipartVersioningTransitions|PlanReview|RecoveredMultipartUpload)' -count=1 -timeout=5m - - name: Run maintenance write, rollback and cleanup tests under race detector + - name: Run maintenance storage timeout and compatibility tests under race detector env: MINIO_API_REQUESTS_MAX: "10000" - run: go test -race -tags kqueue,dev ./cmd ./internal/grid ./internal/ioutil -run '^Test(Mutation|RenameCancellation|PutObjectGenerationFollowsCommitLock|MaintenancePartialFailedOverwriteAvailability|SingleMutationWaitsBeyondDefaultTimeout|WithoutDefaultTimeoutStillEndsOnDisconnectOrCancellation|WithDeadlineAndWait)' -count=1 -timeout=5m + run: go test -race -tags kqueue,dev ./cmd ./internal/grid ./internal/ioutil -run '^Test(StorageTimeout|StorageRollback|StoragePrerelease|MaintenancePartialFailedOverwriteAvailability|SingleRequest|WithDeadline)' -count=1 -timeout=5m + + - name: Compile deferred consistency diagnostics + run: go test -tags kqueue,dev,consistency_strict ./cmd -run '^$' crosscompile: name: Cross Compile diff --git a/CHANGELOG.md b/CHANGELOG.md index 08ca35036..9d6b2c443 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,9 +1,9 @@ # Changelog -## RELEASE.2026-10-25T00-00-00Z +## RELEASE.2026-10-26T00-00-00Z -Package version: `20261025000000.0.0`. This source is prepared for a Draft -Release; public Server artifacts and images remain at 20261018 until promotion. +Package version: `20261026000000.0.0`. Publication status and artifacts are +listed in [GitHub Releases](https://github.com/pgsty/silo/releases). - Fix LDAP login for long normalized DNs without hiding unreadable IAM revocation records. Only length errors for valid keys with an ASCII path @@ -11,13 +11,17 @@ Release; public Server artifacts and images remain at 20261018 until promotion. Thanks to [@910patin](https://github.com/910patin) for the original fix and reproduction in [#250](https://github.com/pgsty/silo/pull/250). - Build the server and standalone debugging tools with Go 1.27.3 and refresh - Go dependencies, including silo-pkg v3.15.1, the published mcli 20261025 - source and upstream minio-go `ab93db7ba435` for multipart, read/seek and - batch-delete fixes. Retain output-format and NetBSD compatibility selections. -- Embed the published Console v2.5.0 source `989fa7150792`, including its - GUI update, diagnostic WebSocket closing handshake and complete health-report - decoding. Bundle published mcli 20261025 in container builds and select it - for the verified client installer and Helm client jobs. + Go dependencies, including silo-pkg v3.15.2, mcli 20261026 + and upstream minio-go `320af050c445` for multipart, read/seek, + batch-delete and bucket-replication removal fixes. Retain output-format and NetBSD compatibility selections. +- Embed Console v2.5.1 source `6a05e8d6ad67`, including its GUI update, + diagnostic WebSocket closing handshake, complete health-report decoding, + group-membership preservation, OIDC group permissions, TLS-proxy session + cookies, credential exports and accessibility fixes. +- Bundle mcli 20261026 in container builds, the verified client installer and + Helm client jobs. Include client identity isolation, resync status and rate + fixes, and bounded batch-status fallback. Scripts must handle the additive + batch `pending` state and nonzero exits for partial resync mutations. - Read current bucket inventory in account-info responses so Console refreshes immediately reflect successful creation and deletion. This management endpoint now performs a live list instead of returning a cached inventory. @@ -39,24 +43,20 @@ Release; public Server artifacts and images remain at 20261018 until promotion. and multi-pool deletion contributions. - Preserve Kafka's default broker protocol 2.1 and terminal-width behavior; fix legacy DARE v1 key unsealing and verify DARE v1/v2 and MinKMS TLS compatibility. -- Wait for admitted storage mutations to finish before releasing local guards, - deciding rollback or cleaning staging data. Rollback verifies ownership and - restores only its overwritten version, preserving later writes and metadata - edits. Abandoned-data cleanup retains backups still needed by commit/undo. - Generate PUT timestamps under the commit lock. Keep the prior read-quorum - and healing policy, including readability after a failed partial overwrite. +- Forward storage RPC delete options, including the previous data directory + needed by overwrite rollback. Preserve existing drive/RPC timeouts, request + cancellation, object locking, read-quorum and healing behavior. - Update bundled curl to 8.23.0 and require patched static libraries. Apply [Red Hat's compromised-CA mitigation](https://access.redhat.com/security/vulnerabilities/RHSB-2026-012); the root blocklist also rejects legitimate chains under the affected roots. - Ship Windows platform archives as ZIP; Linux and macOS retain tar.gz. -Storage limits remain: this does not provide distributed fencing after lease -loss or network partition, eliminate stale reads after partial metadata writes, -or repair previously inconsistent objects. Stricter read refusal and additional -cross-pool recovery remain deferred; their diagnostic tests are retained under -`consistency_strict`. Waiting for admitted I/O can extend same-object waits on -a stuck drive, and interrupted cleanup can retain rollback backups. Upgrade all -writers before relying on the new rollback ownership checks. +Storage timeouts retain their existing meaning: a returned timeout does not +prove that the underlying disk operation stopped or that a write did not +commit. Late commits or delayed rollback can affect newer writes; stale reads +remain possible after partial failures, lease loss or network partitions. This +release does not provide distributed fencing or repair previously inconsistent +objects. These source changes are not included in the published 20261018 artifacts. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index e9ef251a4..34a894632 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -53,15 +53,19 @@ Before opening a pull request: - Explain any preserved `MINIO_*`, `minio_*`, `x-minio-*`, `/minio/*`, `.minio.sys`, ARN, module/import-path, or serialized compatibility name. -The maintenance branch retains the existing read-quorum and healing policy. -Deferred strict-read and cross-pool recovery tests are preserved behind the -`consistency_strict` build tag. Run them explicitly with -`go test -tags kqueue,dev,consistency_strict ./cmd`; known failures must be -reported separately, not described as fixed or included in a passing result. -They cover late commits after lease loss, stale metadata reads, and stricter -read refusal. The normal suite covers the retained write/undo/cleanup fixes -and continued readability after a failed partial overwrite. This scope does -not provide distributed fencing or repair previously inconsistent objects. +The maintenance branch retains existing storage timeouts, cancellation, +object locking, read-quorum and healing behavior. The normal suite checks local +and RPC timeout/cancellation, remote rollback options, and readability after a +failed partial overwrite. A timeout does not prove that a disk worker stopped +or that a write did not commit. + +Stronger mutation lifetime, rollback ownership, generation ordering, strict-read +and cross-pool recovery diagnostics are preserved behind `consistency_strict`. +Compile them with `go test -tags kqueue,dev,consistency_strict ./cmd -run '^$'` +and run individual scenarios explicitly. They describe guarantees outside the +maintenance contract and may fail; report their results separately from the +normal suite. They must not be used to claim distributed fencing or repair of +previously inconsistent objects. ### Commit changes diff --git a/Dockerfile.goreleaser b/Dockerfile.goreleaser index 7bb4bad49..5a2fbc284 100644 --- a/Dockerfile.goreleaser +++ b/Dockerfile.goreleaser @@ -17,7 +17,7 @@ ENV GOPATH=/go ENV CGO_ENABLED=0 ARG MC_REPO=pgsty/mc -ARG MC_VERSION=RELEASE.2026-10-25T00-00-00Z +ARG MC_VERSION=RELEASE.2026-10-26T00-00-00Z ARG MC_AMD64_SHA256=54bd131039dad6677f083a75207b28f82818c067c8e65fae1973f51510d268e1 ARG MC_ARM64_SHA256=9bb294f1912f8c6b9ac08c9a0b8d2964b86306a7c440b987ee5f0570d18de3c5 diff --git a/README.md b/README.md index e05f8387c..bf8f338ef 100644 --- a/README.md +++ b/README.md @@ -43,11 +43,11 @@ This release strengthens Object Lock and IAM authorization, makes scoped STS rev Read the [release notes](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-18T00-00-00Z) before upgrading, especially the removal of strict multipart listing, coordinated server upgrades, and measured performance costs. See [CONTRIBUTORS.md](CONTRIBUTORS.md) for the people behind the code, reports, and documentation. -The 20261025 source is prepared for a Draft Release with Go 1.27.3, -silo-pkg v3.15.1, released mcli 20261025, Console v2.5.0 (`989fa7150792`), -and upstream minio-go `ab93db7ba435`. It includes the Server IAM cache repair -needed by Console's **Reset to default** action and bounded write/rollback/cleanup -repairs. Read the [changes and remaining storage limits](CHANGELOG.md#release2026-10-25t00-00-00z). +The 20261026 maintenance release is in preparation with Go 1.27.3, +silo-pkg v3.15.2, mcli 20261026, Console v2.5.1 (`6a05e8d6ad67`), +and upstream minio-go `320af050c445`. It includes the Server IAM cache repair +needed by Console's **Reset to default** action and remote rollback-option propagation, preserving existing storage +timeouts and cancellation behavior. Read the [changes and remaining storage limits](CHANGELOG.md#release2026-10-26t00-00-00z). The published artifacts above retain their original component versions. ## Overview diff --git a/README_ZH.md b/README_ZH.md index 97a38d30b..bbfc22eec 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -43,11 +43,11 @@ 升级前请阅读[发行说明](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-18T00-00-00Z)，特别是 strict 分片列表的移除、服务端协调升级要求与实测性能成本。代码、问题报告和文档贡献者见 [CONTRIBUTORS.md](CONTRIBUTORS.md)。 -20261025 源码正在准备 Draft Release：使用 Go 1.27.3、silo-pkg v3.15.1、 -已发布的 mcli 20261025、Console v2.5.0（`989fa7150792`）与上游 minio-go -`ab93db7ba435`。其中包含 Console **Reset to default（恢复默认）** 所需的 -Server IAM 缓存修复，以及限定范围的写入、回滚与清理修复。 -请阅读[变更与仍存的存储限制](CHANGELOG.md#release2026-10-25t00-00-00z)； +20261026 维护版本正在准备中：使用 Go 1.27.3、silo-pkg v3.15.2、 +mcli 20261026、Console v2.5.1（`6a05e8d6ad67`）与上游 minio-go +`320af050c445`。其中包含 Console **Reset to default（恢复默认）** 所需的 +Server IAM 缓存修复，以及远端回滚参数透传修复；保持既有存储超时与取消行为。 +请阅读[变更与仍存的存储限制](CHANGELOG.md#release2026-10-26t00-00-00z)； 上述已发布制品仍保留原有组件版本。 ## 概述 diff --git a/buildscripts/install-mcli.sh b/buildscripts/install-mcli.sh index c7bdf91b4..c4a83b5e8 100755 --- a/buildscripts/install-mcli.sh +++ b/buildscripts/install-mcli.sh @@ -40,7 +40,7 @@ if [ -n "${MCLI_BIN:-}" ]; then exit 0 fi -release=${MCLI_RELEASE:-RELEASE.2026-10-25T00-00-00Z} +release=${MCLI_RELEASE:-RELEASE.2026-10-26T00-00-00Z} version_hyphen=${release#RELEASE.} package_version=$(printf '%s\n' "${version_hyphen}" | sed -E 's/^([0-9]{4})-([0-9]{2})-([0-9]{2})T([0-9]{2})-([0-9]{2})-([0-9]{2})Z$/\1\2\3\4\5\6.0.0/') if [ "${package_version}" = "${version_hyphen}" ]; then diff --git a/cmd/abandoned-mutation_test.go b/cmd/abandoned-mutation_test.go index 2822eaf43..d80da8866 100644 --- a/cmd/abandoned-mutation_test.go +++ b/cmd/abandoned-mutation_test.go @@ -1,3 +1,5 @@ +//go:build consistency_strict + // Copyright (c) 2026 Feng Ruohang // SPDX-License-Identifier: AGPL-3.0-or-later diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go index 16dc41326..5c9ee5d80 100644 --- a/cmd/erasure-object.go +++ b/cmd/erasure-object.go @@ -1094,19 +1094,11 @@ func (er erasureObjects) getObjectInfoAndQuorum(ctx context.Context, bucket, obj // Similar to rename but renames data from srcEntry to dstEntry at dataDir func renameData(ctx context.Context, disks []StorageAPI, srcBucket, srcEntry string, metadata []FileInfo, dstBucket, dstEntry string, writeQuorum int) ([]StorageAPI, []byte, string, error) { - if err := ctx.Err(); err != nil { - return nil, nil, "", err - } - // Admission is cancelable. After admission use actual disk results for - // commit/rollback, even if the client or namespace lease is canceled. - ctx = context.WithoutCancel(ctx) g := errgroup.WithNErrs(len(disks)) fvID := mustGetUUID() for index := range disks { - metadata[index].Metadata = maps.Clone(metadata[index].Metadata) metadata[index].SetTierFreeVersionID(fvID) - metadata[index].Metadata[rollbackIDKey] = fvID } diskVersions := make([][]byte, len(disks)) @@ -1564,18 +1556,6 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st compIndex = opts.IndexCB() } - if !opts.NoLock { - lk := er.NewNSLock(bucket, object) - lkctx, err := lk.GetLock(ctx, globalOperationTimeout) - if err != nil { - return ObjectInfo{}, err - } - ctx = lkctx.Context() - defer lk.Unlock(lkctx) - } - - // Mint the generation while holding the commit lock. Staged uploads may - // acquire the lock in a different order from their encoding completion. modTime := opts.MTime if opts.MTime.IsZero() { modTime = UTCNow() @@ -1672,6 +1652,16 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st } } + if !opts.NoLock { + lk := er.NewNSLock(bucket, object) + lkctx, err := lk.GetLock(ctx, globalOperationTimeout) + if err != nil { + return ObjectInfo{}, err + } + ctx = lkctx.Context() + defer lk.Unlock(lkctx) + } + if !opts.ReplicationRequest && !opts.ReplicaLockReconcile && !opts.DataMovement && (fi.VersionID == "" || fi.VersionID == nullVersionID) { getObjectInfo := er.getObjectInfo diff --git a/cmd/erasure-write-generation_test.go b/cmd/erasure-write-generation_test.go index f59913c4e..32cba743f 100644 --- a/cmd/erasure-write-generation_test.go +++ b/cmd/erasure-write-generation_test.go @@ -1,3 +1,5 @@ +//go:build consistency_strict + // Copyright (c) 2026 Feng Ruohang // // This file is part of Silo Object Storage. diff --git a/cmd/maintenance-read-compatibility_test.go b/cmd/maintenance-read-compatibility_test.go index 8e37ecb52..b9be5f580 100644 --- a/cmd/maintenance-read-compatibility_test.go +++ b/cmd/maintenance-read-compatibility_test.go @@ -26,8 +26,7 @@ func (d compatibilityFailedOverwrite) RenameData(ctx context.Context, sv, sp str return RenameDataResp{}, errDiskNotFound } -// Maintenance compatibility requirement: after a completely joined failed -// overwrite with one residue, keep the previously acknowledged quorum readable. +// Maintenance compatibility requirement: after a failed overwrite with one residue, keep the previously acknowledged quorum readable. // This tests a known history; it does not claim the resulting snapshot proves it. func TestMaintenancePartialFailedOverwriteAvailability(t *testing.T) { // Lock RPC tests configure a single-drive backend (EC:0). Pin this diff --git a/cmd/object-api-utils.go b/cmd/object-api-utils.go index be9fd14f1..dbe202f50 100644 --- a/cmd/object-api-utils.go +++ b/cmd/object-api-utils.go @@ -399,7 +399,10 @@ func getCompleteMultipartMD5(parts []CompletePart) string { return s3MD5 } -// Clean unwanted fields from metadata +// Keep metadata written by pre-release builds out of public object metadata. +const rollbackIDKey = ReservedMetadataPrefixLower + "rollback-id" + +// Clean unwanted fields from metadata. func cleanMetadata(metadata map[string]string) map[string]string { // Remove STANDARD StorageClass metadata = removeStandardStorageClass(metadata) diff --git a/cmd/object-mutation-consistency_test.go b/cmd/object-mutation-consistency_test.go index cbf86a090..41ad13e77 100644 --- a/cmd/object-mutation-consistency_test.go +++ b/cmd/object-mutation-consistency_test.go @@ -1,3 +1,5 @@ +//go:build consistency_strict + // Copyright (c) 2026 Feng Ruohang // // This file is part of Silo Object Storage. @@ -472,7 +474,7 @@ func TestMutationDeleteJoinsNestedWorker(t *testing.T) { defer globalDriveConfig.Update(drive.Config{MaxTimeout: oldTimeout}) done := make(chan error, 1) go func() { - _, e := xioutil.WithDeadlineAndWait(context.Background(), globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (struct{}, error) { + _, e := xioutil.WithDeadline(context.Background(), globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (struct{}, error) { return struct{}{}, disk.storage.DeleteVersion(ctx, bucket, object, fi, false, DeleteOptions{UndoWrite: true}) }) done <- e diff --git a/cmd/object-mutation-helpers-strict_test.go b/cmd/object-mutation-helpers-strict_test.go new file mode 100644 index 000000000..bbe804e16 --- /dev/null +++ b/cmd/object-mutation-helpers-strict_test.go @@ -0,0 +1,57 @@ +//go:build consistency_strict + +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "bytes" + "maps" +) + +// Compare logical version state, excluding per-shard checksums/index, physical +// DataDir placement and properties of other versions in the same xl.meta. +// Different DataDir values remain readable for the existing rebalance recovery +// contract; rollback separately requires an exact DataDir match. +func sameFileInfoVersion(a, b FileInfo) bool { + if a.VersionID != b.VersionID || a.Deleted != b.Deleted || a.XLV1 != b.XLV1 || a.Size != b.Size || + a.TransitionStatus != b.TransitionStatus || a.TransitionTier != b.TransitionTier || + a.TransitionedObjName != b.TransitionedObjName || a.TransitionVersionID != b.TransitionVersionID || + !sameVersionMetadata(a.Metadata, b.Metadata) || !bytes.Equal(a.Checksum, b.Checksum) || len(a.Parts) != len(b.Parts) { + return false + } + if !a.Deleted && a.Size != 0 && !a.Erasure.Equal(b.Erasure) { + return false + } + for i, p := range a.Parts { + q := b.Parts[i] + if p.Number != q.Number || p.Size != q.Size || p.ActualSize != q.ActualSize || p.ETag != q.ETag || !bytes.Equal(p.Index, q.Index) { + return false + } + } + return true +} + +// The undo capability controls rollback ownership, not the logical contents. +// A partial no-op update can revoke it without changing a single public field. +func sameVersionMetadata(a, b map[string]string) bool { + if a[rollbackIDKey] == b[rollbackIDKey] { + return maps.Equal(a, b) + } + for k, v := range a { + if k != rollbackIDKey { + if other, ok := b[k]; !ok || other != v { + return false + } + } + } + for k := range b { + if k != rollbackIDKey { + if _, ok := a[k]; !ok { + return false + } + } + } + return true +} diff --git a/cmd/object-mutation-isolation_test.go b/cmd/object-mutation-isolation_test.go deleted file mode 100644 index 072f8453e..000000000 --- a/cmd/object-mutation-isolation_test.go +++ /dev/null @@ -1,439 +0,0 @@ -// Copyright (c) 2026 Feng Ruohang -// -// This file is part of Silo Object Storage. -// -// 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 . - -package cmd - -import ( - "context" - "errors" - "fmt" - "path/filepath" - "slices" - "sync" - "sync/atomic" - "testing" - "time" -) - -// Keep the old slot calculation here to reproduce the false sharing between -// distinct resources, including resources on separate storage drives. -func mutationGuardLegacySlot(drive, path string) uint64 { - h := uint64(14695981039346656037) - for _, part := range []string{drive, path} { - for i := range part { - h = (h ^ uint64(part[i])) * 1099511628211 - } - h *= 1099511628211 - } - return h % 65536 -} - -func mutationGuardCollision(t *testing.T, drive string, target uint64) string { - t.Helper() - for i := 0; i < 2000000; i++ { - key := fmt.Sprintf("independent-%d", i) - if mutationGuardLegacySlot(drive, pathJoin("bucket", key)) == target { - return key - } - } - t.Fatal("no collision found in bounded search") - return "" -} - -func TestMutationGuardIsolatesStorageResources(t *testing.T) { - blocked, rootA, err := newXLStorageTestSetup(t) - if err != nil { - t.Fatal(err) - } - defer blocked.Close() - healthy, rootB, err := newXLStorageTestSetup(t) - if err != nil { - t.Fatal(err) - } - defer healthy.Close() - defer mutationGuardCheckEmpty(t, rootA, rootB) - for _, disk := range []*xlStorageDiskIDCheck{blocked, healthy} { - if err := disk.MakeVol(t.Context(), "bucket"); err != nil { - t.Fatal(err) - } - } - const object = "blocked" - target := mutationGuardLegacySlot(rootA, pathJoin("bucket", object)) - type mutation struct { - name, key string - disk *xlStorageDiskIDCheck - done chan error - } - mutations := []mutation{ - {name: "same-key", key: object, disk: blocked}, - {name: "different-key-same-drive-collision", key: mutationGuardCollision(t, rootA, target), disk: blocked}, - {name: "different-key-healthy-drive-collision", key: mutationGuardCollision(t, rootB, target), disk: healthy}, - {name: "same-path-healthy-drive", key: object, disk: healthy}, - } - fixture := func(name string) FileInfo { - fi := newFileInfo(name, 2, 2) - fi.DataDir, fi.ModTime, fi.Erasure.Index = mustGetUUID(), UTCNow(), 1 - fi.Fresh = true - fi.Metadata = map[string]string{"x-amz-meta-probe": "before"} - return fi - } - for _, m := range mutations { - if err := m.disk.WriteMetadata(t.Context(), "", "bucket", m.key, fixture(m.key)); err != nil { - t.Fatal(err) - } - } - ctx, cancel := context.WithCancel(t.Context()) - defer cancel() - var lastSuccess int64 - gate := &stagedRenameContext{ - Context: ctx, meta: filepath.Join(rootA, minioMetaTmpBucket, "staged", xlStorageFormatFile), - entered: make(chan struct{}), release: make(chan struct{}), - tracker: healthDiskCtxValue{lastSuccess: &lastSuccess}, - } - var once sync.Once - release := func() { once.Do(func() { close(gate.release) }) } - var workers sync.WaitGroup - defer func() { - release() - workers.Wait() - }() - holderDone := make(chan error, 1) - workers.Go(func() { - _, err := blocked.RenameData(gate, minioMetaTmpBucket, "staged", fixture(object), "bucket", object, RenameOptions{}) - holderDone <- err - }) - select { - case <-gate.entered: - case <-time.After(5 * time.Second): - t.Fatal("RenameData did not reach the staged metadata barrier") - } - // Cancellation cannot release the guard while the admitted worker can - // still commit. Unrelated mutations must nevertheless be able to finish. - cancel() - for i := range mutations { - m := &mutations[i] - m.done = make(chan error, 1) - workers.Go(func() { - m.done <- m.disk.UpdateMetadata(t.Context(), "bucket", m.key, FileInfo{ - Metadata: map[string]string{"x-amz-meta-probe": "after"}, - }, UpdateMetadataOpts{}) - }) - } - mutationGuardWaitRefs(t, rootA, pathJoin("bucket", object), 2) - for _, m := range mutations[1:] { - select { - case err := <-m.done: - if err != nil { - t.Errorf("%s: %v", m.name, err) - } - t.Logf("%s completed while RenameData remained paused", m.name) - case <-time.After(2 * time.Second): - t.Errorf("%s blocked by an unrelated paused mutation", m.name) - } - } - select { - case err := <-holderDone: - t.Fatalf("canceled RenameData returned before its worker finished: %v", err) - default: - } - select { - case err := <-mutations[0].done: - t.Fatalf("same-key mutation escaped the held guard: %v", err) - default: - } - release() - select { - case err := <-holderDone: - if !errors.Is(err, context.Canceled) { - t.Errorf("canceled RenameData result: %v", err) - } - case <-time.After(5 * time.Second): - t.Fatal("RenameData did not finish after releasing the barrier") - } - select { - case err := <-mutations[0].done: - if err != nil { - t.Fatal(err) - } - case <-time.After(5 * time.Second): - t.Fatal("same-key mutation did not finish after releasing the holder") - } - workers.Wait() - for _, m := range mutations { - fi, err := m.disk.ReadVersion(t.Context(), "", "bucket", m.key, "", ReadOptions{}) - if err != nil || fi.Metadata["x-amz-meta-probe"] != "after" { - t.Errorf("%s persisted metadata: %q, %v", m.name, fi.Metadata["x-amz-meta-probe"], err) - } - } -} - -func mutationGuardCheckEmpty(t *testing.T, drives ...string) { - t.Helper() - storageMutationLocks.Lock() - defer storageMutationLocks.Unlock() - for key, lock := range storageMutationLocks.locks { - if slices.Contains(drives, key.drive) { - t.Errorf("unused guard retained for %+v with %d references", key, lock.refs) - } - } -} - -func mutationGuardWaitRefs(t *testing.T, drive, path string, refs int) *storageMutationLock { - t.Helper() - deadline := time.NewTimer(5 * time.Second) - defer deadline.Stop() - tick := time.NewTicker(time.Millisecond) - defer tick.Stop() - key := storageMutationKey{drive, path} - for { - storageMutationLocks.Lock() - lock := storageMutationLocks.locks[key] - ready := lock != nil && lock.refs == refs - storageMutationLocks.Unlock() - if ready { - return lock - } - select { - case <-deadline.C: - t.Fatalf("guard %+v did not reach %d references", key, refs) - case <-tick.C: - } - } -} - -func TestMutationGuardRetainsWaitersAndRemovesIdleEntries(t *testing.T) { - s := &xlStorage{drivePath: t.TempDir()} - defer mutationGuardCheckEmpty(t, s.drivePath) - unlock := sync.OnceFunc(s.lockMutations("object")) - defer unlock() - original := mutationGuardWaitRefs(t, s.drivePath, "object", 1) - acquired, release, finished := make(chan struct{}), make(chan struct{}), make(chan struct{}) - unblock := sync.OnceFunc(func() { close(release) }) - defer unblock() - go func() { - defer close(finished) - defer s.lockMutations("object", "object")() - close(acquired) - <-release - }() - if lock := mutationGuardWaitRefs(t, s.drivePath, "object", 2); lock != original { - t.Error("waiting operation registered a second mutex for the same resource") - } - select { - case <-acquired: - t.Error("waiter acquired before the holder released") - default: - } - unlock() - select { - case <-acquired: - case <-time.After(5 * time.Second): - t.Fatal("waiter did not acquire after release") - } - if lock := mutationGuardWaitRefs(t, s.drivePath, "object", 1); lock != original { - t.Error("holder release discarded the queued operation's mutex") - } - unblock() - select { - case <-finished: - case <-time.After(5 * time.Second): - t.Fatal("waiter did not release") - } - mutationGuardCheckEmpty(t, s.drivePath) - // The registry must not grow with the historical object count. - for i := 0; i < 1000; i++ { - s.lockMutations(fmt.Sprintf("object-%d", i))() - } - s.lockMutations()() -} - -func TestMutationGuardOrdersAndDeduplicatesPaths(t *testing.T) { - s := &xlStorage{drivePath: t.TempDir()} - defer mutationGuardCheckEmpty(t, s.drivePath) - paths := []string{"b", "a", "b", "a"} - before := slices.Clone(paths) - unlock := s.lockMutations(paths...) - mutationGuardWaitRefs(t, s.drivePath, "a", 1) - mutationGuardWaitRefs(t, s.drivePath, "b", 1) - unlock() - if !slices.Equal(paths, before) { - t.Fatalf("changed caller's path slice: %v", paths) - } - const workers, iterations = 8, 500 - start, done := make(chan struct{}), make(chan struct{}) - var wg sync.WaitGroup - var protected atomic.Int64 - var active atomic.Int32 - for i := 0; i < workers; i++ { - worker := i - wg.Go(func() { - <-start - order := []string{"a", "b", "a"} - if worker%2 != 0 { - order = []string{"b", "a", "b"} - } - for j := 0; j < iterations; j++ { - unlock := s.lockMutations(order...) - if active.Add(1) != 1 { - t.Error("overlapping guards entered concurrently") - } - protected.Add(1) - active.Add(-1) - unlock() - } - }) - } - go func() { wg.Wait(); close(done) }() - close(start) - select { - case <-done: - if got := protected.Load(); got != workers*iterations { - t.Fatalf("overlapping guards lost updates: got %d, want %d", got, workers*iterations) - } - case <-time.After(5 * time.Second): - t.Fatal("opposing path orders or duplicate paths deadlocked") - } -} - -// Measure registry/allocation cost separately from storage I/O. Independent -// workers use two resources, as RenameData does, and shared workers use one. -func BenchmarkMutationGuard(b *testing.B) { - s := &xlStorage{drivePath: "/benchmark/drive"} - b.Run("single", func(b *testing.B) { - b.ReportAllocs() - for b.Loop() { - s.lockMutations("bucket/object")() - } - }) - b.Run("parallel-independent", func(b *testing.B) { - b.ReportAllocs() - var worker atomic.Uint64 - b.RunParallel(func(pb *testing.PB) { - id := worker.Add(1) - paths := []string{fmt.Sprintf("staging/%d", id), fmt.Sprintf("bucket/object-%d", id)} - for pb.Next() { - s.lockMutations(paths...)() - } - }) - }) - b.Run("parallel-shared", func(b *testing.B) { - b.ReportAllocs() - b.RunParallel(func(pb *testing.PB) { - for pb.Next() { - s.lockMutations("bucket/object")() - } - }) - }) -} - -// A call must retain paths it has not reached yet, even across xlStorage -// instances for the same drive. No resource wait may hold the registry mutex. -func TestMutationGuardRetainsUnacquiredPaths(t *testing.T) { - s := &xlStorage{drivePath: t.TempDir()} - sameDrive := &xlStorage{drivePath: s.drivePath} - defer mutationGuardCheckEmpty(t, s.drivePath) - unlockA := sync.OnceFunc(s.lockMutations("a")) - defer unlockA() - acquired, release, done := make(chan struct{}), make(chan struct{}), make(chan struct{}) - unblock := sync.OnceFunc(func() { close(release) }) - defer unblock() - go func() { - defer close(done) - defer sameDrive.lockMutations("b", "a")() - close(acquired) - <-release - }() - mutationGuardWaitRefs(t, s.drivePath, "a", 2) - original := mutationGuardWaitRefs(t, s.drivePath, "b", 1) - // The queued call has not locked b yet: sorting must put a first. - unlockB := sync.OnceFunc(s.lockMutations("b")) - defer unlockB() - if got := mutationGuardWaitRefs(t, s.drivePath, "b", 2); got != original { - t.Fatal("unacquired path got a different mutex") - } - unlockA() - select { - case <-acquired: - t.Fatal("multi-path call escaped the independently held second path") - default: - } - unlockB() - select { - case <-acquired: - case <-time.After(5 * time.Second): - t.Fatal("multi-path call did not acquire after both paths were released") - } - if got := mutationGuardWaitRefs(t, s.drivePath, "b", 1); got != original { - t.Fatal("release replaced an entry retained by a pending multi-path call") - } - unblock() - <-done - // The variadic helper has no hidden four-path limit. - s.lockMutations("e", "d", "c", "b", "a", "a")() -} - -func TestMutationGuardCanceledWaiterRetainsSerialization(t *testing.T) { - disk, root, err := newXLStorageTestSetup(t) - if err != nil { - t.Fatal(err) - } - defer disk.Close() - defer mutationGuardCheckEmpty(t, root) - if err := disk.MakeVol(t.Context(), "bucket"); err != nil { - t.Fatal(err) - } - fi := newFileInfo("object", 2, 2) - fi.DataDir, fi.ModTime, fi.Erasure.Index = mustGetUUID(), UTCNow(), 1 - fi.Fresh = true - fi.Metadata = map[string]string{"x-amz-meta-probe": "before"} - if err := disk.WriteMetadata(t.Context(), "", "bucket", "object", fi); err != nil { - t.Fatal(err) - } - // Use a separate instance with the same drive identity as the API wrapper. - s := &xlStorage{drivePath: root} - unlock := sync.OnceFunc(s.lockMutations("bucket/object")) - defer unlock() - ctx, cancel := context.WithCancel(t.Context()) - defer cancel() - done := make(chan error, 1) - go func() { - done <- disk.UpdateMetadata(ctx, "bucket", "object", FileInfo{ - Metadata: map[string]string{"x-amz-meta-probe": "canceled"}, - }, UpdateMetadataOpts{}) - }() - mutationGuardWaitRefs(t, root, "bucket/object", 2) - cancel() - select { - case err := <-done: - t.Fatalf("canceled API returned with worker still queued: %v", err) - case <-time.After(50 * time.Millisecond): - } - mutationGuardWaitRefs(t, root, "bucket/object", 2) - unlock() - select { - case err := <-done: - if !errors.Is(err, context.Canceled) { - t.Fatalf("canceled waiter: %v", err) - } - case <-time.After(5 * time.Second): - t.Fatal("canceled waiter did not finish after holder release") - } - got, err := disk.ReadVersion(t.Context(), "", "bucket", "object", "", ReadOptions{}) - if err != nil || got.Metadata["x-amz-meta-probe"] != "before" { - t.Fatalf("canceled metadata update persisted: %v, %v", got.Metadata, err) - } -} diff --git a/cmd/object-mutation-lease_test.go b/cmd/object-mutation-lease_test.go index 90e4b4004..e51e3db88 100644 --- a/cmd/object-mutation-lease_test.go +++ b/cmd/object-mutation-lease_test.go @@ -1,3 +1,5 @@ +//go:build consistency_strict + // Copyright (c) 2026 Feng Ruohang // // This file is part of Silo Object Storage. diff --git a/cmd/object-mutation.go b/cmd/object-mutation.go deleted file mode 100644 index baa05a5d4..000000000 --- a/cmd/object-mutation.go +++ /dev/null @@ -1,218 +0,0 @@ -// Copyright (c) 2026 Feng Ruohang -// SPDX-License-Identifier: AGPL-3.0-or-later - -package cmd - -import ( - "bytes" - "context" - "errors" - "maps" - "slices" - "strings" - "sync" -) - -// This is an undo capability, not a distributed fencing token. Any subsequent -// mutation of the same version revokes it, including a no-op metadata update. -const rollbackIDKey = ReservedMetadataPrefixLower + "rollback-id" - -type storageMutationKey struct { - drive, path string -} - -type storageMutationLock struct { - mu sync.Mutex - refs int // Holders and waiters; protected by the registry mutex. -} - -// Keep exact resource locks only while they have holders or waiters. The -// registry mutex protects references, never storage I/O or resource waits. -var storageMutationLocks = struct { - sync.Mutex - locks map[storageMutationKey]*storageMutationLock -}{locks: make(map[storageMutationKey]*storageMutationLock)} - -func (s *xlStorage) lockMutations(paths ...string) func() { - // Each call uses one drive. Sorting its unique paths gives all overlapping - // calls the same acquisition order without changing the caller's slice. - paths = slices.Clone(paths) - slices.Sort(paths) - paths = slices.Compact(paths) - drive := s.drivePath - locks := make([]*storageMutationLock, len(paths)) - // Retain every entry before waiting so unlock cannot remove a mutex that - // a queued operation still needs. - storageMutationLocks.Lock() - for i, path := range paths { - key := storageMutationKey{drive, path} - lock := storageMutationLocks.locks[key] - if lock == nil { - lock = &storageMutationLock{} - storageMutationLocks.locks[key] = lock - } - lock.refs++ - locks[i] = lock - } - storageMutationLocks.Unlock() - for _, lock := range locks { - lock.mu.Lock() - } - return func() { - for i := len(locks) - 1; i >= 0; i-- { - locks[i].mu.Unlock() - } - storageMutationLocks.Lock() - for i, path := range paths { - locks[i].refs-- - if locks[i].refs == 0 { - delete(storageMutationLocks.locks, storageMutationKey{drive, path}) - } - } - storageMutationLocks.Unlock() - } -} - -// Compare logical version state, excluding per-shard checksums/index, physical -// DataDir placement and properties of other versions in the same xl.meta. -// Different DataDir values remain readable for the existing rebalance recovery -// contract; rollback separately requires an exact DataDir match. -func sameFileInfoVersion(a, b FileInfo) bool { - if a.VersionID != b.VersionID || a.Deleted != b.Deleted || a.XLV1 != b.XLV1 || a.Size != b.Size || - a.TransitionStatus != b.TransitionStatus || a.TransitionTier != b.TransitionTier || - a.TransitionedObjName != b.TransitionedObjName || a.TransitionVersionID != b.TransitionVersionID || - !sameVersionMetadata(a.Metadata, b.Metadata) || !bytes.Equal(a.Checksum, b.Checksum) || len(a.Parts) != len(b.Parts) { - return false - } - if !a.Deleted && a.Size != 0 && !a.Erasure.Equal(b.Erasure) { - return false - } - for i, p := range a.Parts { - q := b.Parts[i] - if p.Number != q.Number || p.Size != q.Size || p.ActualSize != q.ActualSize || p.ETag != q.ETag || !bytes.Equal(p.Index, q.Index) { - return false - } - } - return true -} - -// The undo capability controls rollback ownership, not the logical contents. -// A partial no-op update can revoke it without changing a single public field. -func sameVersionMetadata(a, b map[string]string) bool { - if a[rollbackIDKey] == b[rollbackIDKey] { - return maps.Equal(a, b) - } - for k, v := range a { - if k != rollbackIDKey { - if other, ok := b[k]; !ok || other != v { - return false - } - } - } - for k := range b { - if k != rollbackIDKey { - if _, ok := a[k]; !ok { - return false - } - } - } - return true -} - -// Legacy callers without an undo capability must match the persisted state, -// rather than transient coordinator-only fields in their original FileInfo. -func samePersistedVersion(current, expected FileInfo) bool { - if current.Metadata[rollbackIDKey] != expected.Metadata[rollbackIDKey] { - return false - } - var meta xlMetaV2 - if err := meta.AddVersion(expected); err != nil { - return false - } - fi, err := meta.ToFileInfo(current.Volume, current.Name, expected.VersionID, true, true) - return err == nil && sameFileInfoVersion(current, fi) -} - -// Caller holds the object's mutation guard. Restore just the overwritten -// version, and persist that decision before removing any external shards. -func (s *xlStorage) undoWrite(ctx context.Context, volume, path string, fi FileInfo, opts DeleteOptions, meta *xlMetaV2) error { - versionID := fi.VersionID - if versionID == "" { - versionID = nullVersionID - } - current, err := meta.ToFileInfo(volume, path, versionID, true, true) - if err != nil { - return err - } - if current.DataDir != fi.DataDir || !current.ModTime.Equal(fi.ModTime) || !samePersistedVersion(current, fi) { - return errFileVersionNotFound - } - var previous xlMetaV2 - var oldVersion *xlMetaV2Version - var oldData []byte - volumeDir, err := s.getVolDir(volume) - if err != nil { - return err - } - if opts.OldDataDir != "" { - buf, err := s.readAllData(ctx, volume, volumeDir, pathJoin(volumeDir, path, opts.OldDataDir, xlStorageFormatFileBackup)) - if err != nil { - return err - } - defer metaDataPoolPut(buf) - if err := previous.LoadOrConvert(buf); err != nil { - return err - } - _, oldVersion, err = previous.findVersionStr(versionID) - if err != nil { - return err - } - oldData = previous.data.find(versionID) - if oldData == nil { - oldData = previous.data.find(oldVersion.getDataDir()) - } - } - dataDir, err := meta.DeleteVersion(fi) - if err != nil { - return err - } - if dataDir != "" { - meta.data.remove(versionID, dataDir) - } - if oldVersion != nil { - if err := meta.addVersion(*oldVersion); err != nil { - return err - } - if oldData != nil { - meta.data.replace(versionID, oldData) - } - if dataDir == oldVersion.getDataDir() { - dataDir = "" - } - } - if len(meta.versions) > 0 { - buf, err := meta.AppendTo(metaDataPoolGet()) - defer metaDataPoolPut(buf) - if err != nil { - return err - } - if err := s.writeAllMeta(ctx, volume, pathJoin(path, xlStorageFormatFile), buf, true); err != nil { - return err - } - } else if err := s.deleteFile(volumeDir, pathJoin(volumeDir, path, xlStorageFormatFile), true, false); err != nil { - return err - } - if dataDir != "" { - if err := s.moveToTrash(pathJoin(volumeDir, path, dataDir), true, false); err != nil && !errors.Is(err, errFileNotFound) { - return err - } - } - if opts.OldDataDir != "" { - backupDir := pathJoin(volumeDir, path, opts.OldDataDir) - Remove(pathJoin(backupDir, xlStorageFormatFileBackup)) - if strings.HasSuffix(opts.OldDataDir, "-rollback") { - Remove(backupDir) - } - } - return nil -} diff --git a/cmd/rename-cancellation_test.go b/cmd/rename-cancellation_test.go index 0c1a071dc..073b60ba1 100644 --- a/cmd/rename-cancellation_test.go +++ b/cmd/rename-cancellation_test.go @@ -1,3 +1,5 @@ +//go:build consistency_strict + // Copyright (c) 2026 Feng Ruohang // // This file is part of Silo Object Storage. diff --git a/cmd/storage-rest-client.go b/cmd/storage-rest-client.go index 6431e9e65..0ba573847 100644 --- a/cmd/storage-rest-client.go +++ b/cmd/storage-rest-client.go @@ -408,11 +408,8 @@ func (client *storageRESTClient) CreateFile(ctx context.Context, origvolume, vol } func (client *storageRESTClient) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) error { - if err := ctx.Err(); err != nil { - return err - } - // As with RenameData, admitted metadata writes must finish before unlock. - ctx = grid.WithoutDefaultTimeout(context.WithoutCancel(ctx)) + ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout()) + defer cancel() _, err := storageWriteMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{ DiskID: *client.diskID.Load(), @@ -425,10 +422,8 @@ func (client *storageRESTClient) WriteMetadata(ctx context.Context, origvolume, } func (client *storageRESTClient) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo, opts UpdateMetadataOpts) error { - if err := ctx.Err(); err != nil { - return err - } - ctx = grid.WithoutDefaultTimeout(context.WithoutCancel(ctx)) + ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout()) + defer cancel() _, err := storageUpdateMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{ DiskID: *client.diskID.Load(), @@ -441,14 +436,8 @@ func (client *storageRESTClient) UpdateMetadata(ctx context.Context, volume, pat } func (client *storageRESTClient) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool, opts DeleteOptions) (err error) { - if opts.UndoWrite { - // Rollback is part of the commit and must finish before unlocking. - ctx = grid.WithoutDefaultTimeout(context.WithoutCancel(ctx)) - } else { - var cancel context.CancelFunc - ctx, cancel = context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout()) - defer cancel() - } + ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout()) + defer cancel() _, err = storageDeleteVersionRPC.Call(ctx, client.gridConn, &DeleteVersionHandlerParams{ DiskID: *client.diskID.Load(), @@ -498,13 +487,6 @@ func (client *storageRESTClient) CheckParts(ctx context.Context, volume string, func (client *storageRESTClient) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string, opts RenameOptions, ) (res RenameDataResp, err error) { - if err := ctx.Err(); err != nil { - return res, err - } - // The remote handler cannot cancel an admitted RenameData. Keep waiting - // for its result so renameData can roll back every completed mutation - // before the caller cleans staging files and releases the namespace lock. - ctx = grid.WithoutDefaultTimeout(context.WithoutCancel(ctx)) params := RenameDataHandlerParams{ DiskID: *client.diskID.Load(), SrcVolume: srcVolume, diff --git a/cmd/storage-rollback_test.go b/cmd/storage-rollback_test.go new file mode 100644 index 000000000..1141ede47 --- /dev/null +++ b/cmd/storage-rollback_test.go @@ -0,0 +1,139 @@ +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "context" + "sync" + "testing" + "testing/synctest" + + "github.com/minio/minio/internal/grid" +) + +type cancelableRenameDisk struct { + StorageAPI + entered chan<- struct{} + release <-chan struct{} +} + +func (d cancelableRenameDisk) RenameData(ctx context.Context, _, _ string, _ FileInfo, _, _ string, _ RenameOptions) (RenameDataResp, error) { + d.entered <- struct{}{} + select { + case <-ctx.Done(): + return RenameDataResp{}, ctx.Err() + case <-d.release: + return RenameDataResp{}, errDiskNotFound + } +} + +func TestStorageTimeoutCoordinatorCancellation(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + entered, release := make(chan struct{}, 4), make(chan struct{}) + defer close(release) + disks := make([]StorageAPI, 4) + metadata := make([]FileInfo, 4) + for i := range disks { + disks[i] = cancelableRenameDisk{entered: entered, release: release} + metadata[i] = FileInfo{Deleted: true} + } + done := make(chan error, 1) + go func() { + _, _, _, err := renameData(ctx, disks, minioMetaTmpBucket, "staged", metadata, "bucket", "object", 3) + done <- err + }() + for range disks { + <-entered + } + cancel() + synctest.Wait() + select { + case err := <-done: + if err == nil { + t.Fatal("canceled commit unexpectedly succeeded") + } + default: + t.Fatal("coordinator did not propagate cancellation to disk calls") + } + }) +} + +type rollbackOptionsDisk struct { + StorageAPI + mu sync.Mutex + opts DeleteOptions +} + +func (d *rollbackOptionsDisk) DeleteVersion(_ context.Context, _, _ string, _ FileInfo, _ bool, opts DeleteOptions) error { + d.mu.Lock() + defer d.mu.Unlock() + d.opts = opts + return nil +} + +func TestStorageRollbackRPCOptions(t *testing.T) { + endpoint, err := NewEndpoint(t.TempDir()) + if err != nil { + t.Fatal(err) + } + disk := &rollbackOptionsDisk{} + globalLocalDrivesMu.Lock() + oldDrives, oldSets := globalLocalDrivesMap, globalLocalSetDrives + globalLocalDrivesMap = map[string]StorageAPI{endpoint.String(): disk} + globalLocalSetDrives = nil + globalLocalDrivesMu.Unlock() + defer func() { + globalLocalDrivesMu.Lock() + globalLocalDrivesMap, globalLocalSetDrives = oldDrives, oldSets + globalLocalDrivesMu.Unlock() + }() + tg, err := grid.SetupTestGrid(2) + if err != nil { + t.Fatal(err) + } + defer tg.Cleanup() + server := storageRESTServer{endpoint: endpoint} + const route = "/rollback-options" + if err := storageDeleteVersionRPC.Register(tg.Managers[1], server.DeleteVersionHandler, route); err != nil { + t.Fatal(err) + } + client := &storageRESTClient{gridConn: tg.Managers[0].Connection(tg.Hosts[1]).Subroute(route)} + client.diskID.Store(new(string)) + want := DeleteOptions{UndoWrite: true, OldDataDir: mustGetUUID()} + if err := client.DeleteVersion(t.Context(), "bucket", "object", FileInfo{}, false, want); err != nil { + t.Fatal(err) + } + disk.mu.Lock() + defer disk.mu.Unlock() + if disk.opts != want { + t.Fatalf("remote handler lost rollback options: got %+v, want %+v", disk.opts, want) + } +} + +func TestStoragePrereleaseMetadataHidden(t *testing.T) { + disk, _, err := newXLStorageTestSetup(t) + if err != nil { + t.Fatal(err) + } + defer disk.Close() + if err := disk.MakeVol(t.Context(), "bucket"); err != nil { + t.Fatal(err) + } + fi := newFileInfo("object", 2, 2) + fi.DataDir, fi.ModTime, fi.Erasure.Index = mustGetUUID(), UTCNow(), 1 + fi.Metadata = map[string]string{rollbackIDKey: mustGetUUID(), "user-key": "value"} + if err := disk.WriteMetadata(t.Context(), "", "bucket", "object", fi); err != nil { + t.Fatal(err) + } + got, err := disk.ReadVersion(t.Context(), "", "bucket", "object", "", ReadOptions{}) + if err != nil { + t.Fatal(err) + } + public := cleanMetadata(got.Metadata) + if _, ok := public[rollbackIDKey]; ok || public["user-key"] != "value" { + t.Fatalf("pre-release metadata exposed or user metadata lost: %+v", public) + } +} diff --git a/cmd/storage-timeout_test.go b/cmd/storage-timeout_test.go new file mode 100644 index 000000000..f6f8dcd5f --- /dev/null +++ b/cmd/storage-timeout_test.go @@ -0,0 +1,228 @@ +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later + +package cmd + +import ( + "context" + "errors" + "sync" + "sync/atomic" + "testing" + "testing/synctest" + "time" + + "github.com/minio/minio/internal/grid" +) + +// TrackDiskHealth makes the first lookup. Pause the real disk worker at its +// next health callback, without requiring a broken drive or a wall-clock sleep. +type timeoutDiskContext struct { + context.Context + calls atomic.Int32 + entered, release chan struct{} + tracker healthDiskCtxValue + pauseDeadline bool + once sync.Once +} + +func (c *timeoutDiskContext) Deadline() (time.Time, bool) { + if c.pauseDeadline { + // Metadata operations create a nested deadline in their disk worker. + c.once.Do(func() { close(c.entered); <-c.release }) + } + return c.Context.Deadline() +} + +func (c *timeoutDiskContext) Value(key any) any { + if _, ok := key.(healthDiskCtxKey); ok { + if !c.pauseDeadline && c.calls.Add(1) == 2 { + close(c.entered) + <-c.release + } + return &c.tracker + } + return c.Context.Value(key) +} + +func TestStorageTimeoutLocal(t *testing.T) { + for _, operation := range []string{"rename", "rename-cancel", "write-metadata", "update-metadata", "undo"} { + t.Run(operation, func(t *testing.T) { + disk, _, err := newXLStorageTestSetup(t) + if err != nil { + t.Fatal(err) + } + defer disk.Close() + if err := disk.MakeVol(t.Context(), "bucket"); err != nil { + t.Fatal(err) + } + fi := newFileInfo("object", 2, 2) + fi.DataDir, fi.ModTime, fi.Erasure.Index = mustGetUUID(), UTCNow(), 1 + fi.Metadata = map[string]string{"test": "original"} + if err := disk.WriteMetadata(t.Context(), "", "bucket", "object", fi); err != nil { + t.Fatal(err) + } + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + var lastSuccess int64 + gate := &timeoutDiskContext{ + Context: ctx, entered: make(chan struct{}), release: make(chan struct{}), + tracker: healthDiskCtxValue{lastSuccess: &lastSuccess}, + pauseDeadline: operation != "rename" && operation != "rename-cancel", + } + unblock := sync.OnceFunc(func() { close(gate.release) }) + defer unblock() + done := make(chan error, 1) + go func() { + var err error + switch operation { + case "rename", "rename-cancel": + _, err = disk.RenameData(gate, minioMetaTmpBucket, "staged", fi, "bucket", "object", RenameOptions{}) + case "write-metadata": + err = disk.WriteMetadata(gate, "", "bucket", "object", fi) + case "update-metadata": + err = disk.UpdateMetadata(gate, "bucket", "object", fi, UpdateMetadataOpts{}) + case "undo": + err = disk.DeleteVersion(gate, "bucket", "object", fi, false, DeleteOptions{UndoWrite: true}) + } + done <- err + }() + select { + case <-gate.entered: + case err := <-done: + t.Fatalf("operation did not reach the disk worker: %v", err) + } + want := context.DeadlineExceeded + if operation == "rename-cancel" { + cancel() + want = context.Canceled + } else { + time.Sleep(globalDriveConfig.GetMaxTimeout() + time.Second) + } + synctest.Wait() + select { + case err := <-done: + if !errors.Is(err, want) { + t.Fatalf("expected %v, got %v", want, err) + } + default: + t.Fatal("operation still waits for the blocked worker after its deadline") + } + // A timed-out worker must not leave a new physical mutation lock + // preventing subsequent work on this object. + if err := disk.WriteMetadata(t.Context(), "", "bucket", "object", fi); err != nil { + t.Fatalf("subsequent metadata write: %v", err) + } + unblock() + // synctest waits for the released worker before disk teardown. + }) + }) + } +} + +func TestStorageTimeoutRPC(t *testing.T) { + for _, operation := range []string{"rename", "inline", "write-metadata", "update-metadata", "undo"} { + for _, deadline := range []bool{false, true} { + mode := "cancel" + if deadline { + mode = "deadline" + } + t.Run(operation+"/"+mode, func(t *testing.T) { + tg, err := grid.SetupTestGrid(2) + if err != nil { + t.Fatal(err) + } + defer tg.Cleanup() + entered, release, finished := make(chan struct{}), make(chan struct{}), make(chan struct{}) + unblock := sync.OnceFunc(func() { close(release) }) + defer unblock() + handle := func() { + close(entered) + <-release + close(finished) + } + const route = "/storage-timeout" + if err := storageRenameDataRPC.Register(tg.Managers[1], func(*RenameDataHandlerParams) (*RenameDataResp, *grid.RemoteErr) { + handle() + return &RenameDataResp{}, nil + }, route); err != nil { + t.Fatal(err) + } + if err := storageRenameDataInlineRPC.Register(tg.Managers[1], func(p *RenameDataInlineHandlerParams) (*RenameDataResp, *grid.RemoteErr) { + defer p.Recycle() + handle() + return &RenameDataResp{}, nil + }, route); err != nil { + t.Fatal(err) + } + meta := func(*MetadataHandlerParams) (grid.NoPayload, *grid.RemoteErr) { + handle() + return grid.NewNoPayload(), nil + } + if err := storageWriteMetadataRPC.Register(tg.Managers[1], meta, route); err != nil { + t.Fatal(err) + } + if err := storageUpdateMetadataRPC.Register(tg.Managers[1], meta, route); err != nil { + t.Fatal(err) + } + if err := storageDeleteVersionRPC.Register(tg.Managers[1], func(p *DeleteVersionHandlerParams) (grid.NoPayload, *grid.RemoteErr) { + if !p.Opts.UndoWrite || p.Opts.OldDataDir != "previous" { + t.Error("rollback options were lost on the wire") + } + return meta(nil) + }, route); err != nil { + t.Fatal(err) + } + client := &storageRESTClient{gridConn: tg.Managers[0].Connection(tg.Hosts[1]).Subroute(route)} + client.diskID.Store(new(string)) + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + if deadline { + var stop context.CancelFunc + ctx, stop = context.WithTimeout(ctx, 500*time.Millisecond) + defer stop() + } + done := make(chan error, 1) + go func() { + var err error + fi := FileInfo{} + switch operation { + case "write-metadata": + err = client.WriteMetadata(ctx, "", "bucket", "object", fi) + case "update-metadata": + err = client.UpdateMetadata(ctx, "bucket", "object", fi, UpdateMetadataOpts{}) + case "undo": + err = client.DeleteVersion(ctx, "bucket", "object", fi, false, DeleteOptions{UndoWrite: true, OldDataDir: "previous"}) + default: + if operation == "inline" { + fi.Data = []byte("payload") + } + _, err = client.RenameData(ctx, minioMetaTmpBucket, "staged", fi, "bucket", "object", RenameOptions{}) + } + done <- err + }() + select { + case <-entered: + case <-time.After(5 * time.Second): + t.Fatal("remote worker did not start") + } + want := context.DeadlineExceeded + if !deadline { + cancel() + want = context.Canceled + } + select { + case err := <-done: + if !errors.Is(err, want) { + t.Fatalf("expected %v, got %v", want, err) + } + case <-time.After(5 * time.Second): + t.Fatal("RPC still waits for the remote worker") + } + unblock() + <-finished + }) + } + } +} diff --git a/cmd/xl-storage-disk-id-check.go b/cmd/xl-storage-disk-id-check.go index 408981c2f..39fe68ff1 100644 --- a/cmd/xl-storage-disk-id-check.go +++ b/cmd/xl-storage-disk-id-check.go @@ -495,7 +495,7 @@ func (p *xlStorageDiskIDCheck) RenameData(ctx context.Context, srcVolume, srcPat if len(fi.Data) > 0 { fi.Data = append(grid.GetByteBufferCap(len(fi.Data))[:0], fi.Data...) } - return xioutil.WithDeadlineAndWait[RenameDataResp](ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (res RenameDataResp, err error) { + return xioutil.WithDeadline[RenameDataResp](ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (res RenameDataResp, err error) { if len(fi.Data) > 0 { defer grid.PutByteBuffer(fi.Data) } @@ -634,12 +634,6 @@ func (p *xlStorageDiskIDCheck) DeleteVersion(ctx context.Context, volume, path s } }() - if opts.UndoWrite { - _, err = xioutil.WithDeadlineAndWait(ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (struct{}, error) { - return struct{}{}, p.storage.DeleteVersion(ctx, volume, path, fi, forceDelMarker, opts) - }) - return err - } w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout()) return w.Run(func() error { return p.storage.DeleteVersion(ctx, volume, path, fi, forceDelMarker, opts) }) } @@ -651,10 +645,8 @@ func (p *xlStorageDiskIDCheck) UpdateMetadata(ctx context.Context, volume, path } defer done(0, &err) - _, err = xioutil.WithDeadlineAndWait(ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (struct{}, error) { - return struct{}{}, p.storage.UpdateMetadata(ctx, volume, path, fi, opts) - }) - return err + w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout()) + return w.Run(func() error { return p.storage.UpdateMetadata(ctx, volume, path, fi, opts) }) } func (p *xlStorageDiskIDCheck) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) (err error) { @@ -664,10 +656,8 @@ func (p *xlStorageDiskIDCheck) WriteMetadata(ctx context.Context, origvolume, vo } defer done(0, &err) - _, err = xioutil.WithDeadlineAndWait(ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (struct{}, error) { - return struct{}{}, p.storage.WriteMetadata(ctx, origvolume, volume, path, fi) - }) - return err + w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout()) + return w.Run(func() error { return p.storage.WriteMetadata(ctx, origvolume, volume, path, fi) }) } func (p *xlStorageDiskIDCheck) ReadVersion(ctx context.Context, origvolume, volume, path, versionID string, opts ReadOptions) (fi FileInfo, err error) { diff --git a/cmd/xl-storage-format-v2.go b/cmd/xl-storage-format-v2.go index 42d683c37..77248f72a 100644 --- a/cmd/xl-storage-format-v2.go +++ b/cmd/xl-storage-format-v2.go @@ -1570,11 +1570,7 @@ func (x *xlMetaV2) UpdateObjectVersion(fi FileInfo) error { if err != nil { return err } - delete(ver.ObjectV2.MetaSys, rollbackIDKey) for k, v := range fi.Metadata { - if k == rollbackIDKey { - continue - } if len(k) > len(ReservedMetadataPrefixLower) && strings.EqualFold(k[:len(ReservedMetadataPrefixLower)], ReservedMetadataPrefixLower) { ver.ObjectV2.MetaSys[k] = []byte(v) } else { diff --git a/cmd/xl-storage.go b/cmd/xl-storage.go index e17c8a5c0..81e415553 100644 --- a/cmd/xl-storage.go +++ b/cmd/xl-storage.go @@ -25,7 +25,6 @@ import ( "errors" "fmt" "io" - "maps" "os" pathutil "path" "path/filepath" @@ -1101,7 +1100,6 @@ func (s *xlStorage) ListDir(ctx context.Context, origvolume, volume, dirPath str } func (s *xlStorage) deleteVersions(ctx context.Context, volume, path string, fis ...FileInfo) error { - defer s.lockMutations(pathJoin(volume, path))() volumeDir, err := s.getVolDir(volume) if err != nil { return err @@ -1310,10 +1308,11 @@ func (s *xlStorage) readAllData(ctx context.Context, volume, volumeDir string, f }) } -func (s *xlStorage) moveToTrash(filePath string, recursive, immediatePurge bool) error { - // The caller must retain its mutation guard until this rename completes. - // Only purging the already detached unique trash directory is asynchronous. - return s.moveToTrashNoDeadline(filePath, recursive, immediatePurge) +func (s *xlStorage) moveToTrash(filePath string, recursive, immediatePurge bool) (err error) { + w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout()) + return w.Run(func() (err error) { + return s.moveToTrashNoDeadline(filePath, recursive, immediatePurge) + }) } // DeleteVersion - deletes FileInfo metadata for path at `xl.meta`. forceDelMarker @@ -1326,8 +1325,6 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F }) } - defer s.lockMutations(pathJoin(volume, path))() - volumeDir, err := s.getVolDir(volume) if err != nil { return err @@ -1348,7 +1345,7 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F metaDataPoolPut(buf) // Never used, return it if fi.Deleted && forceDelMarker { // Create a new xl.meta with a delete marker in it - return s.writeMetadataNoLock(ctx, "", volume, path, fi) + return s.WriteMetadata(ctx, "", volume, path, fi) } s.RLock() @@ -1384,10 +1381,6 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F return err } - if opts.UndoWrite { - return s.undoWrite(ctx, volume, path, fi, opts, &xlMeta) - } - dataDir, err := xlMeta.DeleteVersion(fi) if err != nil { return err @@ -1428,12 +1421,15 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F return s.writeAllMeta(ctx, volume, pathJoin(path, xlStorageFormatFile), buf, true) } + if opts.UndoWrite && opts.OldDataDir != "" { + return renameAll(pathJoin(filePath, opts.OldDataDir, xlStorageFormatFileBackup), pathJoin(filePath, xlStorageFormatFile), filePath) + } + return s.deleteFile(volumeDir, pathJoin(volumeDir, path, xlStorageFormatFile), true, false) } // Updates only metadata for a given version. func (s *xlStorage) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo, opts UpdateMetadataOpts) (err error) { - defer s.lockMutations(pathJoin(volume, path))() if len(fi.Metadata) == 0 { return errInvalidArgument } @@ -1482,16 +1478,6 @@ func (s *xlStorage) UpdateMetadata(ctx context.Context, volume, path string, fi // WriteMetadata - writes FileInfo metadata for path at `xl.meta` func (s *xlStorage) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) (err error) { - defer s.lockMutations(pathJoin(volume, path))() - return s.writeMetadataNoLock(ctx, origvolume, volume, path, fi) -} - -func (s *xlStorage) writeMetadataNoLock(ctx context.Context, origvolume, volume, path string, fi FileInfo) (err error) { - // Any later write of this version revokes the original writer's undo. - if _, ok := fi.Metadata[rollbackIDKey]; ok { - fi.Metadata = maps.Clone(fi.Metadata) - delete(fi.Metadata, rollbackIDKey) - } if fi.Fresh { if origvolume != "" { origvolumeDir, err := s.getVolDir(origvolume) @@ -2554,10 +2540,7 @@ func (s *xlStorage) DeleteBulk(ctx context.Context, volume string, paths ...stri return err } - unlock := s.lockMutations(pathJoin(volume, fp), pathJoin(volume, pathutil.Dir(fp))) - err = s.moveToTrash(filePath, false, false) - unlock() - if err != nil { + if err = s.moveToTrash(filePath, false, false); err != nil { return err } } @@ -2567,7 +2550,6 @@ func (s *xlStorage) DeleteBulk(ctx context.Context, volume string, paths ...stri // Delete - delete a file at path. func (s *xlStorage) Delete(ctx context.Context, volume string, path string, deleteOpts DeleteOptions) (err error) { - defer s.lockMutations(pathJoin(volume, path), pathJoin(volume, pathutil.Dir(path)))() volumeDir, err := s.getVolDir(volume) if err != nil { return err @@ -2597,7 +2579,6 @@ func skipAccessChecks(volume string) (ok bool) { // RenameData - rename source path to destination path atomically, metadata and data directory. func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo, dstVolume, dstPath string, opts RenameOptions) (res RenameDataResp, err error) { - defer s.lockMutations(pathJoin(srcVolume, srcPath), pathJoin(dstVolume, dstPath))() defer func() { ignoredErrs := []error{ errFileNotFound, @@ -2841,12 +2822,6 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f xlMeta.data.remove(reqVID, dataDir) } } - if !healing && res.OldDataDir == "" { - // Shared data must survive both commit and rollback. Keep the - // previous metadata in a separate transaction directory instead - // of using a data directory still owned by another version. - res.OldDataDir = fi.DataDir + "-rollback" - } } if err = xlMeta.AddVersion(fi); err != nil { @@ -3299,7 +3274,6 @@ func (s *xlStorage) CleanAbandonedData(ctx context.Context, volume string, path if volume == "" || path == "" { return nil // Ignore } - defer s.lockMutations(pathJoin(volume, path))() volumeDir, err := s.getVolDir(volume) if err != nil { @@ -3351,14 +3325,6 @@ func (s *xlStorage) CleanAbandonedData(ctx context.Context, volume string, path // Do not abort on context errors. for dir := range foundDirs { toRemove := pathJoin(volumeDir, path, dir+SlashSeparator) - // A completed disk mutation may still be awaiting its quorum result. - // Its previous generation is needed if that transaction must undo. - // Only commit/undo cleanup owns a directory containing this backup. - if _, err := Lstat(pathJoin(toRemove, xlStorageFormatFileBackup)); err == nil { - continue - } else if !osIsNotExist(err) { - return osErrToFileErr(err) - } err = s.deleteFile(volumeDir, toRemove, true, true) diskHealthCheckOK(ctx, err) if err != nil { diff --git a/go.mod b/go.mod index 1a6f6d4d6..f8acd9baa 100644 --- a/go.mod +++ b/go.mod @@ -4,9 +4,9 @@ go 1.27.3 // Console and MC retain their historical module paths for best-effort upstream // compatibility. Pin the maintained PGSTY implementations used by SILO. -replace github.com/minio/console => github.com/pgsty/silo-console v0.0.0-20261025145732-989fa7150792 +replace github.com/minio/console => github.com/pgsty/silo-console v0.0.0-20261026143515-6a05e8d6ad67 -replace github.com/minio/mc => github.com/pgsty/mc v0.0.0-20261025104502-dcc78d506f94 +replace github.com/minio/mc => github.com/pgsty/mc v0.0.0-20261026120214-0ceb298cdcd0 // v22.7.0 does not compile on NetBSD because its unix implementation uses // CLOCK_MONOTONIC, which is unavailable there. Keep the last portable release @@ -68,7 +68,7 @@ require ( github.com/minio/kms-go/kes v0.3.1 github.com/minio/kms-go/kms v0.7.0 github.com/minio/madmin-go/v3 v3.0.110 - github.com/minio/minio-go/v7 v7.4.2-0.20261020194015-ab93db7ba435 + github.com/minio/minio-go/v7 v7.4.2-0.20261025165926-320af050c445 github.com/minio/mux v1.10.1 github.com/minio/selfupdate v0.6.0 github.com/minio/simdjson-go v0.4.5 @@ -81,7 +81,7 @@ require ( github.com/nats-io/stan.go v0.10.4 github.com/ncw/directio v1.0.5 github.com/nsqio/go-nsq v1.1.0 - github.com/pgsty/silo-pkg/v3 v3.15.1 + github.com/pgsty/silo-pkg/v3 v3.15.2 github.com/philhofer/fwd v1.2.1 github.com/pierrec/lz4/v4 v4.1.33 github.com/pkg/errors v0.9.1 diff --git a/go.sum b/go.sum index 2c4969e46..9aad1fd2d 100644 --- a/go.sum +++ b/go.sum @@ -442,8 +442,8 @@ github.com/minio/madmin-go/v3 v3.0.110 h1:FIYekj7YPc430ffpXFWiUtyut3qBt/unIAcDzJ github.com/minio/madmin-go/v3 v3.0.110/go.mod h1:WOe2kYmYl1OIlY2DSRHVQ8j1v4OItARQ6jGyQqcCud8= github.com/minio/md5-simd v1.1.2 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34= github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM= -github.com/minio/minio-go/v7 v7.4.2-0.20261020194015-ab93db7ba435 h1:BIn1BJi2zZ+z5BIMsr02USFB+Wmipc4SrIAA8nksNH0= -github.com/minio/minio-go/v7 v7.4.2-0.20261020194015-ab93db7ba435/go.mod h1:nQnQi7zTfXR4lREtYXZQWqo5aw0Rqv+ZqCGrTORygRk= +github.com/minio/minio-go/v7 v7.4.2-0.20261025165926-320af050c445 h1:TIdz2CWnaQQDML9I03RY0pdms4zfqpkGHPmnJxfvCrE= +github.com/minio/minio-go/v7 v7.4.2-0.20261025165926-320af050c445/go.mod h1:nQnQi7zTfXR4lREtYXZQWqo5aw0Rqv+ZqCGrTORygRk= github.com/minio/mux v1.10.1 h1:grrK8SwRKbkNFE6qG7WAvFGH09bB46d5teOOtKfQ14s= github.com/minio/mux v1.10.1/go.mod h1:INYT4sMSTJy0QWUEA/E2DZNxJ5sAxIwbnyZjkzNFRfE= github.com/minio/pkg/v3 v3.11.0 h1:WjzWGP7lyuupJgUSRim1RDtNujHtESyMQF7qSR4r5Dg= @@ -506,12 +506,12 @@ github.com/olekukonko/tablewriter v0.0.5/go.mod h1:hPp6KlRPjbx+hW8ykQs1w3UBbZlj6 github.com/orisano/pixelmatch v0.0.0-20220722002657-fb0b55479cde/go.mod h1:nZgzbfBr3hhjoZnS66nKrHmduYNpc34ny7RK4z5/HM0= github.com/pborman/getopt v0.0.0-20170112200414-7148bc3a4c30/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o= github.com/pelletier/go-toml v1.2.0/go.mod h1:5z9KED0ma1S8pY6P1sdut58dfprrGBbd/94hg7ilaic= -github.com/pgsty/mc v0.0.0-20261025104502-dcc78d506f94 h1:M5n0rCPOTyo3RiQBeSCQW/klx08efFpiVYmPq6eoM3w= -github.com/pgsty/mc v0.0.0-20261025104502-dcc78d506f94/go.mod h1:H3Uwlm1Qj3kPsbmojhNnv3aj5a+QdjdTKwv4w4gMSF8= -github.com/pgsty/silo-console v0.0.0-20261025145732-989fa7150792 h1:wPnltE28wxotYQIKeg5OykplBcrhaWeTnWRxbYXOTtQ= -github.com/pgsty/silo-console v0.0.0-20261025145732-989fa7150792/go.mod h1:fMxNcVxan3mu9mc7HCRiczSyQoFrbZEGumEmpatxtds= -github.com/pgsty/silo-pkg/v3 v3.15.1 h1:T5rqY3HR1WGybcjnx/ghBvliEgo7yNCA7JAC3f1LpTk= -github.com/pgsty/silo-pkg/v3 v3.15.1/go.mod h1:GmTF6fIEkTrzgM+ysWWfDZIlVRD2DYAJkKjA5bjiWpQ= +github.com/pgsty/mc v0.0.0-20261026120214-0ceb298cdcd0 h1:kxlOeGEFoBW/lgLQVBL9pcw3uq5KTMO6VmqdeDebTig= +github.com/pgsty/mc v0.0.0-20261026120214-0ceb298cdcd0/go.mod h1:dUu9hAVOMyAC+/9XrMdPxISLwJp537jXuv3bNW5+q5E= +github.com/pgsty/silo-console v0.0.0-20261026143515-6a05e8d6ad67 h1:8wWjVNEqYjcO1rm+QVNyQKVSNmJT8Q8HJfWR3rnEJlo= +github.com/pgsty/silo-console v0.0.0-20261026143515-6a05e8d6ad67/go.mod h1:0CxxVsLvy+u91xMHwrLGI1gwc96sBKpB8JTpGWIOm7I= +github.com/pgsty/silo-pkg/v3 v3.15.2 h1:PXgFAgkymHXhq8m5Rbs4QofHENOy5M7hLbC0KKowwvs= +github.com/pgsty/silo-pkg/v3 v3.15.2/go.mod h1:uk5B2JR5/NwmlGwwzPFUybtvn8z5ikPcxNLtPwBvcZ0= github.com/philhofer/fwd v1.2.1 h1:UtpS73q8bFgy/yH3gCq8LSNvj5gv/FowNOiWZa7uKwQ= github.com/philhofer/fwd v1.2.1/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM= github.com/pierrec/lz4/v4 v4.1.33 h1:GjG1TJ1V4IzKP8L96muuuDNpTwd7D+l2ccXrjAbe014= diff --git a/helm/silo/values.yaml b/helm/silo/values.yaml index c276dc3a3..cd5ddfdb4 100644 --- a/helm/silo/values.yaml +++ b/helm/silo/values.yaml @@ -25,7 +25,7 @@ imagePullSecrets: [] ## mcImage: repository: pgsty/mc - tag: RELEASE.2026-10-25T00-00-00Z + tag: RELEASE.2026-10-26T00-00-00Z pullPolicy: IfNotPresent ## Silo mode, i.e. standalone or distributed. diff --git a/internal/grid/connection.go b/internal/grid/connection.go index 513c77f23..bf8a1fe9d 100644 --- a/internal/grid/connection.go +++ b/internal/grid/connection.go @@ -344,8 +344,7 @@ func (c *Subroute) newMuxClient(ctx context.Context) (*muxClient, error) { // Request allows to do a single remote request. // 'req' will not be used after the call and caller can reuse. -// If no deadline is set on ctx, a 1-minute deadline will be added unless -// WithoutDefaultTimeout was requested. +// If no deadline is set on ctx, a 1-minute deadline will be added. func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error) { if !h.valid() { return nil, ErrUnknownHandler @@ -371,8 +370,7 @@ func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]by // Request allows to do a single remote request. // 'req' will not be used after the call and caller can reuse. -// If no deadline is set on ctx, a 1-minute deadline will be added unless -// WithoutDefaultTimeout was requested. +// If no deadline is set on ctx, a 1-minute deadline will be added. func (c *Subroute) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error) { if !h.valid() { return nil, ErrUnknownHandler diff --git a/internal/grid/handlers.go b/internal/grid/handlers.go index 12797c4be..1d20fa9b9 100644 --- a/internal/grid/handlers.go +++ b/internal/grid/handlers.go @@ -550,8 +550,7 @@ type Requester interface { // Call the remote with the request and return the response. // The response should be returned with PutResponse when no error. -// If no deadline is set, a 1-minute deadline is added unless -// WithoutDefaultTimeout was requested. +// If no deadline is set, a 1-minute deadline is added. func (h *SingleHandler[Req, Resp]) Call(ctx context.Context, c Requester, req Req) (resp Resp, err error) { if c == nil { if h.ignoreNilConn { diff --git a/internal/grid/muxclient.go b/internal/grid/muxclient.go index 19fe90809..769f0854e 100644 --- a/internal/grid/muxclient.go +++ b/internal/grid/muxclient.go @@ -59,16 +59,6 @@ type Response struct { Err error } -type withoutDefaultTimeoutKey struct{} - -// WithoutDefaultTimeout disables the implicit single-request timeout. -// Explicit context cancellation and connection loss still end the request. -// Use this for admitted mutations that must return their actual result before -// the caller releases its lock. -func WithoutDefaultTimeout(ctx context.Context) context.Context { - return context.WithValue(ctx, withoutDefaultTimeoutKey{}, true) -} - func newMuxClient(ctx context.Context, muxID uint64, parent *Connection) *muxClient { ctx, cancelFn := context.WithCancelCause(ctx) return &muxClient{ @@ -111,10 +101,8 @@ func (m *muxClient) roundtrip(h HandlerID, req []byte) ([]byte, error) { m.respMu.Unlock() ctx := m.ctx - // Admitted storage mutations must not time out while the handler can - // still commit. A zero wire deadline also keeps their late response. - noDefaultTimeout, _ := ctx.Value(withoutDefaultTimeoutKey{}).(bool) - if msg.DeadlineMS == 0 && !noDefaultTimeout { + // Add deadline if none. + if msg.DeadlineMS == 0 { msg.DeadlineMS = uint32(defaultSingleRequestTimeout / time.Millisecond) var cancel context.CancelFunc ctx, cancel = context.WithTimeout(ctx, defaultSingleRequestTimeout) diff --git a/internal/grid/muxclient_timeout_test.go b/internal/grid/muxclient_timeout_test.go index d07a10b5c..994960723 100644 --- a/internal/grid/muxclient_timeout_test.go +++ b/internal/grid/muxclient_timeout_test.go @@ -25,67 +25,43 @@ import ( "time" ) -func TestSingleMutationWaitsBeyondDefaultTimeout(t *testing.T) { - for _, admitted := range []bool{false, true} { - t.Run(map[bool]string{false: "ordinary", true: "admitted"}[admitted], func(t *testing.T) { - synctest.Test(t, func(t *testing.T) { - ctx := t.Context() - if admitted { - ctx = WithoutDefaultTimeout(ctx) - } - conn := &Connection{outQueue: make(chan []byte, 1)} - client := newMuxClient(ctx, 1, conn) - defer client.cancelFn(context.Canceled) - done := make(chan Response, 1) - go func() { - msg, err := client.roundtrip(handlerTest, nil) - done <- Response{Msg: msg, Err: err} - }() - wire := <-conn.outQueue - var request message - if _, err := request.UnmarshalMsg(wire); err != nil { - t.Fatal(err) - } - PutByteBuffer(wire) - if admitted && request.DeadlineMS != 0 { - t.Errorf("admitted mutation tells the server to discard a late result after %dms", request.DeadlineMS) - } - time.Sleep(defaultSingleRequestTimeout + time.Second) - synctest.Wait() - if !admitted { - select { - case got := <-done: - if !errors.Is(got.Err, context.DeadlineExceeded) { - t.Fatalf("ordinary timeout: %v", got.Err) - } - default: - t.Fatal("ordinary request lost its default timeout") - } - return - } - select { - case got := <-done: - t.Fatalf("mutation returned before its commit completed: %v", got.Err) - default: - } - client.response(0, Response{Msg: []byte("committed")}) - got := <-done - if got.Err != nil || string(got.Msg) != "committed" { - t.Fatalf("lost late commit result: %q, %v", got.Msg, got.Err) - } - }) - }) - } +func TestSingleRequestDefaultTimeout(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + conn := &Connection{outQueue: make(chan []byte, 1)} + client := newMuxClient(t.Context(), 1, conn) + defer client.cancelFn(context.Canceled) + done := make(chan error, 1) + go func() { _, err := client.roundtrip(handlerTest, nil); done <- err }() + wire := <-conn.outQueue + var request message + if _, err := request.UnmarshalMsg(wire); err != nil { + t.Fatal(err) + } + PutByteBuffer(wire) + if request.DeadlineMS != uint32(defaultSingleRequestTimeout/time.Millisecond) { + t.Fatalf("wire deadline: %d", request.DeadlineMS) + } + time.Sleep(defaultSingleRequestTimeout + time.Second) + synctest.Wait() + select { + case err := <-done: + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatal(err) + } + default: + t.Fatal("request outlived the default timeout") + } + }) } -func TestWithoutDefaultTimeoutStillEndsOnDisconnectOrCancellation(t *testing.T) { +func TestSingleRequestCancellationAndDisconnect(t *testing.T) { for _, disconnect := range []bool{false, true} { t.Run(map[bool]string{false: "explicit-cancel", true: "disconnect"}[disconnect], func(t *testing.T) { synctest.Test(t, func(t *testing.T) { ctx, cancel := context.WithCancel(t.Context()) defer cancel() conn := &Connection{outQueue: make(chan []byte, 1)} - client := newMuxClient(WithoutDefaultTimeout(ctx), 1, conn) + client := newMuxClient(ctx, 1, conn) defer client.cancelFn(context.Canceled) done := make(chan error, 1) go func() { diff --git a/internal/ioutil/deadline_test.go b/internal/ioutil/deadline_test.go new file mode 100644 index 000000000..eb0cbee5c --- /dev/null +++ b/internal/ioutil/deadline_test.go @@ -0,0 +1,62 @@ +// Copyright (c) 2026 Feng Ruohang +// SPDX-License-Identifier: AGPL-3.0-or-later + +package ioutil + +import ( + "context" + "errors" + "strings" + "testing" + "testing/synctest" + "time" +) + +func TestWithDeadlineReturnsBeforeWorker(t *testing.T) { + for _, cancelEarly := range []bool{false, true} { + synctest.Test(t, func(t *testing.T) { + ctx, cancel := context.WithCancel(t.Context()) + defer cancel() + entered, release := make(chan struct{}), make(chan struct{}) + defer close(release) + done := make(chan error, 1) + go func() { + value, err := WithDeadline(ctx, time.Second, func(context.Context) (int, error) { + close(entered) + <-release + return 42, nil + }) + if value != 0 { + t.Errorf("timeout returned a late worker value: %d", value) + } + done <- err + }() + <-entered + want := context.DeadlineExceeded + if cancelEarly { + cancel() + want = context.Canceled + } else { + time.Sleep(2 * time.Second) + } + synctest.Wait() + select { + case err := <-done: + if !errors.Is(err, want) { + t.Fatalf("expected %v, got %v", want, err) + } + default: + t.Fatal("caller still waits for the blocked worker") + } + }) + } +} + +func TestWithDeadlineContainsPanic(t *testing.T) { + value, err := WithDeadline(t.Context(), time.Second, func(context.Context) (int, error) { + panic("deadline-test") + }) + if value != 0 || err == nil || !strings.Contains(err.Error(), "deadline-test") { + t.Fatalf("panic was not converted to an error: %d, %v", value, err) + } +} diff --git a/internal/ioutil/ioutil.go b/internal/ioutil/ioutil.go index 87c628b6e..8aabf699e 100644 --- a/internal/ioutil/ioutil.go +++ b/internal/ioutil/ioutil.go @@ -127,17 +127,6 @@ var deadlineWorkPanics atomic.Int64 // If the deadline/context passes before the function finishes executing, // the zero value and the context error is returned. func WithDeadline[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error)) (result V, err error) { - return withDeadline(ctx, timeout, work, false) -} - -// WithDeadlineAndWait cancels work at the deadline, but waits for it to finish -// and returns its actual result. Use it for mutations whose caller must not -// release a lock or start cleanup while the work can still change storage. -func WithDeadlineAndWait[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error)) (result V, err error) { - return withDeadline(ctx, timeout, work, true) -} - -func withDeadline[V any](ctx context.Context, timeout time.Duration, work func(ctx context.Context) (result V, err error), wait bool) (result V, err error) { ctx, cancel := context.WithTimeout(ctx, timeout) defer cancel() @@ -174,10 +163,6 @@ func withDeadline[V any](ctx context.Context, timeout time.Duration, work func(c case v := <-c: return v.val, v.err case <-ctx.Done(): - if wait { - v := <-c - return v.val, v.err - } var zero V return zero, ctx.Err() } diff --git a/internal/ioutil/ioutil_test.go b/internal/ioutil/ioutil_test.go index 3e9984039..b71c2f2e0 100644 --- a/internal/ioutil/ioutil_test.go +++ b/internal/ioutil/ioutil_test.go @@ -25,62 +25,9 @@ import ( "os" "strings" "testing" - "testing/synctest" "time" ) -func TestWithDeadlineAndWait(t *testing.T) { - for _, deadline := range []bool{false, true} { - name := "cancel" - if deadline { - name = "deadline" - } - t.Run(name, func(t *testing.T) { - synctest.Test(t, func(t *testing.T) { - ctx, cancel := context.WithCancel(t.Context()) - defer cancel() - entered, release := make(chan struct{}), make(chan struct{}) - done := make(chan ioret[int], 1) - go func() { - v, err := WithDeadlineAndWait(ctx, time.Second, func(context.Context) (int, error) { - close(entered) - <-release - return 42, nil - }) - done <- ioret[int]{v, err} - }() - <-entered - if deadline { - time.Sleep(2 * time.Second) - } else { - cancel() - } - synctest.Wait() - select { - case r := <-done: - t.Fatalf("returned before mutation completed: %+v", r) - default: - } - close(release) - synctest.Wait() - r := <-done - if r.err != nil || r.val != 42 { - t.Fatalf("lost actual commit result: %+v", r) - } - }) - }) - } -} - -func TestWithDeadlineAndWaitPanic(t *testing.T) { - _, err := WithDeadlineAndWait(t.Context(), time.Second, func(context.Context) (int, error) { - panic("mutation failure") - }) - if err == nil || !strings.Contains(err.Error(), "mutation failure") { - t.Fatalf("worker panic was not returned: %v", err) - } -} - type sleepWriter struct { timeout time.Duration }