mirror of
https://github.com/pgsty/minio.git
synced 2026-10-31 21:02:06 +03:00
Preserve maintenance timeout behavior and prepare SILO 20261026
Signed-off-by: Feng Ruohang <rh@vonng.com>
This commit is contained in:
@@ -118,10 +118,13 @@ jobs:
|
|||||||
MINIO_API_REQUESTS_MAX: "10000"
|
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
|
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:
|
env:
|
||||||
MINIO_API_REQUESTS_MAX: "10000"
|
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:
|
crosscompile:
|
||||||
name: Cross Compile
|
name: Cross Compile
|
||||||
|
|||||||
+23
-23
@@ -1,9 +1,9 @@
|
|||||||
# Changelog
|
# 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
|
Package version: `20261026000000.0.0`. Publication status and artifacts are
|
||||||
Release; public Server artifacts and images remain at 20261018 until promotion.
|
listed in [GitHub Releases](https://github.com/pgsty/silo/releases).
|
||||||
|
|
||||||
- Fix LDAP login for long normalized DNs without hiding unreadable IAM
|
- Fix LDAP login for long normalized DNs without hiding unreadable IAM
|
||||||
revocation records. Only length errors for valid keys with an ASCII path
|
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
|
Thanks to [@910patin](https://github.com/910patin) for the original fix and
|
||||||
reproduction in [#250](https://github.com/pgsty/silo/pull/250).
|
reproduction in [#250](https://github.com/pgsty/silo/pull/250).
|
||||||
- Build the server and standalone debugging tools with Go 1.27.3 and refresh
|
- 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
|
Go dependencies, including silo-pkg v3.15.2, mcli 20261026
|
||||||
source and upstream minio-go `ab93db7ba435` for multipart, read/seek and
|
and upstream minio-go `320af050c445` for multipart, read/seek,
|
||||||
batch-delete fixes. Retain output-format and NetBSD compatibility selections.
|
batch-delete and bucket-replication removal fixes. Retain output-format and NetBSD compatibility selections.
|
||||||
- Embed the published Console v2.5.0 source `989fa7150792`, including its
|
- Embed Console v2.5.1 source `6a05e8d6ad67`, including its GUI update,
|
||||||
GUI update, diagnostic WebSocket closing handshake and complete health-report
|
diagnostic WebSocket closing handshake, complete health-report decoding,
|
||||||
decoding. Bundle published mcli 20261025 in container builds and select it
|
group-membership preservation, OIDC group permissions, TLS-proxy session
|
||||||
for the verified client installer and Helm client jobs.
|
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
|
- Read current bucket inventory in account-info responses so Console refreshes
|
||||||
immediately reflect successful creation and deletion. This management endpoint
|
immediately reflect successful creation and deletion. This management endpoint
|
||||||
now performs a live list instead of returning a cached inventory.
|
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.
|
and multi-pool deletion contributions.
|
||||||
- Preserve Kafka's default broker protocol 2.1 and terminal-width behavior;
|
- 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.
|
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,
|
- Forward storage RPC delete options, including the previous data directory
|
||||||
deciding rollback or cleaning staging data. Rollback verifies ownership and
|
needed by overwrite rollback. Preserve existing drive/RPC timeouts, request
|
||||||
restores only its overwritten version, preserving later writes and metadata
|
cancellation, object locking, read-quorum and healing behavior.
|
||||||
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.
|
|
||||||
- Update bundled curl to 8.23.0 and require patched static libraries. Apply
|
- 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);
|
[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.
|
the root blocklist also rejects legitimate chains under the affected roots.
|
||||||
- Ship Windows platform archives as ZIP; Linux and macOS retain tar.gz.
|
- Ship Windows platform archives as ZIP; Linux and macOS retain tar.gz.
|
||||||
|
|
||||||
Storage limits remain: this does not provide distributed fencing after lease
|
Storage timeouts retain their existing meaning: a returned timeout does not
|
||||||
loss or network partition, eliminate stale reads after partial metadata writes,
|
prove that the underlying disk operation stopped or that a write did not
|
||||||
or repair previously inconsistent objects. Stricter read refusal and additional
|
commit. Late commits or delayed rollback can affect newer writes; stale reads
|
||||||
cross-pool recovery remain deferred; their diagnostic tests are retained under
|
remain possible after partial failures, lease loss or network partitions. This
|
||||||
`consistency_strict`. Waiting for admitted I/O can extend same-object waits on
|
release does not provide distributed fencing or repair previously inconsistent
|
||||||
a stuck drive, and interrupted cleanup can retain rollback backups. Upgrade all
|
objects.
|
||||||
writers before relying on the new rollback ownership checks.
|
|
||||||
|
|
||||||
These source changes are not included in the published 20261018 artifacts.
|
These source changes are not included in the published 20261018 artifacts.
|
||||||
|
|
||||||
|
|||||||
+13
-9
@@ -53,15 +53,19 @@ Before opening a pull request:
|
|||||||
- Explain any preserved `MINIO_*`, `minio_*`, `x-minio-*`, `/minio/*`,
|
- Explain any preserved `MINIO_*`, `minio_*`, `x-minio-*`, `/minio/*`,
|
||||||
`.minio.sys`, ARN, module/import-path, or serialized compatibility name.
|
`.minio.sys`, ARN, module/import-path, or serialized compatibility name.
|
||||||
|
|
||||||
The maintenance branch retains the existing read-quorum and healing policy.
|
The maintenance branch retains existing storage timeouts, cancellation,
|
||||||
Deferred strict-read and cross-pool recovery tests are preserved behind the
|
object locking, read-quorum and healing behavior. The normal suite checks local
|
||||||
`consistency_strict` build tag. Run them explicitly with
|
and RPC timeout/cancellation, remote rollback options, and readability after a
|
||||||
`go test -tags kqueue,dev,consistency_strict ./cmd`; known failures must be
|
failed partial overwrite. A timeout does not prove that a disk worker stopped
|
||||||
reported separately, not described as fixed or included in a passing result.
|
or that a write did not commit.
|
||||||
They cover late commits after lease loss, stale metadata reads, and stricter
|
|
||||||
read refusal. The normal suite covers the retained write/undo/cleanup fixes
|
Stronger mutation lifetime, rollback ownership, generation ordering, strict-read
|
||||||
and continued readability after a failed partial overwrite. This scope does
|
and cross-pool recovery diagnostics are preserved behind `consistency_strict`.
|
||||||
not provide distributed fencing or repair previously inconsistent objects.
|
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
|
### Commit changes
|
||||||
|
|
||||||
|
|||||||
@@ -17,7 +17,7 @@ ENV GOPATH=/go
|
|||||||
ENV CGO_ENABLED=0
|
ENV CGO_ENABLED=0
|
||||||
|
|
||||||
ARG MC_REPO=pgsty/mc
|
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_AMD64_SHA256=54bd131039dad6677f083a75207b28f82818c067c8e65fae1973f51510d268e1
|
||||||
ARG MC_ARM64_SHA256=9bb294f1912f8c6b9ac08c9a0b8d2964b86306a7c440b987ee5f0570d18de3c5
|
ARG MC_ARM64_SHA256=9bb294f1912f8c6b9ac08c9a0b8d2964b86306a7c440b987ee5f0570d18de3c5
|
||||||
|
|
||||||
|
|||||||
@@ -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.
|
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,
|
The 20261026 maintenance release is in preparation with Go 1.27.3,
|
||||||
silo-pkg v3.15.1, released mcli 20261025, Console v2.5.0 (`989fa7150792`),
|
silo-pkg v3.15.2, mcli 20261026, Console v2.5.1 (`6a05e8d6ad67`),
|
||||||
and upstream minio-go `ab93db7ba435`. It includes the Server IAM cache repair
|
and upstream minio-go `320af050c445`. It includes the Server IAM cache repair
|
||||||
needed by Console's **Reset to default** action and bounded write/rollback/cleanup
|
needed by Console's **Reset to default** action and remote rollback-option propagation, preserving existing storage
|
||||||
repairs. Read the [changes and remaining storage limits](CHANGELOG.md#release2026-10-25t00-00-00z).
|
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.
|
The published artifacts above retain their original component versions.
|
||||||
|
|
||||||
## Overview
|
## Overview
|
||||||
|
|||||||
+5
-5
@@ -43,11 +43,11 @@
|
|||||||
|
|
||||||
升级前请阅读[发行说明](https://github.com/pgsty/silo/releases/tag/RELEASE.2026-10-18T00-00-00Z),特别是 strict 分片列表的移除、服务端协调升级要求与实测性能成本。代码、问题报告和文档贡献者见 [CONTRIBUTORS.md](CONTRIBUTORS.md)。
|
升级前请阅读[发行说明](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、
|
20261026 维护版本正在准备中:使用 Go 1.27.3、silo-pkg v3.15.2、
|
||||||
已发布的 mcli 20261025、Console v2.5.0(`989fa7150792`)与上游 minio-go
|
mcli 20261026、Console v2.5.1(`6a05e8d6ad67`)与上游 minio-go
|
||||||
`ab93db7ba435`。其中包含 Console **Reset to default(恢复默认)** 所需的
|
`320af050c445`。其中包含 Console **Reset to default(恢复默认)** 所需的
|
||||||
Server IAM 缓存修复,以及限定范围的写入、回滚与清理修复。
|
Server IAM 缓存修复,以及远端回滚参数透传修复;保持既有存储超时与取消行为。
|
||||||
请阅读[变更与仍存的存储限制](CHANGELOG.md#release2026-10-25t00-00-00z);
|
请阅读[变更与仍存的存储限制](CHANGELOG.md#release2026-10-26t00-00-00z);
|
||||||
上述已发布制品仍保留原有组件版本。
|
上述已发布制品仍保留原有组件版本。
|
||||||
|
|
||||||
## 概述
|
## 概述
|
||||||
|
|||||||
@@ -40,7 +40,7 @@ if [ -n "${MCLI_BIN:-}" ]; then
|
|||||||
exit 0
|
exit 0
|
||||||
fi
|
fi
|
||||||
|
|
||||||
release=${MCLI_RELEASE:-RELEASE.2026-10-25T00-00-00Z}
|
release=${MCLI_RELEASE:-RELEASE.2026-10-26T00-00-00Z}
|
||||||
version_hyphen=${release#RELEASE.}
|
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/')
|
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
|
if [ "${package_version}" = "${version_hyphen}" ]; then
|
||||||
|
|||||||
@@ -1,3 +1,5 @@
|
|||||||
|
//go:build consistency_strict
|
||||||
|
|
||||||
// Copyright (c) 2026 Feng Ruohang
|
// Copyright (c) 2026 Feng Ruohang
|
||||||
// SPDX-License-Identifier: AGPL-3.0-or-later
|
// SPDX-License-Identifier: AGPL-3.0-or-later
|
||||||
|
|
||||||
|
|||||||
+10
-20
@@ -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
|
// 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) {
|
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))
|
g := errgroup.WithNErrs(len(disks))
|
||||||
|
|
||||||
fvID := mustGetUUID()
|
fvID := mustGetUUID()
|
||||||
for index := range disks {
|
for index := range disks {
|
||||||
metadata[index].Metadata = maps.Clone(metadata[index].Metadata)
|
|
||||||
metadata[index].SetTierFreeVersionID(fvID)
|
metadata[index].SetTierFreeVersionID(fvID)
|
||||||
metadata[index].Metadata[rollbackIDKey] = fvID
|
|
||||||
}
|
}
|
||||||
|
|
||||||
diskVersions := make([][]byte, len(disks))
|
diskVersions := make([][]byte, len(disks))
|
||||||
@@ -1564,18 +1556,6 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
|
|||||||
compIndex = opts.IndexCB()
|
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
|
modTime := opts.MTime
|
||||||
if opts.MTime.IsZero() {
|
if opts.MTime.IsZero() {
|
||||||
modTime = UTCNow()
|
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 &&
|
if !opts.ReplicationRequest && !opts.ReplicaLockReconcile && !opts.DataMovement &&
|
||||||
(fi.VersionID == "" || fi.VersionID == nullVersionID) {
|
(fi.VersionID == "" || fi.VersionID == nullVersionID) {
|
||||||
getObjectInfo := er.getObjectInfo
|
getObjectInfo := er.getObjectInfo
|
||||||
|
|||||||
@@ -1,3 +1,5 @@
|
|||||||
|
//go:build consistency_strict
|
||||||
|
|
||||||
// Copyright (c) 2026 Feng Ruohang
|
// Copyright (c) 2026 Feng Ruohang
|
||||||
//
|
//
|
||||||
// This file is part of Silo Object Storage.
|
// This file is part of Silo Object Storage.
|
||||||
|
|||||||
@@ -26,8 +26,7 @@ func (d compatibilityFailedOverwrite) RenameData(ctx context.Context, sv, sp str
|
|||||||
return RenameDataResp{}, errDiskNotFound
|
return RenameDataResp{}, errDiskNotFound
|
||||||
}
|
}
|
||||||
|
|
||||||
// Maintenance compatibility requirement: after a completely joined failed
|
// Maintenance compatibility requirement: after a failed overwrite with one residue, keep the previously acknowledged quorum readable.
|
||||||
// overwrite with one residue, keep the previously acknowledged quorum readable.
|
|
||||||
// This tests a known history; it does not claim the resulting snapshot proves it.
|
// This tests a known history; it does not claim the resulting snapshot proves it.
|
||||||
func TestMaintenancePartialFailedOverwriteAvailability(t *testing.T) {
|
func TestMaintenancePartialFailedOverwriteAvailability(t *testing.T) {
|
||||||
// Lock RPC tests configure a single-drive backend (EC:0). Pin this
|
// Lock RPC tests configure a single-drive backend (EC:0). Pin this
|
||||||
|
|||||||
@@ -399,7 +399,10 @@ func getCompleteMultipartMD5(parts []CompletePart) string {
|
|||||||
return s3MD5
|
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 {
|
func cleanMetadata(metadata map[string]string) map[string]string {
|
||||||
// Remove STANDARD StorageClass
|
// Remove STANDARD StorageClass
|
||||||
metadata = removeStandardStorageClass(metadata)
|
metadata = removeStandardStorageClass(metadata)
|
||||||
|
|||||||
@@ -1,3 +1,5 @@
|
|||||||
|
//go:build consistency_strict
|
||||||
|
|
||||||
// Copyright (c) 2026 Feng Ruohang
|
// Copyright (c) 2026 Feng Ruohang
|
||||||
//
|
//
|
||||||
// This file is part of Silo Object Storage.
|
// This file is part of Silo Object Storage.
|
||||||
@@ -472,7 +474,7 @@ func TestMutationDeleteJoinsNestedWorker(t *testing.T) {
|
|||||||
defer globalDriveConfig.Update(drive.Config{MaxTimeout: oldTimeout})
|
defer globalDriveConfig.Update(drive.Config{MaxTimeout: oldTimeout})
|
||||||
done := make(chan error, 1)
|
done := make(chan error, 1)
|
||||||
go func() {
|
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})
|
return struct{}{}, disk.storage.DeleteVersion(ctx, bucket, object, fi, false, DeleteOptions{UndoWrite: true})
|
||||||
})
|
})
|
||||||
done <- e
|
done <- e
|
||||||
|
|||||||
@@ -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
|
||||||
|
}
|
||||||
@@ -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 <http://www.gnu.org/licenses/>.
|
|
||||||
|
|
||||||
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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,3 +1,5 @@
|
|||||||
|
//go:build consistency_strict
|
||||||
|
|
||||||
// Copyright (c) 2026 Feng Ruohang
|
// Copyright (c) 2026 Feng Ruohang
|
||||||
//
|
//
|
||||||
// This file is part of Silo Object Storage.
|
// This file is part of Silo Object Storage.
|
||||||
|
|||||||
@@ -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
|
|
||||||
}
|
|
||||||
@@ -1,3 +1,5 @@
|
|||||||
|
//go:build consistency_strict
|
||||||
|
|
||||||
// Copyright (c) 2026 Feng Ruohang
|
// Copyright (c) 2026 Feng Ruohang
|
||||||
//
|
//
|
||||||
// This file is part of Silo Object Storage.
|
// This file is part of Silo Object Storage.
|
||||||
|
|||||||
@@ -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 {
|
func (client *storageRESTClient) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) error {
|
||||||
if err := ctx.Err(); err != nil {
|
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
|
||||||
return err
|
defer cancel()
|
||||||
}
|
|
||||||
// As with RenameData, admitted metadata writes must finish before unlock.
|
|
||||||
ctx = grid.WithoutDefaultTimeout(context.WithoutCancel(ctx))
|
|
||||||
|
|
||||||
_, err := storageWriteMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{
|
_, err := storageWriteMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{
|
||||||
DiskID: *client.diskID.Load(),
|
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 {
|
func (client *storageRESTClient) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo, opts UpdateMetadataOpts) error {
|
||||||
if err := ctx.Err(); err != nil {
|
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
|
||||||
return err
|
defer cancel()
|
||||||
}
|
|
||||||
ctx = grid.WithoutDefaultTimeout(context.WithoutCancel(ctx))
|
|
||||||
|
|
||||||
_, err := storageUpdateMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{
|
_, err := storageUpdateMetadataRPC.Call(ctx, client.gridConn, &MetadataHandlerParams{
|
||||||
DiskID: *client.diskID.Load(),
|
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) {
|
func (client *storageRESTClient) DeleteVersion(ctx context.Context, volume, path string, fi FileInfo, forceDelMarker bool, opts DeleteOptions) (err error) {
|
||||||
if opts.UndoWrite {
|
ctx, cancel := context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
|
||||||
// Rollback is part of the commit and must finish before unlocking.
|
defer cancel()
|
||||||
ctx = grid.WithoutDefaultTimeout(context.WithoutCancel(ctx))
|
|
||||||
} else {
|
|
||||||
var cancel context.CancelFunc
|
|
||||||
ctx, cancel = context.WithTimeout(ctx, globalDriveConfig.GetMaxTimeout())
|
|
||||||
defer cancel()
|
|
||||||
}
|
|
||||||
|
|
||||||
_, err = storageDeleteVersionRPC.Call(ctx, client.gridConn, &DeleteVersionHandlerParams{
|
_, err = storageDeleteVersionRPC.Call(ctx, client.gridConn, &DeleteVersionHandlerParams{
|
||||||
DiskID: *client.diskID.Load(),
|
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,
|
func (client *storageRESTClient) RenameData(ctx context.Context, srcVolume, srcPath string, fi FileInfo,
|
||||||
dstVolume, dstPath string, opts RenameOptions,
|
dstVolume, dstPath string, opts RenameOptions,
|
||||||
) (res RenameDataResp, err error) {
|
) (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{
|
params := RenameDataHandlerParams{
|
||||||
DiskID: *client.diskID.Load(),
|
DiskID: *client.diskID.Load(),
|
||||||
SrcVolume: srcVolume,
|
SrcVolume: srcVolume,
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -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
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -495,7 +495,7 @@ func (p *xlStorageDiskIDCheck) RenameData(ctx context.Context, srcVolume, srcPat
|
|||||||
if len(fi.Data) > 0 {
|
if len(fi.Data) > 0 {
|
||||||
fi.Data = append(grid.GetByteBufferCap(len(fi.Data))[:0], fi.Data...)
|
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 {
|
if len(fi.Data) > 0 {
|
||||||
defer grid.PutByteBuffer(fi.Data)
|
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())
|
w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
|
||||||
return w.Run(func() error { return p.storage.DeleteVersion(ctx, volume, path, fi, forceDelMarker, opts) })
|
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)
|
defer done(0, &err)
|
||||||
|
|
||||||
_, err = xioutil.WithDeadlineAndWait(ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (struct{}, error) {
|
w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
|
||||||
return struct{}{}, p.storage.UpdateMetadata(ctx, volume, path, fi, opts)
|
return w.Run(func() error { return p.storage.UpdateMetadata(ctx, volume, path, fi, opts) })
|
||||||
})
|
|
||||||
return err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *xlStorageDiskIDCheck) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) (err error) {
|
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)
|
defer done(0, &err)
|
||||||
|
|
||||||
_, err = xioutil.WithDeadlineAndWait(ctx, globalDriveConfig.GetMaxTimeout(), func(ctx context.Context) (struct{}, error) {
|
w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
|
||||||
return struct{}{}, p.storage.WriteMetadata(ctx, origvolume, volume, path, fi)
|
return w.Run(func() error { return p.storage.WriteMetadata(ctx, origvolume, volume, path, fi) })
|
||||||
})
|
|
||||||
return err
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (p *xlStorageDiskIDCheck) ReadVersion(ctx context.Context, origvolume, volume, path, versionID string, opts ReadOptions) (fi FileInfo, err error) {
|
func (p *xlStorageDiskIDCheck) ReadVersion(ctx context.Context, origvolume, volume, path, versionID string, opts ReadOptions) (fi FileInfo, err error) {
|
||||||
|
|||||||
@@ -1570,11 +1570,7 @@ func (x *xlMetaV2) UpdateObjectVersion(fi FileInfo) error {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
delete(ver.ObjectV2.MetaSys, rollbackIDKey)
|
|
||||||
for k, v := range fi.Metadata {
|
for k, v := range fi.Metadata {
|
||||||
if k == rollbackIDKey {
|
|
||||||
continue
|
|
||||||
}
|
|
||||||
if len(k) > len(ReservedMetadataPrefixLower) && strings.EqualFold(k[:len(ReservedMetadataPrefixLower)], ReservedMetadataPrefixLower) {
|
if len(k) > len(ReservedMetadataPrefixLower) && strings.EqualFold(k[:len(ReservedMetadataPrefixLower)], ReservedMetadataPrefixLower) {
|
||||||
ver.ObjectV2.MetaSys[k] = []byte(v)
|
ver.ObjectV2.MetaSys[k] = []byte(v)
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
+11
-45
@@ -25,7 +25,6 @@ import (
|
|||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"maps"
|
|
||||||
"os"
|
"os"
|
||||||
pathutil "path"
|
pathutil "path"
|
||||||
"path/filepath"
|
"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 {
|
func (s *xlStorage) deleteVersions(ctx context.Context, volume, path string, fis ...FileInfo) error {
|
||||||
defer s.lockMutations(pathJoin(volume, path))()
|
|
||||||
volumeDir, err := s.getVolDir(volume)
|
volumeDir, err := s.getVolDir(volume)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
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 {
|
func (s *xlStorage) moveToTrash(filePath string, recursive, immediatePurge bool) (err error) {
|
||||||
// The caller must retain its mutation guard until this rename completes.
|
w := xioutil.NewDeadlineWorker(globalDriveConfig.GetMaxTimeout())
|
||||||
// Only purging the already detached unique trash directory is asynchronous.
|
return w.Run(func() (err error) {
|
||||||
return s.moveToTrashNoDeadline(filePath, recursive, immediatePurge)
|
return s.moveToTrashNoDeadline(filePath, recursive, immediatePurge)
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// DeleteVersion - deletes FileInfo metadata for path at `xl.meta`. forceDelMarker
|
// 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)
|
volumeDir, err := s.getVolDir(volume)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -1348,7 +1345,7 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F
|
|||||||
metaDataPoolPut(buf) // Never used, return it
|
metaDataPoolPut(buf) // Never used, return it
|
||||||
if fi.Deleted && forceDelMarker {
|
if fi.Deleted && forceDelMarker {
|
||||||
// Create a new xl.meta with a delete marker in it
|
// 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()
|
s.RLock()
|
||||||
@@ -1384,10 +1381,6 @@ func (s *xlStorage) DeleteVersion(ctx context.Context, volume, path string, fi F
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
if opts.UndoWrite {
|
|
||||||
return s.undoWrite(ctx, volume, path, fi, opts, &xlMeta)
|
|
||||||
}
|
|
||||||
|
|
||||||
dataDir, err := xlMeta.DeleteVersion(fi)
|
dataDir, err := xlMeta.DeleteVersion(fi)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
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)
|
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)
|
return s.deleteFile(volumeDir, pathJoin(volumeDir, path, xlStorageFormatFile), true, false)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Updates only metadata for a given version.
|
// Updates only metadata for a given version.
|
||||||
func (s *xlStorage) UpdateMetadata(ctx context.Context, volume, path string, fi FileInfo, opts UpdateMetadataOpts) (err error) {
|
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 {
|
if len(fi.Metadata) == 0 {
|
||||||
return errInvalidArgument
|
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`
|
// WriteMetadata - writes FileInfo metadata for path at `xl.meta`
|
||||||
func (s *xlStorage) WriteMetadata(ctx context.Context, origvolume, volume, path string, fi FileInfo) (err error) {
|
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 fi.Fresh {
|
||||||
if origvolume != "" {
|
if origvolume != "" {
|
||||||
origvolumeDir, err := s.getVolDir(origvolume)
|
origvolumeDir, err := s.getVolDir(origvolume)
|
||||||
@@ -2554,10 +2540,7 @@ func (s *xlStorage) DeleteBulk(ctx context.Context, volume string, paths ...stri
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
unlock := s.lockMutations(pathJoin(volume, fp), pathJoin(volume, pathutil.Dir(fp)))
|
if err = s.moveToTrash(filePath, false, false); err != nil {
|
||||||
err = s.moveToTrash(filePath, false, false)
|
|
||||||
unlock()
|
|
||||||
if err != nil {
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -2567,7 +2550,6 @@ func (s *xlStorage) DeleteBulk(ctx context.Context, volume string, paths ...stri
|
|||||||
|
|
||||||
// Delete - delete a file at path.
|
// Delete - delete a file at path.
|
||||||
func (s *xlStorage) Delete(ctx context.Context, volume string, path string, deleteOpts DeleteOptions) (err error) {
|
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)
|
volumeDir, err := s.getVolDir(volume)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
@@ -2597,7 +2579,6 @@ func skipAccessChecks(volume string) (ok bool) {
|
|||||||
|
|
||||||
// RenameData - rename source path to destination path atomically, metadata and data directory.
|
// 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) {
|
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() {
|
defer func() {
|
||||||
ignoredErrs := []error{
|
ignoredErrs := []error{
|
||||||
errFileNotFound,
|
errFileNotFound,
|
||||||
@@ -2841,12 +2822,6 @@ func (s *xlStorage) RenameData(ctx context.Context, srcVolume, srcPath string, f
|
|||||||
xlMeta.data.remove(reqVID, dataDir)
|
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 {
|
if err = xlMeta.AddVersion(fi); err != nil {
|
||||||
@@ -3299,7 +3274,6 @@ func (s *xlStorage) CleanAbandonedData(ctx context.Context, volume string, path
|
|||||||
if volume == "" || path == "" {
|
if volume == "" || path == "" {
|
||||||
return nil // Ignore
|
return nil // Ignore
|
||||||
}
|
}
|
||||||
defer s.lockMutations(pathJoin(volume, path))()
|
|
||||||
|
|
||||||
volumeDir, err := s.getVolDir(volume)
|
volumeDir, err := s.getVolDir(volume)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -3351,14 +3325,6 @@ func (s *xlStorage) CleanAbandonedData(ctx context.Context, volume string, path
|
|||||||
// Do not abort on context errors.
|
// Do not abort on context errors.
|
||||||
for dir := range foundDirs {
|
for dir := range foundDirs {
|
||||||
toRemove := pathJoin(volumeDir, path, dir+SlashSeparator)
|
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)
|
err = s.deleteFile(volumeDir, toRemove, true, true)
|
||||||
diskHealthCheckOK(ctx, err)
|
diskHealthCheckOK(ctx, err)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -4,9 +4,9 @@ go 1.27.3
|
|||||||
|
|
||||||
// Console and MC retain their historical module paths for best-effort upstream
|
// Console and MC retain their historical module paths for best-effort upstream
|
||||||
// compatibility. Pin the maintained PGSTY implementations used by SILO.
|
// 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
|
// v22.7.0 does not compile on NetBSD because its unix implementation uses
|
||||||
// CLOCK_MONOTONIC, which is unavailable there. Keep the last portable release
|
// 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/kes v0.3.1
|
||||||
github.com/minio/kms-go/kms v0.7.0
|
github.com/minio/kms-go/kms v0.7.0
|
||||||
github.com/minio/madmin-go/v3 v3.0.110
|
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/mux v1.10.1
|
||||||
github.com/minio/selfupdate v0.6.0
|
github.com/minio/selfupdate v0.6.0
|
||||||
github.com/minio/simdjson-go v0.4.5
|
github.com/minio/simdjson-go v0.4.5
|
||||||
@@ -81,7 +81,7 @@ require (
|
|||||||
github.com/nats-io/stan.go v0.10.4
|
github.com/nats-io/stan.go v0.10.4
|
||||||
github.com/ncw/directio v1.0.5
|
github.com/ncw/directio v1.0.5
|
||||||
github.com/nsqio/go-nsq v1.1.0
|
github.com/nsqio/go-nsq v1.1.0
|
||||||
github.com/pgsty/silo-pkg/v3 v3.15.1
|
github.com/pgsty/silo-pkg/v3 v3.15.2
|
||||||
github.com/philhofer/fwd v1.2.1
|
github.com/philhofer/fwd v1.2.1
|
||||||
github.com/pierrec/lz4/v4 v4.1.33
|
github.com/pierrec/lz4/v4 v4.1.33
|
||||||
github.com/pkg/errors v0.9.1
|
github.com/pkg/errors v0.9.1
|
||||||
|
|||||||
@@ -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/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 h1:Gdi1DZK69+ZVMoNHRXJyNcxrMA4dSxoYHZSQbirFg34=
|
||||||
github.com/minio/md5-simd v1.1.2/go.mod h1:MzdKDxYpY2BT9XQFocsiZf/NKVtR7nkE4RoEpN+20RM=
|
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.20261025165926-320af050c445 h1:TIdz2CWnaQQDML9I03RY0pdms4zfqpkGHPmnJxfvCrE=
|
||||||
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/go.mod h1:nQnQi7zTfXR4lREtYXZQWqo5aw0Rqv+ZqCGrTORygRk=
|
||||||
github.com/minio/mux v1.10.1 h1:grrK8SwRKbkNFE6qG7WAvFGH09bB46d5teOOtKfQ14s=
|
github.com/minio/mux v1.10.1 h1:grrK8SwRKbkNFE6qG7WAvFGH09bB46d5teOOtKfQ14s=
|
||||||
github.com/minio/mux v1.10.1/go.mod h1:INYT4sMSTJy0QWUEA/E2DZNxJ5sAxIwbnyZjkzNFRfE=
|
github.com/minio/mux v1.10.1/go.mod h1:INYT4sMSTJy0QWUEA/E2DZNxJ5sAxIwbnyZjkzNFRfE=
|
||||||
github.com/minio/pkg/v3 v3.11.0 h1:WjzWGP7lyuupJgUSRim1RDtNujHtESyMQF7qSR4r5Dg=
|
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/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/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/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-20261026120214-0ceb298cdcd0 h1:kxlOeGEFoBW/lgLQVBL9pcw3uq5KTMO6VmqdeDebTig=
|
||||||
github.com/pgsty/mc v0.0.0-20261025104502-dcc78d506f94/go.mod h1:H3Uwlm1Qj3kPsbmojhNnv3aj5a+QdjdTKwv4w4gMSF8=
|
github.com/pgsty/mc v0.0.0-20261026120214-0ceb298cdcd0/go.mod h1:dUu9hAVOMyAC+/9XrMdPxISLwJp537jXuv3bNW5+q5E=
|
||||||
github.com/pgsty/silo-console v0.0.0-20261025145732-989fa7150792 h1:wPnltE28wxotYQIKeg5OykplBcrhaWeTnWRxbYXOTtQ=
|
github.com/pgsty/silo-console v0.0.0-20261026143515-6a05e8d6ad67 h1:8wWjVNEqYjcO1rm+QVNyQKVSNmJT8Q8HJfWR3rnEJlo=
|
||||||
github.com/pgsty/silo-console v0.0.0-20261025145732-989fa7150792/go.mod h1:fMxNcVxan3mu9mc7HCRiczSyQoFrbZEGumEmpatxtds=
|
github.com/pgsty/silo-console v0.0.0-20261026143515-6a05e8d6ad67/go.mod h1:0CxxVsLvy+u91xMHwrLGI1gwc96sBKpB8JTpGWIOm7I=
|
||||||
github.com/pgsty/silo-pkg/v3 v3.15.1 h1:T5rqY3HR1WGybcjnx/ghBvliEgo7yNCA7JAC3f1LpTk=
|
github.com/pgsty/silo-pkg/v3 v3.15.2 h1:PXgFAgkymHXhq8m5Rbs4QofHENOy5M7hLbC0KKowwvs=
|
||||||
github.com/pgsty/silo-pkg/v3 v3.15.1/go.mod h1:GmTF6fIEkTrzgM+ysWWfDZIlVRD2DYAJkKjA5bjiWpQ=
|
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 h1:UtpS73q8bFgy/yH3gCq8LSNvj5gv/FowNOiWZa7uKwQ=
|
||||||
github.com/philhofer/fwd v1.2.1/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
|
github.com/philhofer/fwd v1.2.1/go.mod h1:RqIHx9QI14HlwKwm98g9Re5prTQ6LdeRQn+gXJFxsJM=
|
||||||
github.com/pierrec/lz4/v4 v4.1.33 h1:GjG1TJ1V4IzKP8L96muuuDNpTwd7D+l2ccXrjAbe014=
|
github.com/pierrec/lz4/v4 v4.1.33 h1:GjG1TJ1V4IzKP8L96muuuDNpTwd7D+l2ccXrjAbe014=
|
||||||
|
|||||||
@@ -25,7 +25,7 @@ imagePullSecrets: []
|
|||||||
##
|
##
|
||||||
mcImage:
|
mcImage:
|
||||||
repository: pgsty/mc
|
repository: pgsty/mc
|
||||||
tag: RELEASE.2026-10-25T00-00-00Z
|
tag: RELEASE.2026-10-26T00-00-00Z
|
||||||
pullPolicy: IfNotPresent
|
pullPolicy: IfNotPresent
|
||||||
|
|
||||||
## Silo mode, i.e. standalone or distributed.
|
## Silo mode, i.e. standalone or distributed.
|
||||||
|
|||||||
@@ -344,8 +344,7 @@ func (c *Subroute) newMuxClient(ctx context.Context) (*muxClient, error) {
|
|||||||
|
|
||||||
// Request allows to do a single remote request.
|
// Request allows to do a single remote request.
|
||||||
// 'req' will not be used after the call and caller can reuse.
|
// '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
|
// If no deadline is set on ctx, a 1-minute deadline will be added.
|
||||||
// WithoutDefaultTimeout was requested.
|
|
||||||
func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error) {
|
func (c *Connection) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error) {
|
||||||
if !h.valid() {
|
if !h.valid() {
|
||||||
return nil, ErrUnknownHandler
|
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.
|
// Request allows to do a single remote request.
|
||||||
// 'req' will not be used after the call and caller can reuse.
|
// '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
|
// If no deadline is set on ctx, a 1-minute deadline will be added.
|
||||||
// WithoutDefaultTimeout was requested.
|
|
||||||
func (c *Subroute) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error) {
|
func (c *Subroute) Request(ctx context.Context, h HandlerID, req []byte) ([]byte, error) {
|
||||||
if !h.valid() {
|
if !h.valid() {
|
||||||
return nil, ErrUnknownHandler
|
return nil, ErrUnknownHandler
|
||||||
|
|||||||
@@ -550,8 +550,7 @@ type Requester interface {
|
|||||||
|
|
||||||
// Call the remote with the request and return the response.
|
// Call the remote with the request and return the response.
|
||||||
// The response should be returned with PutResponse when no error.
|
// The response should be returned with PutResponse when no error.
|
||||||
// If no deadline is set, a 1-minute deadline is added unless
|
// If no deadline is set, a 1-minute deadline is added.
|
||||||
// WithoutDefaultTimeout was requested.
|
|
||||||
func (h *SingleHandler[Req, Resp]) Call(ctx context.Context, c Requester, req Req) (resp Resp, err error) {
|
func (h *SingleHandler[Req, Resp]) Call(ctx context.Context, c Requester, req Req) (resp Resp, err error) {
|
||||||
if c == nil {
|
if c == nil {
|
||||||
if h.ignoreNilConn {
|
if h.ignoreNilConn {
|
||||||
|
|||||||
@@ -59,16 +59,6 @@ type Response struct {
|
|||||||
Err error
|
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 {
|
func newMuxClient(ctx context.Context, muxID uint64, parent *Connection) *muxClient {
|
||||||
ctx, cancelFn := context.WithCancelCause(ctx)
|
ctx, cancelFn := context.WithCancelCause(ctx)
|
||||||
return &muxClient{
|
return &muxClient{
|
||||||
@@ -111,10 +101,8 @@ func (m *muxClient) roundtrip(h HandlerID, req []byte) ([]byte, error) {
|
|||||||
m.respMu.Unlock()
|
m.respMu.Unlock()
|
||||||
ctx := m.ctx
|
ctx := m.ctx
|
||||||
|
|
||||||
// Admitted storage mutations must not time out while the handler can
|
// Add deadline if none.
|
||||||
// still commit. A zero wire deadline also keeps their late response.
|
if msg.DeadlineMS == 0 {
|
||||||
noDefaultTimeout, _ := ctx.Value(withoutDefaultTimeoutKey{}).(bool)
|
|
||||||
if msg.DeadlineMS == 0 && !noDefaultTimeout {
|
|
||||||
msg.DeadlineMS = uint32(defaultSingleRequestTimeout / time.Millisecond)
|
msg.DeadlineMS = uint32(defaultSingleRequestTimeout / time.Millisecond)
|
||||||
var cancel context.CancelFunc
|
var cancel context.CancelFunc
|
||||||
ctx, cancel = context.WithTimeout(ctx, defaultSingleRequestTimeout)
|
ctx, cancel = context.WithTimeout(ctx, defaultSingleRequestTimeout)
|
||||||
|
|||||||
@@ -25,67 +25,43 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
)
|
)
|
||||||
|
|
||||||
func TestSingleMutationWaitsBeyondDefaultTimeout(t *testing.T) {
|
func TestSingleRequestDefaultTimeout(t *testing.T) {
|
||||||
for _, admitted := range []bool{false, true} {
|
synctest.Test(t, func(t *testing.T) {
|
||||||
t.Run(map[bool]string{false: "ordinary", true: "admitted"}[admitted], func(t *testing.T) {
|
conn := &Connection{outQueue: make(chan []byte, 1)}
|
||||||
synctest.Test(t, func(t *testing.T) {
|
client := newMuxClient(t.Context(), 1, conn)
|
||||||
ctx := t.Context()
|
defer client.cancelFn(context.Canceled)
|
||||||
if admitted {
|
done := make(chan error, 1)
|
||||||
ctx = WithoutDefaultTimeout(ctx)
|
go func() { _, err := client.roundtrip(handlerTest, nil); done <- err }()
|
||||||
}
|
wire := <-conn.outQueue
|
||||||
conn := &Connection{outQueue: make(chan []byte, 1)}
|
var request message
|
||||||
client := newMuxClient(ctx, 1, conn)
|
if _, err := request.UnmarshalMsg(wire); err != nil {
|
||||||
defer client.cancelFn(context.Canceled)
|
t.Fatal(err)
|
||||||
done := make(chan Response, 1)
|
}
|
||||||
go func() {
|
PutByteBuffer(wire)
|
||||||
msg, err := client.roundtrip(handlerTest, nil)
|
if request.DeadlineMS != uint32(defaultSingleRequestTimeout/time.Millisecond) {
|
||||||
done <- Response{Msg: msg, Err: err}
|
t.Fatalf("wire deadline: %d", request.DeadlineMS)
|
||||||
}()
|
}
|
||||||
wire := <-conn.outQueue
|
time.Sleep(defaultSingleRequestTimeout + time.Second)
|
||||||
var request message
|
synctest.Wait()
|
||||||
if _, err := request.UnmarshalMsg(wire); err != nil {
|
select {
|
||||||
t.Fatal(err)
|
case err := <-done:
|
||||||
}
|
if !errors.Is(err, context.DeadlineExceeded) {
|
||||||
PutByteBuffer(wire)
|
t.Fatal(err)
|
||||||
if admitted && request.DeadlineMS != 0 {
|
}
|
||||||
t.Errorf("admitted mutation tells the server to discard a late result after %dms", request.DeadlineMS)
|
default:
|
||||||
}
|
t.Fatal("request outlived the default timeout")
|
||||||
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 TestWithoutDefaultTimeoutStillEndsOnDisconnectOrCancellation(t *testing.T) {
|
func TestSingleRequestCancellationAndDisconnect(t *testing.T) {
|
||||||
for _, disconnect := range []bool{false, true} {
|
for _, disconnect := range []bool{false, true} {
|
||||||
t.Run(map[bool]string{false: "explicit-cancel", true: "disconnect"}[disconnect], func(t *testing.T) {
|
t.Run(map[bool]string{false: "explicit-cancel", true: "disconnect"}[disconnect], func(t *testing.T) {
|
||||||
synctest.Test(t, func(t *testing.T) {
|
synctest.Test(t, func(t *testing.T) {
|
||||||
ctx, cancel := context.WithCancel(t.Context())
|
ctx, cancel := context.WithCancel(t.Context())
|
||||||
defer cancel()
|
defer cancel()
|
||||||
conn := &Connection{outQueue: make(chan []byte, 1)}
|
conn := &Connection{outQueue: make(chan []byte, 1)}
|
||||||
client := newMuxClient(WithoutDefaultTimeout(ctx), 1, conn)
|
client := newMuxClient(ctx, 1, conn)
|
||||||
defer client.cancelFn(context.Canceled)
|
defer client.cancelFn(context.Canceled)
|
||||||
done := make(chan error, 1)
|
done := make(chan error, 1)
|
||||||
go func() {
|
go func() {
|
||||||
|
|||||||
@@ -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)
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -127,17 +127,6 @@ var deadlineWorkPanics atomic.Int64
|
|||||||
// If the deadline/context passes before the function finishes executing,
|
// If the deadline/context passes before the function finishes executing,
|
||||||
// the zero value and the context error is returned.
|
// 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) {
|
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)
|
ctx, cancel := context.WithTimeout(ctx, timeout)
|
||||||
defer cancel()
|
defer cancel()
|
||||||
|
|
||||||
@@ -174,10 +163,6 @@ func withDeadline[V any](ctx context.Context, timeout time.Duration, work func(c
|
|||||||
case v := <-c:
|
case v := <-c:
|
||||||
return v.val, v.err
|
return v.val, v.err
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
if wait {
|
|
||||||
v := <-c
|
|
||||||
return v.val, v.err
|
|
||||||
}
|
|
||||||
var zero V
|
var zero V
|
||||||
return zero, ctx.Err()
|
return zero, ctx.Err()
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -25,62 +25,9 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"strings"
|
"strings"
|
||||||
"testing"
|
"testing"
|
||||||
"testing/synctest"
|
|
||||||
"time"
|
"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 {
|
type sleepWriter struct {
|
||||||
timeout time.Duration
|
timeout time.Duration
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user