diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2ded81d919ca9..4cacc7dafad6c 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,17 @@
# Changelog
+## Unreleased
+
+- Add an `extended` object naming mode, seeded with `--object-naming=extended`
+ or `MINIO_OBJECT_NAMING=extended` on a cluster's first start and recorded in
+ every `format.json`. In this mode an object and a prefix of the same name
+ (`obj1` and `obj1/obj2`) coexist and are both listed: object names are
+ escaped and regular objects carry a terminal marker on disk, so an object
+ directory can never double as another key's parent. The default (safe) mode
+ keeps its existing behavior and disk layout; the mode cannot be changed or
+ migrated in place, and mixed modes across drives, pools or nodes are rejected
+ at startup. See `docs/minio-limits.md` for the resulting restrictions.
+
## RELEASE.2026-10-26T00-00-00Z
Package version: `20261026000000.0.0`; published from source `c3ad836aed0d`.
diff --git a/buildscripts/rebrand-guard/compat-baseline.json b/buildscripts/rebrand-guard/compat-baseline.json
index 2ff232340efc4..0f1e8f820d8c2 100644
--- a/buildscripts/rebrand-guard/compat-baseline.json
+++ b/buildscripts/rebrand-guard/compat-baseline.json
@@ -457,6 +457,7 @@
"MINIO_NOTIFY_WEBHOOK_QUEUE_DIR",
"MINIO_NOTIFY_WEBHOOK_QUEUE_LIMIT",
"MINIO_NTP_SERVER",
+ "MINIO_OBJECT_NAMING",
"MINIO_OLD",
"MINIO_OPERATOR_VERSION",
"MINIO_OPTS",
diff --git a/cmd/batch-expire.go b/cmd/batch-expire.go
index 6de32920a23ee..c4b7c30429f5d 100644
--- a/cmd/batch-expire.go
+++ b/cmd/batch-expire.go
@@ -456,7 +456,7 @@ func batchObjsForDelete(ctx context.Context, r *BatchJobExpire, ri *batchJobInfo
default:
}
stopFn := globalBatchJobsMetrics.trace(batchJobMetricExpire, ri.JobID, attempts)
- _, err := api.DeleteObject(ctx, exp.Bucket, encodeDirObject(exp.Name), ObjectOptions{
+ _, err := api.DeleteObject(ctx, exp.Bucket, exp.Name, ObjectOptions{
DeletePrefix: true,
DeletePrefixObject: true, // use prefix delete on exact object (this is an optimization to avoid fan-out calls)
})
diff --git a/cmd/bucket-object-lock.go b/cmd/bucket-object-lock.go
index 397b039ef038d..9b624fccbd0a6 100644
--- a/cmd/bucket-object-lock.go
+++ b/cmd/bucket-object-lock.go
@@ -275,17 +275,17 @@ func checkObjectLockedForOverwrite(ctx context.Context, bucket, object string, g
return err
}
if objectlock.GetObjectLegalHoldMeta(objInfo.UserDefined).Status == objectlock.LegalHoldOn {
- return ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: nullVersionID}
+ return ObjectLocked{Bucket: bucket, Object: decodeObjectName(object), VersionID: nullVersionID}
}
r := objectlock.GetObjectRetentionMeta(objInfo.UserDefined)
if r.Mode == objectlock.RetCompliance || r.Mode == objectlock.RetGovernance {
t, err := objectlock.UTCNowNTP()
if err != nil {
internalLogIf(ctx, err, logger.WarningKind)
- return ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: nullVersionID}
+ return ObjectLocked{Bucket: bucket, Object: decodeObjectName(object), VersionID: nullVersionID}
}
if r.RetainUntilDate.After(t) {
- return ObjectLocked{Bucket: bucket, Object: decodeDirObject(object), VersionID: nullVersionID}
+ return ObjectLocked{Bucket: bucket, Object: decodeObjectName(object), VersionID: nullVersionID}
}
}
return nil
diff --git a/cmd/common-main.go b/cmd/common-main.go
index 3b5a08c01b3d5..c4b53d9b3378e 100644
--- a/cmd/common-main.go
+++ b/cmd/common-main.go
@@ -442,6 +442,7 @@ func buildServerCtxt(ctx *cli.Context, ctxt *serverCtxt) (err error) {
ctxt.FTP = ctx.StringSlice("ftp")
ctxt.SFTP = ctx.StringSlice("sftp")
ctxt.Interface = ctx.String("interface")
+ ctxt.ObjectNaming = ctx.String("object-naming")
ctxt.UserTimeout = ctx.Duration("conn-user-timeout")
ctxt.SendBufSize = ctx.Int("send-buf-size")
ctxt.RecvBufSize = ctx.Int("recv-buf-size")
@@ -771,6 +772,16 @@ func serverHandleEarlyEnvVars() {
func serverHandleEnvVars() {
var err error
+ // Object naming mode must be valid before any format is written. The
+ // recorded format is authoritative for an existing cluster; a mismatch is
+ // rejected when the object layer starts.
+ if mode := globalServerCtxt.ObjectNaming; mode != "" {
+ if !isObjectNamingModeValid(mode) {
+ logger.Fatal(fmt.Errorf("invalid object naming mode %q: must be %q or %q", mode, objectNamingSafe, objectNamingExtended), "Invalid %s value in environment variable", config.EnvObjectNaming)
+ }
+ }
+ globalObjectNaming = globalServerCtxt.ObjectNaming
+
// Re-read the source-address trust policy now that loadEnvVarsFromFiles has
// run: a policy taken at package initialisation would miss every deployment
// configured through MINIO_CONFIG_ENV_FILE. Refuse to start on a malformed
diff --git a/cmd/data-scanner.go b/cmd/data-scanner.go
index 0475c4a530c34..87525f2e4f6e1 100644
--- a/cmd/data-scanner.go
+++ b/cmd/data-scanner.go
@@ -419,6 +419,9 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
// If there are lifecycle rules for the prefix.
_, prefix := path2BucketObjectWithBasePath(f.root, folder.name)
+ // Folder names are disk paths; rules and reported prefixes use the
+ // user-domain key, which differs in extended naming mode.
+ prefix = decodeExtendedDiskPath(prefix)
var activeLifeCycle *lifecycle.Lifecycle
if f.oldCache.Info.lifeCycle != nil && f.oldCache.Info.lifeCycle.HasActiveRules(prefix) {
if f.dataUsageScannerDebug {
@@ -447,7 +450,13 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
}
return nil
}
- bucket, prefix := path2BucketObjectWithBasePath(f.root, entName)
+ // Object API calls below run in the user domain, while Path stays
+ // the raw disk path for metadata reads.
+ userEntName := entName
+ if globalObjectNamingExtended {
+ userEntName = decodeExtendedDiskPath(entName)
+ }
+ bucket, prefix := path2BucketObjectWithBasePath(f.root, userEntName)
if bucket == "" {
if f.dataUsageScannerDebug {
console.Debugf(scannerLogPrefix+" no bucket (%s,%s)\n", f.root, entName)
@@ -522,8 +531,8 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
// Object already accounted for, remove from heal map,
// simply because getSize() function already heals the
- // object.
- delete(abandonedChildren, pathJoin(item.bucket, item.objectPath()))
+ // object. Cache children are keyed by disk paths.
+ delete(abandonedChildren, path.Dir(entName))
if err != errIgnoreFileContrib {
into.addSizes(sz)
@@ -713,6 +722,8 @@ func (f *folderScanner) scanFolder(ctx context.Context, folder cachedFolder, int
break
}
bucket, prefix := path2BucketObject(k)
+ // WalkDir takes user-domain paths.
+ prefix = decodeObjectName(prefix)
stopFn := globalScannerMetrics.time(scannerMetricCheckMissing)
f.updateCurrentPath(k)
@@ -1260,7 +1271,7 @@ func applyExpiryOnNonTransitionedObjects(ctx context.Context, objLayer ObjectLay
}
}()
- dobj, err = objLayer.DeleteObject(ctx, obj.Bucket, encodeDirObject(obj.Name), opts)
+ dobj, err = objLayer.DeleteObject(ctx, obj.Bucket, obj.Name, opts)
if err != nil {
if isErrObjectNotFound(err) || isErrVersionNotFound(err) {
traceFn(ILMExpiry, nil, nil)
diff --git a/cmd/data-usage.go b/cmd/data-usage.go
index 51227e106defb..e8ada75b6c70f 100644
--- a/cmd/data-usage.go
+++ b/cmd/data-usage.go
@@ -96,8 +96,8 @@ func loadPrefixUsageFromBackend(ctx context.Context, objAPI ObjectLayer, bucket
}
for id, usageInfo := range cache.flattenChildrens(*root) {
- prefix := decodeDirObject(strings.TrimPrefix(id, bucket+slashSeparator))
- // decodeDirObject to avoid any __XLDIR__ objects
+ prefix := decodeObjectName(strings.TrimPrefix(id, bucket+slashSeparator))
+ // decodeObjectName undoes __XLDIR__ and extended naming.
m[prefix] += uint64(usageInfo.Size)
}
}
diff --git a/cmd/erasure-healing.go b/cmd/erasure-healing.go
index 8fb3712708a29..013fab4f63e5b 100644
--- a/cmd/erasure-healing.go
+++ b/cmd/erasure-healing.go
@@ -248,7 +248,7 @@ func (er *erasureObjects) auditHealObject(ctx context.Context, bucket, object, v
opts := AuditLogOptions{
Event: "HealObject",
Bucket: bucket,
- Object: decodeDirObject(object),
+ Object: decodeObjectName(object),
VersionID: versionID,
}
if err != nil {
@@ -1120,7 +1120,7 @@ func healTrace(funcName healingMetric, startTime time.Time, bucket, object strin
NodeName: globalLocalNodeName,
FuncName: "heal." + funcName.String(),
Duration: time.Since(startTime),
- Path: pathJoin(bucket, decodeDirObject(object)),
+ Path: pathJoin(bucket, decodeObjectName(object)),
}
if opts != nil {
tr.Custom = map[string]string{
diff --git a/cmd/erasure-object.go b/cmd/erasure-object.go
index 5c9ee5d805179..e63e692b07efb 100644
--- a/cmd/erasure-object.go
+++ b/cmd/erasure-object.go
@@ -777,7 +777,7 @@ func (er erasureObjects) readListingFileInfo(ctx context.Context, bucket, object
}
func listingReadQuorumError(bucket, object string) error {
- return InsufficientReadQuorum{Bucket: bucket, Object: decodeDirObject(object), Err: errErasureReadQuorum, Type: RQInconsistentMeta}
+ return InsufficientReadQuorum{Bucket: bucket, Object: decodeObjectName(object), Err: errErasureReadQuorum, Type: RQInconsistentMeta}
}
func (er erasureObjects) getObjectFileInfo(ctx context.Context, bucket, object string, opts ObjectOptions, readData bool) (FileInfo, []FileInfo, []StorageAPI, error) {
diff --git a/cmd/erasure-server-pool-consistency.go b/cmd/erasure-server-pool-consistency.go
index 776e1e4cc4cfe..cea6d8edade8b 100644
--- a/cmd/erasure-server-pool-consistency.go
+++ b/cmd/erasure-server-pool-consistency.go
@@ -56,9 +56,9 @@ func (z *erasureServerPools) objectPoolInfos(ctx context.Context, bucket, object
})
if len(copies) == 0 {
if opts.VersionID != "" {
- return nil, VersionNotFound{Bucket: bucket, Object: decodeDirObject(object), VersionID: opts.VersionID}
+ return nil, VersionNotFound{Bucket: bucket, Object: decodeObjectNameForBucket(bucket, object), VersionID: opts.VersionID}
}
- return nil, ObjectNotFound{Bucket: bucket, Object: decodeDirObject(object)}
+ return nil, ObjectNotFound{Bucket: bucket, Object: decodeObjectNameForBucket(bucket, object)}
}
return copies, nil
}
@@ -102,7 +102,7 @@ func (z *erasureServerPools) checkDataMovementCopies(ctx context.Context, bucket
}
}
if !sourcePresent {
- return nil, VersionNotFound{Bucket: bucket, Object: decodeDirObject(object), VersionID: versionID}
+ return nil, VersionNotFound{Bucket: bucket, Object: decodeObjectNameForBucket(bucket, object), VersionID: versionID}
}
for _, copy := range copies {
if copy.Index != opts.SrcPoolIdx && copy.ObjInfo.ModTime.After(opts.MTime) {
@@ -131,7 +131,7 @@ func (z *erasureServerPools) checkDataMovementCopies(ctx context.Context, bucket
// snapshot. A concurrent delete can create a new marker in the source pool;
// deleting the whole prefix would remove that marker and expose moved data.
func (z *erasureServerPools) removeMovedVersions(ctx context.Context, bucket, object string, source int, versions []FileInfo) error {
- object = encodeDirObject(object)
+ object = encodeObjectNameForBucket(bucket, object)
lk := z.NewNSLock(bucket, object)
lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
if err != nil {
@@ -265,7 +265,7 @@ func replicaCopiesInfo(copies []PoolObjInfo, err error, bucket, object, versionI
return ObjectInfo{}, err
}
if copies[0].ObjInfo.DeleteMarker {
- return copies[0].ObjInfo, MethodNotAllowed{Bucket: bucket, Object: decodeDirObject(object), VersionID: versionID}
+ return copies[0].ObjInfo, MethodNotAllowed{Bucket: bucket, Object: decodeObjectNameForBucket(bucket, object), VersionID: versionID}
}
return mergedPoolObjectInfo(copies), nil
}
@@ -278,7 +278,7 @@ func (z *erasureServerPools) metadataPoolInfos(ctx context.Context, bucket, obje
return nil, err
}
if copies[0].ObjInfo.DeleteMarker {
- return nil, MethodNotAllowed{Bucket: bucket, Object: decodeDirObject(object), VersionID: opts.VersionID}
+ return nil, MethodNotAllowed{Bucket: bucket, Object: decodeObjectNameForBucket(bucket, object), VersionID: opts.VersionID}
}
if opts.VersionID == "" {
opts.VersionID = copies[0].ObjInfo.VersionID
@@ -417,7 +417,7 @@ func (z *erasureServerPools) retireReplicaCopies(ctx context.Context, bucket, ob
}
}
if len(retained) == 0 {
- return VersionNotFound{Bucket: bucket, Object: decodeDirObject(object), VersionID: versionID}
+ return VersionNotFound{Bucket: bucket, Object: decodeObjectNameForBucket(bucket, object), VersionID: versionID}
}
for i, copy := range copies {
if copy.Index == keep {
@@ -513,7 +513,7 @@ func (z *erasureServerPools) deleteObjectReconciled(ctx context.Context, bucket,
// A single new delete marker hides the current version. Older versions
// remain history and must not be removed by an unqualified DELETE.
oi, err := z.serverPools[primary.Index].DeleteObject(ctx, bucket, object, opts)
- oi.Name = decodeDirObject(object)
+ oi.Name = decodeObjectNameForBucket(bucket, object)
oi.replicationDecision = opts.DeleteReplication.ReplicateDecisionStr
return oi, err
}
@@ -530,7 +530,7 @@ func (z *erasureServerPools) deleteObjectReconciled(ctx context.Context, bucket,
}
}
oi, err := z.serverPools[primary.Index].DeleteObject(ctx, bucket, object, opts)
- oi.Name = decodeDirObject(object)
+ oi.Name = decodeObjectNameForBucket(bucket, object)
oi.replicationDecision = opts.DeleteReplication.ReplicateDecisionStr
return oi, err
}
@@ -562,7 +562,7 @@ func (z *erasureServerPools) deleteObjectsReconciled(ctx context.Context, bucket
}
} else if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
if isErrReadQuorum(err) {
- err = InsufficientWriteQuorum{Bucket: bucket, Object: decodeDirObject(obj.ObjectName)}
+ err = InsufficientWriteQuorum{Bucket: bucket, Object: decodeObjectNameForBucket(bucket, obj.ObjectName)}
}
errs[i] = err
}
diff --git a/cmd/erasure-server-pool-decom.go b/cmd/erasure-server-pool-decom.go
index 3c21825bb5672..18e69fc62d034 100644
--- a/cmd/erasure-server-pool-decom.go
+++ b/cmd/erasure-server-pool-decom.go
@@ -617,7 +617,7 @@ func (z *erasureServerPools) decommissionObject(ctx context.Context, idx int, bu
if objInfo.isMultipart() {
// Completion repeats this check under the object write lock.
- if _, err := z.checkDataMovementCopies(ctx, bucket, encodeDirObject(objInfo.Name), ObjectOptions{
+ if _, err := z.checkDataMovementCopies(ctx, bucket, encodeObjectNameForBucket(bucket, objInfo.Name), ObjectOptions{
VersionID: objInfo.VersionID, MTime: objInfo.ModTime, SrcPoolIdx: idx,
}); err != nil {
return err
@@ -921,7 +921,7 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
}
gr, err := set.GetObjectNInfo(ctx,
bi.Name,
- encodeDirObject(version.Name),
+ encodeObjectNameForBucket(bi.Name, version.Name),
nil,
http.Header{},
ObjectOptions{
@@ -951,6 +951,11 @@ func (z *erasureServerPools) decommissionPool(ctx context.Context, idx int, pool
stopFn(version.Size, err)
continue
}
+ // Restore the user-domain name: the set layer works on the
+ // storage-stack form, while the pooled APIs below expect the
+ // original key.
+ gr.ObjInfo.Name = version.Name
+
if err = z.decommissionObject(ctx, idx, bi.Name, gr); err != nil && !isDataMovementSuperseded(err) {
if isErrObjectNotFound(err) || isErrVersionNotFound(err) || isDataMovementOverWriteErr(err) {
ignore = true
diff --git a/cmd/erasure-server-pool-rebalance.go b/cmd/erasure-server-pool-rebalance.go
index 6c64f9f289add..694e61fe7acff 100644
--- a/cmd/erasure-server-pool-rebalance.go
+++ b/cmd/erasure-server-pool-rebalance.go
@@ -690,7 +690,7 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
// GetObjectReader.Close is called by rebalanceObject
gr, err := set.GetObjectNInfo(ctx,
bucket,
- encodeDirObject(version.Name),
+ encodeObjectNameForBucket(bucket, version.Name),
nil,
http.Header{},
ObjectOptions{
@@ -712,6 +712,11 @@ func (z *erasureServerPools) rebalanceBucket(ctx context.Context, bucket string,
continue
}
+ // Restore the user-domain name: the set layer works on the
+ // storage-stack form, while the pooled APIs below expect the
+ // original key.
+ gr.ObjInfo.Name = version.Name
+
if err = z.rebalanceObject(ctx, poolIdx, bucket, gr); err != nil && !isDataMovementSuperseded(err) {
// This can happen when rebalance stop races with ongoing rebalance workers.
// These rebalance failures can be ignored.
@@ -849,7 +854,7 @@ func (z *erasureServerPools) rebalanceObject(ctx context.Context, poolIdx int, b
if oi.isMultipart() {
// Completion repeats this check under the object write lock.
- if _, err := z.checkDataMovementCopies(ctx, bucket, encodeDirObject(oi.Name), ObjectOptions{
+ if _, err := z.checkDataMovementCopies(ctx, bucket, encodeObjectNameForBucket(bucket, oi.Name), ObjectOptions{
VersionID: oi.VersionID, MTime: oi.ModTime, SrcPoolIdx: poolIdx,
}); err != nil {
return err
diff --git a/cmd/erasure-server-pool.go b/cmd/erasure-server-pool.go
index a75af733996e8..5319ec03323f1 100644
--- a/cmd/erasure-server-pool.go
+++ b/cmd/erasure-server-pool.go
@@ -79,6 +79,7 @@ func (z *erasureServerPools) SinglePool() bool {
func newErasureServerPools(ctx context.Context, endpointServerPools EndpointServerPools) (ObjectLayer, error) {
var (
deploymentID string
+ objectNaming string
commonParityDrives int
err error
@@ -154,6 +155,19 @@ func newErasureServerPools(ctx context.Context, endpointServerPools EndpointServ
return nil, fmt.Errorf("all pools must have same deployment ID - expected %s, got %s for pool(%s)", deploymentID, formats[i].ID, humanize.Ordinal(i+1))
}
+ // Object naming is a format-wide choice: every pool must agree, or
+ // keys would be addressable only in part of the deployment. The mode
+ // recorded on disk always wins over the command line.
+ naming := normalizeObjectNamingMode(formats[i].Erasure.ObjectNaming)
+ if requested := globalObjectNaming; requested != "" && requested != naming {
+ return nil, fmt.Errorf("object naming mode %q requested but the cluster was formatted with %q: object naming cannot be changed after the cluster was formatted", requested, naming)
+ }
+ if objectNaming == "" {
+ objectNaming = naming
+ } else if naming != objectNaming {
+ return nil, fmt.Errorf("all pools must use the same object naming mode - expected %s, got %s for pool(%s)", objectNaming, naming, humanize.Ordinal(i+1))
+ }
+
bootstrapTrace(fmt.Sprintf("newErasureSets: initializing %s pool", humanize.Ordinal(i+1)), func() {
z.serverPools[i], err = newErasureSets(ctx, ep, storageDisks[i], formats[i], commonParityDrives, i)
})
@@ -172,6 +186,9 @@ func newErasureServerPools(ctx context.Context, endpointServerPools EndpointServ
}
}
+ // The mode recorded in the formats is authoritative from here on.
+ globalObjectNamingExtended = objectNaming == objectNamingExtended
+
if !globalIsDistErasure {
globalLocalDrivesMu.Lock()
globalLocalDrivesMap = make(map[string]StorageAPI, len(localDrives))
@@ -961,10 +978,14 @@ func (z *erasureServerPools) GetObjectNInfo(ctx context.Context, bucket, object
gr = nil
err = nil
- object = encodeDirObject(object)
+ object = encodeObjectNameForBucket(bucket, object)
if z.SinglePool() {
- return z.serverPools[0].GetObjectNInfo(ctx, bucket, object, rs, h, opts)
+ gr, err := z.serverPools[0].GetObjectNInfo(ctx, bucket, object, rs, h, opts)
+ if err == nil {
+ gr.ObjInfo.Name = decodeObjectNameForBucket(bucket, object)
+ }
+ return gr, err
}
var unlockOnDefer bool
@@ -1017,6 +1038,8 @@ func (z *erasureServerPools) GetObjectNInfo(ctx context.Context, bucket, object
return nil, err
}
+ gr.ObjInfo.Name = decodeObjectNameForBucket(bucket, object)
+
if unlockOnDefer {
unlockOnDefer = gr.ObjInfo.Inlined
}
@@ -1031,7 +1054,8 @@ func (z *erasureServerPools) GetObjectNInfo(ctx context.Context, bucket, object
// is present in-case there were duplicate writes to both pools, this function also returns the
// additional index where the latest object exists, that is used to start the GetObject stream.
func (z *erasureServerPools) getLatestObjectInfoWithIdx(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, int, error) {
- object = encodeDirObject(object)
+ // Callers pass the stack-domain name (already encoded by the public
+ // method they were entered through).
results := make([]struct {
zIdx int
oi ObjectInfo
@@ -1064,11 +1088,13 @@ func (z *erasureServerPools) getLatestObjectInfoWithIdx(ctx context.Context, buc
for _, res := range results {
err := res.err
if err == nil {
+ res.oi.Name = decodeObjectNameForBucket(bucket, object)
return res.oi, res.zIdx, nil
}
if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
// some errors such as MethodNotAllowed for delete marker
// should be returned upwards.
+ res.oi.Name = decodeObjectNameForBucket(bucket, object)
return res.oi, res.zIdx, err
}
// When its a delete marker and versionID is empty
@@ -1078,7 +1104,7 @@ func (z *erasureServerPools) getLatestObjectInfoWithIdx(ctx context.Context, buc
}
}
- object = decodeDirObject(object)
+ object = decodeObjectNameForBucket(bucket, object)
if opts.VersionID != "" {
return ObjectInfo{}, -1, VersionNotFound{Bucket: bucket, Object: object, VersionID: opts.VersionID}
}
@@ -1100,10 +1126,14 @@ func (z *erasureServerPools) GetObjectInfo(ctx context.Context, bucket, object s
err = nil
objInfo = ObjectInfo{}
- object = encodeDirObject(object)
+ object = encodeObjectNameForBucket(bucket, object)
if z.SinglePool() {
- return z.serverPools[0].GetObjectInfo(ctx, bucket, object, opts)
+ oi, err := z.serverPools[0].GetObjectInfo(ctx, bucket, object, opts)
+ if err == nil {
+ oi.Name = decodeObjectNameForBucket(bucket, object)
+ }
+ return oi, err
}
if !opts.NoLock {
@@ -1130,13 +1160,17 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec
return ObjectInfo{}, err
}
- object = encodeDirObject(object)
+ object = encodeObjectNameForBucket(bucket, object)
if z.SinglePool() {
_, err := z.getPoolIdx(ctx, bucket, object, data.Size())
if err != nil {
return ObjectInfo{}, err
}
- return z.serverPools[0].PutObject(ctx, bucket, object, data, opts)
+ oi, err := z.serverPools[0].PutObject(ctx, bucket, object, data, opts)
+ if err == nil {
+ oi.Name = decodeObjectNameForBucket(bucket, object)
+ }
+ return oi, err
}
if !opts.NoLock {
@@ -1218,12 +1252,17 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec
if err == nil && opts.ReplicaLockReconcile && !opts.DataMovement {
err = z.retireReplicaCopies(ctx, bucket, object, idx, oi)
}
+ if err == nil {
+ oi.Name = decodeObjectNameForBucket(bucket, object)
+ }
return oi, err
}
func (z *erasureServerPools) deletePrefix(ctx context.Context, bucket string, prefix string) error {
for _, pool := range z.serverPools {
- if _, err := pool.DeleteObject(ctx, bucket, prefix, ObjectOptions{DeletePrefix: true}); err != nil {
+ // Call the set layer directly: prefix is already in the storage-stack
+ // form, and re-entering the public method would escape it again.
+ if err := pool.deletePrefix(ctx, bucket, prefix); err != nil {
return err
}
}
@@ -1235,8 +1274,15 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
return objInfo, err
}
- if !opts.DeletePrefix { // DeletePrefix handles dir object encoding differently.
- object = encodeDirObject(object)
+ switch {
+ case !opts.DeletePrefix, opts.DeletePrefixObject:
+ // A normal delete, or a prefix delete scoped to one exact object
+ // (lifecycle "expire all versions"), addresses the object.
+ object = encodeObjectNameForBucket(bucket, object)
+ default:
+ // A bare prefix delete addresses a directory tree, not an object:
+ // escape the components but do not add the terminal marker.
+ object = encodeObjectPrefixName(object)
}
// Serialize logical deletion with pooled writes and metadata updates.
@@ -1267,7 +1313,7 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
return objInfo, DataMovementOverwriteErr{Bucket: bucket, Object: object, VersionID: opts.VersionID, Err: errDataMovementSrcDstPoolSame}
}
objInfo, err = z.serverPools[idx].DeleteObject(ctx, bucket, object, opts)
- objInfo.Name = decodeDirObject(object)
+ objInfo.Name = decodeObjectNameForBucket(bucket, object)
return objInfo, err
}
@@ -1352,7 +1398,7 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
// Delete marker already present we are not going to create new delete markers.
if pinfo.ObjInfo.DeleteMarker && opts.VersionID == "" {
- pinfo.ObjInfo.Name = decodeDirObject(object)
+ pinfo.ObjInfo.Name = decodeObjectNameForBucket(bucket, object)
return pinfo.ObjInfo, nil
}
@@ -1369,7 +1415,7 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
// All replication requests needs to go to pool with the object.
if opts.ReplicationRequest {
objInfo, err = z.serverPools[pinfo.Index].DeleteObject(ctx, bucket, object, opts)
- objInfo.Name = decodeDirObject(object)
+ objInfo.Name = decodeObjectNameForBucket(bucket, object)
return objInfo, err
}
@@ -1383,7 +1429,7 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
continue
}
objInfo, err = pool.DeleteObject(ctx, bucket, object, opts)
- objInfo.Name = decodeDirObject(object)
+ objInfo.Name = decodeObjectNameForBucket(bucket, object)
return objInfo, err
}
}
@@ -1391,16 +1437,16 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
for _, pool := range z.serverPools {
objInfo, err := pool.DeleteObject(ctx, bucket, object, opts)
if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
- objInfo.Name = decodeDirObject(object)
+ objInfo.Name = decodeObjectNameForBucket(bucket, object)
return objInfo, err
}
if err == nil {
- objInfo.Name = decodeDirObject(object)
+ objInfo.Name = decodeObjectNameForBucket(bucket, object)
return objInfo, nil
}
}
- objInfo.Name = decodeDirObject(object)
+ objInfo.Name = decodeObjectNameForBucket(bucket, object)
if opts.VersionID != "" {
return objInfo, VersionNotFound{Bucket: bucket, Object: object, VersionID: opts.VersionID}
}
@@ -1408,7 +1454,7 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
}
func (z *erasureServerPools) deleteObjectFromAllPools(ctx context.Context, bucket string, object string, opts ObjectOptions, poolIndices []poolErrs) (objInfo ObjectInfo, err error) {
- objInfo.Name = decodeDirObject(object)
+ objInfo.Name = decodeObjectNameForBucket(bucket, object)
// Delete in all server pools that reported no error or read quorum error
// where the read quorum issue is from metadata inconsistency. The
@@ -1427,11 +1473,11 @@ func (z *erasureServerPools) deleteObjectFromAllPools(ctx context.Context, bucke
break
}
if derr != nil && !isErrObjectNotFound(derr) && !isErrVersionNotFound(derr) {
- dobj.Name = decodeDirObject(object)
+ dobj.Name = decodeObjectNameForBucket(bucket, object)
return dobj, derr
}
}
- objInfo.Name = decodeDirObject(object)
+ objInfo.Name = decodeObjectNameForBucket(bucket, object)
return objInfo, err
}
@@ -1485,14 +1531,16 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o
dobjects := make([]DeletedObject, len(objects))
objSets := set.NewStringSet()
for i := range derrs {
- objects[i].ObjectName = encodeDirObject(objects[i].ObjectName)
+ // Validate the user-domain key before the stack transformation; the
+ // escaped form is longer and would be rejected near the name limit.
+ derrs[i] = checkDelObjArgs(ctx, bucket, objects[i].ObjectName)
+ userName := objects[i].ObjectName
+ objects[i].ObjectName = encodeObjectNameForBucket(bucket, userName)
dobjects[i] = DeletedObject{
- ObjectName: decodeDirObject(objects[i].ObjectName),
+ ObjectName: userName,
VersionID: objects[i].VersionID,
ReplicationState: objects[i].ReplicationState(),
}
-
- derrs[i] = checkDelObjArgs(ctx, bucket, objects[i].ObjectName)
objSets.Add(objects[i].ObjectName)
}
@@ -1580,7 +1628,7 @@ func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, o
}
for i := range dobjects {
- dobjects[i].ObjectName = decodeDirObject(dobjects[i].ObjectName)
+ dobjects[i].ObjectName = decodeObjectNameForBucket(bucket, dobjects[i].ObjectName)
}
return dobjects, derrs
}
@@ -1593,8 +1641,8 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
return ObjectInfo{}, err
}
- srcObject = encodeDirObject(srcObject)
- dstObject = encodeDirObject(dstObject)
+ srcObject = encodeObjectNameForBucket(srcBucket, srcObject)
+ dstObject = encodeObjectNameForBucket(dstBucket, dstObject)
cpSrcDstSame := isStringEqual(pathJoin(srcBucket, srcObject), pathJoin(dstBucket, dstObject))
@@ -1621,6 +1669,7 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
oi, err := z.serverPools[idx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)
if err == nil {
err = z.retireReplicaCopies(ctx, dstBucket, dstObject, idx, oi)
+ oi.Name = decodeObjectNameForBucket(dstBucket, dstObject)
}
return oi, err
}
@@ -1641,17 +1690,26 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
}
}
+ // decodeCopyResult restores the user-domain name on a result built from
+ // the stack-domain destination name.
+ decodeCopyResult := func(oi ObjectInfo, err error) (ObjectInfo, error) {
+ if err == nil {
+ oi.Name = decodeObjectNameForBucket(dstBucket, dstObject)
+ }
+ return oi, err
+ }
+
// CopyObjectHandler predicts the outcome of this decision in
// copyRewritesObjectData(); keep the two in sync.
if cpSrcDstSame && srcInfo.metadataOnly {
// Version ID is set for the destination and source == destination version ID.
if dstOpts.VersionID != "" && srcOpts.VersionID == dstOpts.VersionID {
- return z.serverPools[poolIdx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)
+ return decodeCopyResult(z.serverPools[poolIdx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts))
}
// Destination is not versioned and source version ID is empty
// perform an in-place update.
if !dstOpts.Versioned && srcOpts.VersionID == "" {
- return z.serverPools[poolIdx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)
+ return decodeCopyResult(z.serverPools[poolIdx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts))
}
// Destination is versioned, source is not destination version,
// as a special case look for if the source object is not legacy
@@ -1661,7 +1719,7 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
// CopyObject optimization where we don't create an entire copy
// of the content, instead we add a reference.
srcInfo.versionOnly = true
- return z.serverPools[poolIdx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts)
+ return decodeCopyResult(z.serverPools[poolIdx].CopyObject(ctx, srcBucket, srcObject, dstBucket, dstObject, srcInfo, srcOpts, dstOpts))
}
}
@@ -1682,7 +1740,7 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
replicaObjectInfo: dstOpts.replicaObjectInfo,
}
- return z.serverPools[poolIdx].PutObject(ctx, dstBucket, dstObject, srcInfo.PutObjReader, putOpts)
+ return decodeCopyResult(z.serverPools[poolIdx].PutObject(ctx, dstBucket, dstObject, srcInfo.PutObjReader, putOpts))
}
func (z *erasureServerPools) ListObjects(ctx context.Context, bucket, prefix, marker, delimiter string, maxKeys int) (ListObjectsInfo, error) {
@@ -2055,20 +2113,37 @@ func (z *erasureServerPools) ListMultipartUploads(ctx context.Context, bucket, p
return poolResult, nil
}
+ // The non-empty-prefix path addresses one exact upload key: it lists the
+ // SHA directory for that name. Apply the same name transformation used
+ // when the upload was created so the directory is found.
+ searchKey := encodeObjectNameForBucket(bucket, prefix)
+ decodeUploads := func(result *ListMultipartsInfo) {
+ result.Prefix = decodeObjectNameForBucket(bucket, result.Prefix)
+ for i := range result.Uploads {
+ result.Uploads[i].Object = decodeObjectNameForBucket(bucket, result.Uploads[i].Object)
+ }
+ }
+
if z.SinglePool() {
- return z.serverPools[0].getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
+ result, err := z.serverPools[0].getHashedSet(searchKey).listMultipartUploadsExact(ctx, bucket, searchKey, keyMarker, uploadIDMarker, delimiter, maxUploads)
+ decodeUploads(&result)
+ return result, err
}
for idx, pool := range z.serverPools {
if z.IsSuspended(idx) {
continue
}
- result, err := pool.getHashedSet(prefix).listMultipartUploadsExact(ctx, bucket, prefix, keyMarker, uploadIDMarker, delimiter, maxUploads)
+ result, err := pool.getHashedSet(searchKey).listMultipartUploadsExact(ctx, bucket, searchKey, keyMarker, uploadIDMarker, delimiter, maxUploads)
if err != nil {
+ decodeUploads(&result)
return result, err
}
poolResult.Uploads = append(poolResult.Uploads, result.Uploads...)
}
+ for i := range poolResult.Uploads {
+ poolResult.Uploads[i].Object = decodeObjectNameForBucket(bucket, poolResult.Uploads[i].Object)
+ }
return poolResult, nil
}
@@ -2079,11 +2154,14 @@ func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, obj
return nil, err
}
+ userObject := object
+ object = encodeObjectNameForBucket(bucket, object)
+
defer func() {
if err == nil && mp != nil {
z.mpCache.Store(mp.UploadID, MultipartInfo{
Bucket: bucket,
- Object: object,
+ Object: userObject,
UploadID: mp.UploadID,
Initiated: time.Now(),
})
@@ -2121,7 +2199,7 @@ func (z *erasureServerPools) NewMultipartUpload(ctx context.Context, bucket, obj
if opts.DataMovement && idx == opts.SrcPoolIdx {
return nil, DataMovementOverwriteErr{
Bucket: bucket,
- Object: object,
+ Object: userObject,
VersionID: opts.VersionID,
Err: errDataMovementSrcDstPoolSame,
}
@@ -2146,6 +2224,9 @@ func (z *erasureServerPools) PutObjectPart(ctx context.Context, bucket, object,
return PartInfo{}, err
}
+ userObject := object
+ object = encodeObjectNameForBucket(bucket, object)
+
if z.SinglePool() {
_, err := z.getPoolIdx(ctx, bucket, object, data.Size())
if err != nil {
@@ -2172,7 +2253,7 @@ func (z *erasureServerPools) PutObjectPart(ctx context.Context, bucket, object,
return PartInfo{}, InvalidUploadID{
Bucket: bucket,
- Object: object,
+ Object: userObject,
UploadID: uploadID,
}
}
@@ -2182,6 +2263,9 @@ func (z *erasureServerPools) GetMultipartInfo(ctx context.Context, bucket, objec
return MultipartInfo{}, err
}
+ userObject := object
+ object = encodeObjectNameForBucket(bucket, object)
+
uploadIDLock := z.NewNSLock(bucket, pathJoin(object, uploadID))
lkctx, err := uploadIDLock.GetRLock(ctx, globalOperationTimeout)
if err != nil {
@@ -2191,7 +2275,9 @@ func (z *erasureServerPools) GetMultipartInfo(ctx context.Context, bucket, objec
defer uploadIDLock.RUnlock(lkctx)
if z.SinglePool() {
- return z.serverPools[0].GetMultipartInfo(ctx, bucket, object, uploadID, opts)
+ mi, err := z.serverPools[0].GetMultipartInfo(ctx, bucket, object, uploadID, opts)
+ mi.Object = decodeObjectNameForBucket(bucket, object)
+ return mi, err
}
for idx, pool := range z.serverPools {
@@ -2212,7 +2298,7 @@ func (z *erasureServerPools) GetMultipartInfo(ctx context.Context, bucket, objec
return MultipartInfo{}, InvalidUploadID{
Bucket: bucket,
- Object: object,
+ Object: userObject,
UploadID: uploadID,
}
}
@@ -2223,6 +2309,9 @@ func (z *erasureServerPools) ListObjectParts(ctx context.Context, bucket, object
return ListPartsInfo{}, err
}
+ userObject := object
+ object = encodeObjectNameForBucket(bucket, object)
+
uploadIDLock := z.NewNSLock(bucket, pathJoin(object, uploadID))
lkctx, err := uploadIDLock.GetRLock(ctx, globalOperationTimeout)
if err != nil {
@@ -2232,7 +2321,9 @@ func (z *erasureServerPools) ListObjectParts(ctx context.Context, bucket, object
defer uploadIDLock.RUnlock(lkctx)
if z.SinglePool() {
- return z.serverPools[0].ListObjectParts(ctx, bucket, object, uploadID, partNumberMarker, maxParts, opts)
+ result, err := z.serverPools[0].ListObjectParts(ctx, bucket, object, uploadID, partNumberMarker, maxParts, opts)
+ result.Object = userObject
+ return result, err
}
for idx, pool := range z.serverPools {
@@ -2241,6 +2332,7 @@ func (z *erasureServerPools) ListObjectParts(ctx context.Context, bucket, object
}
result, err := pool.ListObjectParts(ctx, bucket, object, uploadID, partNumberMarker, maxParts, opts)
if err == nil {
+ result.Object = userObject
return result, nil
}
if _, ok := err.(InvalidUploadID); ok {
@@ -2251,7 +2343,7 @@ func (z *erasureServerPools) ListObjectParts(ctx context.Context, bucket, object
return ListPartsInfo{}, InvalidUploadID{
Bucket: bucket,
- Object: object,
+ Object: userObject,
UploadID: uploadID,
}
}
@@ -2265,6 +2357,9 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o
return toObjectErr(err, bucket)
}
+ userObject := object
+ object = encodeObjectNameForBucket(bucket, object)
+
// Unlock cancels the derived lock context before this notification runs.
// Keep the request context so successful cancellation reaches peer caches.
defer func(ctx context.Context) {
@@ -2294,7 +2389,7 @@ func (z *erasureServerPools) AbortMultipartUpload(ctx context.Context, bucket, o
}
return InvalidUploadID{
Bucket: bucket,
- Object: object,
+ Object: userObject,
UploadID: uploadID,
}
}
@@ -2305,6 +2400,9 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
return objInfo, err
}
+ userObject := object
+ object = encodeObjectNameForBucket(bucket, object)
+
defer func() {
if err == nil {
z.mpCache.Delete(uploadID)
@@ -2314,7 +2412,7 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
if !z.SinglePool() {
if !opts.NoLock {
- lk := z.NewNSLock(bucket, encodeDirObject(object))
+ lk := z.NewNSLock(bucket, object)
lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
if err != nil {
return objInfo, err
@@ -2334,7 +2432,7 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
// Once satisfied, the callback is cleared so the set layer does not
// re-evaluate it against its local copy.
if opts.CheckPrecondFn != nil {
- copies, lerr := z.objectPoolInfos(ctx, bucket, encodeDirObject(object), ObjectOptions{
+ copies, lerr := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{
// Conditions always compare the logical current object,
// independently of the completion's destination version.
VersionID: "",
@@ -2365,7 +2463,7 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
opts.CheckPrecondFn = nil
}
if opts.DataMovement {
- resolve, err := z.checkDataMovementCopies(ctx, bucket, encodeDirObject(object), opts)
+ resolve, err := z.checkDataMovementCopies(ctx, bucket, object, opts)
if err != nil {
return ObjectInfo{}, err
}
@@ -2384,7 +2482,11 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
defer uploadIDLock.Unlock(wlkctx)
if z.SinglePool() {
- return z.serverPools[0].CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
+ oi, err := z.serverPools[0].CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
+ if err == nil {
+ oi.Name = decodeObjectNameForBucket(bucket, object)
+ }
+ return oi, err
}
if opts.ReplicaLockReconcile || opts.DataMovement {
@@ -2401,8 +2503,9 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
objInfo, err = pool.CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
if err == nil {
if opts.ReplicaLockReconcile && !opts.DataMovement {
- err = z.retireReplicaCopies(ctx, bucket, encodeDirObject(object), idx, objInfo)
+ err = z.retireReplicaCopies(ctx, bucket, object, idx, objInfo)
}
+ objInfo.Name = decodeObjectNameForBucket(bucket, object)
return objInfo, err
}
if _, ok := err.(InvalidUploadID); ok {
@@ -2414,7 +2517,7 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
return objInfo, InvalidUploadID{
Bucket: bucket,
- Object: object,
+ Object: userObject,
UploadID: uploadID,
}
}
@@ -2927,7 +3030,8 @@ func (z *erasureServerPools) HealObjects(ctx context.Context, bucket, prefix str
}
func (z *erasureServerPools) HealObject(ctx context.Context, bucket, object, versionID string, opts madmin.HealOpts) (madmin.HealResultItem, error) {
- object = encodeDirObject(object)
+ userObject := object
+ object = encodeObjectNameForBucket(bucket, object)
errs := make([]error, len(z.serverPools))
results := make([]madmin.HealResultItem, len(z.serverPools))
@@ -2940,7 +3044,7 @@ func (z *erasureServerPools) HealObject(ctx context.Context, bucket, object, ver
go func(idx int, pool *erasureSets) {
defer wg.Done()
result, err := z.healObjectInPool(ctx, pool.getHashedSet(object), bucket, object, versionID, opts)
- result.Object = decodeDirObject(result.Object)
+ result.Object = decodeObjectName(result.Object)
errs[idx] = err
results[idx] = result
}(idx, pool)
@@ -2964,7 +3068,7 @@ func (z *erasureServerPools) HealObject(ctx context.Context, bucket, object, ver
hr := madmin.HealResultItem{
Type: madmin.HealItemObject,
Bucket: bucket,
- Object: object,
+ Object: userObject,
VersionID: versionID,
}
@@ -2972,13 +3076,13 @@ func (z *erasureServerPools) HealObject(ctx context.Context, bucket, object, ver
if versionID != "" {
return hr, VersionNotFound{
Bucket: bucket,
- Object: object,
+ Object: userObject,
VersionID: versionID,
}
}
return hr, ObjectNotFound{
Bucket: bucket,
- Object: object,
+ Object: userObject,
}
}
@@ -3192,9 +3296,13 @@ func (z *erasureServerPools) Health(ctx context.Context, opts HealthOptions) Hea
// PutObjectMetadata - replace or add tags to an existing object
func (z *erasureServerPools) PutObjectMetadata(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
- object = encodeDirObject(object)
+ object = encodeObjectNameForBucket(bucket, object)
if z.SinglePool() {
- return z.serverPools[0].PutObjectMetadata(ctx, bucket, object, opts)
+ oi, err := z.serverPools[0].PutObjectMetadata(ctx, bucket, object, opts)
+ if err == nil {
+ oi.Name = decodeObjectNameForBucket(bucket, object)
+ }
+ return oi, err
}
if !opts.NoLock {
@@ -3208,14 +3316,22 @@ func (z *erasureServerPools) PutObjectMetadata(ctx context.Context, bucket, obje
defer lk.Unlock(lkctx)
}
- return z.updatePoolMetadata(ctx, bucket, object, opts)
+ oi, err := z.updatePoolMetadata(ctx, bucket, object, opts)
+ if err == nil {
+ oi.Name = decodeObjectNameForBucket(bucket, object)
+ }
+ return oi, err
}
// PutObjectTags - replace or add tags to an existing object
func (z *erasureServerPools) PutObjectTags(ctx context.Context, bucket, object string, tags string, opts ObjectOptions) (ObjectInfo, error) {
- object = encodeDirObject(object)
+ object = encodeObjectNameForBucket(bucket, object)
if z.SinglePool() {
- return z.serverPools[0].PutObjectTags(ctx, bucket, object, tags, opts)
+ oi, err := z.serverPools[0].PutObjectTags(ctx, bucket, object, tags, opts)
+ if err == nil {
+ oi.Name = decodeObjectNameForBucket(bucket, object)
+ }
+ return oi, err
}
if !opts.NoLock {
@@ -3257,6 +3373,7 @@ func (z *erasureServerPools) PutObjectTags(ctx context.Context, bucket, object s
primary = oi
}
}
+ primary.Name = decodeObjectNameForBucket(bucket, object)
return primary, nil
}
@@ -3267,11 +3384,12 @@ func (z *erasureServerPools) DeleteObjectTags(ctx context.Context, bucket, objec
// GetObjectTags - get object tags from an existing object
func (z *erasureServerPools) GetObjectTags(ctx context.Context, bucket, object string, opts ObjectOptions) (*tags.Tags, error) {
- object = encodeDirObject(object)
if z.SinglePool() {
- return z.serverPools[0].GetObjectTags(ctx, bucket, object, opts)
+ return z.serverPools[0].GetObjectTags(ctx, bucket, encodeObjectNameForBucket(bucket, object), opts)
}
+ // GetObjectInfo takes the user-domain name and applies the storage-stack
+ // transformation itself.
oi, err := z.GetObjectInfo(ctx, bucket, object, opts)
if err != nil {
return nil, err
@@ -3282,7 +3400,7 @@ func (z *erasureServerPools) GetObjectTags(ctx context.Context, bucket, object s
// TransitionObject - transition object content to target tier.
func (z *erasureServerPools) TransitionObject(ctx context.Context, bucket, object string, opts ObjectOptions) error {
- object = encodeDirObject(object)
+ object = encodeObjectNameForBucket(bucket, object)
if z.SinglePool() {
return z.serverPools[0].TransitionObject(ctx, bucket, object, opts)
}
@@ -3311,7 +3429,7 @@ func (z *erasureServerPools) TransitionObject(ctx context.Context, bucket, objec
// RestoreTransitionedObject - restore transitioned object content locally on this cluster.
func (z *erasureServerPools) RestoreTransitionedObject(ctx context.Context, bucket, object string, opts ObjectOptions) error {
- object = encodeDirObject(object)
+ object = encodeObjectNameForBucket(bucket, object)
if z.SinglePool() {
return z.serverPools[0].RestoreTransitionedObject(ctx, bucket, object, opts)
}
@@ -3339,7 +3457,7 @@ func (z *erasureServerPools) RestoreTransitionedObject(ctx context.Context, buck
}
func (z *erasureServerPools) CheckAbandonedParts(ctx context.Context, bucket, object string, opts madmin.HealOpts) error {
- object = encodeDirObject(object)
+ object = encodeObjectNameForBucket(bucket, object)
if z.SinglePool() {
return z.serverPools[0].CheckAbandonedParts(ctx, bucket, object, opts)
}
@@ -3367,7 +3485,7 @@ func (z *erasureServerPools) CheckAbandonedParts(ctx context.Context, bucket, ob
// DecomTieredObject - moves tiered object to another pool during decommissioning.
func (z *erasureServerPools) DecomTieredObject(ctx context.Context, bucket, object string, fi FileInfo, opts ObjectOptions) error {
- object = encodeDirObject(object)
+ object = encodeObjectNameForBucket(bucket, object)
if z.SinglePool() {
return fmt.Errorf("error decommissioning %s/%s", bucket, object)
}
diff --git a/cmd/erasure-sets.go b/cmd/erasure-sets.go
index daf7d8e995c54..dd4b893029460 100644
--- a/cmd/erasure-sets.go
+++ b/cmd/erasure-sets.go
@@ -561,7 +561,7 @@ func auditObjectErasureSet(ctx context.Context, api, object string, set *erasure
}
op := auditObjectOp{
- Name: decodeDirObject(object),
+ Name: decodeObjectName(object),
Pool: set.poolIndex + 1,
Set: set.setIndex + 1,
}
diff --git a/cmd/format-erasure.go b/cmd/format-erasure.go
index ef730d7394a68..f43c4dcffedb1 100644
--- a/cmd/format-erasure.go
+++ b/cmd/format-erasure.go
@@ -121,6 +121,10 @@ type formatErasureV3 struct {
// Distribution algorithm represents the hashing algorithm
// to pick the right set index for an object.
DistributionAlgo string `json:"distributionAlgo"`
+ // ObjectNaming carries the cluster-wide object naming mode chosen
+ // when the cluster was first formatted: "safe" (the default when
+ // absent) or "extended". It cannot be changed afterwards.
+ ObjectNaming string `json:"objectNaming,omitempty"`
} `json:"xl"`
Info DiskInfo `json:"-"`
}
@@ -155,6 +159,7 @@ func newFormatErasureV3(numSets int, setLen int) *formatErasureV3 {
format.ID = mustGetUUID()
format.Erasure.Version = formatErasureVersionV3
format.Erasure.DistributionAlgo = formatErasureVersionV3DistributionAlgoV3
+ format.Erasure.ObjectNaming = normalizeObjectNamingMode(globalObjectNaming)
format.Erasure.Sets = make([][]string, numSets)
for i := range numSets {
@@ -287,8 +292,12 @@ func formatErasureMigrateV2ToV3(data []byte, export, version string) ([]byte, er
formatV3 := formatErasureV3{}
formatV3.Version = formatV2.Version
formatV3.Format = formatV2.Format
- formatV3.Erasure = formatV2.Erasure
formatV3.Erasure.Version = formatErasureVersionV3
+ formatV3.Erasure.This = formatV2.Erasure.This
+ formatV3.Erasure.Sets = formatV2.Erasure.Sets
+ formatV3.Erasure.DistributionAlgo = formatV2.Erasure.DistributionAlgo
+ // Migrated V2 clusters were written before extended object naming existed,
+ // so they keep the default (safe) mode; ObjectNaming stays empty.
return json.Marshal(formatV3)
}
@@ -426,11 +435,15 @@ func checkFormatErasureValue(formatErasure *formatErasureV3, disk StorageAPI) er
if formatErasure.Erasure.Version != formatErasureVersionV3 {
return fmt.Errorf("Unsupported Erasure backend format found [%s] on %s", formatErasure.Erasure.Version, disk)
}
+ if !isObjectNamingModeValid(formatErasure.Erasure.ObjectNaming) {
+ return fmt.Errorf("Unsupported object naming mode [%s] found on %s", formatErasure.Erasure.ObjectNaming, disk)
+ }
return nil
}
// Check all format values.
func checkFormatErasureValues(formats []*formatErasureV3, disks []StorageAPI, setDriveCount int) error {
+ expectedNaming := ""
for i, formatErasure := range formats {
if formatErasure == nil {
continue
@@ -438,6 +451,14 @@ func checkFormatErasureValues(formats []*formatErasureV3, disks []StorageAPI, se
if err := checkFormatErasureValue(formatErasure, disks[i]); err != nil {
return err
}
+ // Every drive in the deployment must use the same object naming mode;
+ // mixing them would make some keys invisible to part of the cluster.
+ naming := normalizeObjectNamingMode(formatErasure.Erasure.ObjectNaming)
+ if expectedNaming == "" {
+ expectedNaming = naming
+ } else if naming != expectedNaming {
+ return objectNamingModeMismatchError(naming, expectedNaming, disks[i])
+ }
if len(formats) != len(formatErasure.Erasure.Sets)*len(formatErasure.Erasure.Sets[0]) {
return fmt.Errorf("%s drive is already being used in another erasure deployment. (Number of drives specified: %d but the number of drives found in the %s drive's format.json: %d)",
disks[i], len(formats), humanize.Ordinal(i+1), len(formatErasure.Erasure.Sets)*len(formatErasure.Erasure.Sets[0]))
@@ -494,6 +515,11 @@ func formatErasureV3Check(reference *formatErasureV3, format *formatErasureV3) e
tmpFormat := format.Clone()
this := tmpFormat.Erasure.This
tmpFormat.Erasure.This = ""
+ if normalizeObjectNamingMode(reference.Erasure.ObjectNaming) != normalizeObjectNamingMode(format.Erasure.ObjectNaming) {
+ return fmt.Errorf("object naming mode mismatch, expected %q got %q: %w",
+ normalizeObjectNamingMode(reference.Erasure.ObjectNaming),
+ normalizeObjectNamingMode(format.Erasure.ObjectNaming), errInconsistentDisk)
+ }
if len(reference.Erasure.Sets) != len(format.Erasure.Sets) {
return fmt.Errorf("Expected number of sets %d, got %d", len(reference.Erasure.Sets), len(format.Erasure.Sets))
}
@@ -708,6 +734,7 @@ func newHealFormatSets(refFormat *formatErasureV3, setCount, setDriveCount int,
newFormats[i][j].Erasure.Sets = refFormat.Erasure.Sets
newFormats[i][j].Erasure.Version = refFormat.Erasure.Version
newFormats[i][j].Erasure.DistributionAlgo = refFormat.Erasure.DistributionAlgo
+ newFormats[i][j].Erasure.ObjectNaming = refFormat.Erasure.ObjectNaming
}
if format := formats[i*setDriveCount+j]; format != nil && (errs[i*setDriveCount+j] == nil) {
if format.Info.Endpoint != "" {
diff --git a/cmd/format-erasure_test.go b/cmd/format-erasure_test.go
index 732f6728e01e6..fe189381bc949 100644
--- a/cmd/format-erasure_test.go
+++ b/cmd/format-erasure_test.go
@@ -215,6 +215,7 @@ func TestCheckFormatErasureValue(t *testing.T) {
This string `json:"this"`
Sets [][]string `json:"sets"`
DistributionAlgo string `json:"distributionAlgo"`
+ ObjectNaming string `json:"objectNaming,omitempty"`
}{
Version: "2",
},
@@ -233,6 +234,7 @@ func TestCheckFormatErasureValue(t *testing.T) {
This string `json:"this"`
Sets [][]string `json:"sets"`
DistributionAlgo string `json:"distributionAlgo"`
+ ObjectNaming string `json:"objectNaming,omitempty"`
}{
Version: "2",
},
@@ -251,6 +253,7 @@ func TestCheckFormatErasureValue(t *testing.T) {
This string `json:"this"`
Sets [][]string `json:"sets"`
DistributionAlgo string `json:"distributionAlgo"`
+ ObjectNaming string `json:"objectNaming,omitempty"`
}{
Version: "0",
},
diff --git a/cmd/global-heal.go b/cmd/global-heal.go
index ca78df8182f2d..249ab567a8ea6 100644
--- a/cmd/global-heal.go
+++ b/cmd/global-heal.go
@@ -420,7 +420,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
}
// erasureObjects layer needs object names to be encoded
- encodedEntryName := encodeDirObject(entry.name)
+ encodedEntryName := encodeObjectNameForBucket(bucket, entry.name)
var result healEntryResult
fivs, err := entry.fileInfoVersions(bucket)
diff --git a/cmd/globals.go b/cmd/globals.go
index e7129450873d4..c9c096d877769 100644
--- a/cmd/globals.go
+++ b/cmd/globals.go
@@ -168,6 +168,10 @@ type serverCtxt struct {
SendBufSize, RecvBufSize int
CrossDomainXML string
+
+ // ObjectNaming is the requested on-disk object naming mode
+ // ("safe" or "extended"); empty means unspecified.
+ ObjectNaming string
// The layout of disks as interpreted
Layout disksLayout
}
@@ -387,6 +391,17 @@ var (
// If writes to FS backend should be O_SYNC.
globalFSOSync bool
+ // Requested object naming mode, set from --object-naming or
+ // MINIO_OBJECT_NAMING before the object layer is initialized. Empty means
+ // "not specified": a fresh cluster then defaults to safe mode, and an
+ // existing cluster adopts the mode recorded in its format.
+ globalObjectNaming = ""
+
+ // Resolved cluster-wide object naming mode. Set once the loaded formats
+ // have been verified to agree; read-only afterwards. Used by the listing
+ // merge layer, which has no disk context.
+ globalObjectNamingExtended bool
+
globalProxyEndpoints []ProxyEndpoint
globalInternodeTransport http.RoundTripper
diff --git a/cmd/metacache-entries.go b/cmd/metacache-entries.go
index ee45ac492e624..bfd97339ce3f1 100644
--- a/cmd/metacache-entries.go
+++ b/cmd/metacache-entries.go
@@ -810,7 +810,16 @@ func mergeEntryChannelsWithResolver(ctx context.Context, in []chan metaCacheEntr
bestIdx = otherIdx
continue
}
- if path.Clean(best.name) == path.Clean(other.name) {
+ // In extended naming mode an object `name` and the directory of the
+ // prefix `name/` are distinct keys that must both be listed, so the
+ // collision test is exact-name equality. In safe mode a trailing
+ // slash is the only difference that can occur (an object `name/`
+ // versus the directory `name/`), and path.Clean catches it.
+ sameName := best.name == other.name
+ if !globalObjectNamingExtended {
+ sameName = path.Clean(best.name) == path.Clean(other.name)
+ }
+ if sameName {
// We may be in a situation where we have a directory and an object with the same name.
// In that case we will drop the directory entry.
// This should however not be confused with an object with a trailing slash.
diff --git a/cmd/metacache-server-pool.go b/cmd/metacache-server-pool.go
index 2889e9a96c9eb..91697de97322a 100644
--- a/cmd/metacache-server-pool.go
+++ b/cmd/metacache-server-pool.go
@@ -315,7 +315,7 @@ func (z *erasureServerPools) resolveListingGroup(ctx context.Context, bucket str
}
}()
name := group[0].name
- object := encodeDirObject(name)
+ object := encodeObjectNameForBucket(bucket, name)
if z.SinglePool() {
target := z.serverPools[0].getHashedSet(object)
for _, entry := range group {
diff --git a/cmd/metacache-walk.go b/cmd/metacache-walk.go
index fa949d9ab4f3e..200b21e0b18a2 100644
--- a/cmd/metacache-walk.go
+++ b/cmd/metacache-walk.go
@@ -119,6 +119,16 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
return nil
}
+ // Extended object naming keeps objects and their own prefixes in distinct
+ // directories, so the walk emits both and re-sorts each directory level in
+ // the user domain (on disk the terminal marker sorts after '/').
+ // Meta buckets keep their raw layout and the standard walk.
+ if globalObjectNamingExtended && !skipAccessChecks(opts.Bucket) {
+ return s.walkDirExtended(ctx, opts, volumeDir, legacyFS, send, func() bool {
+ return opts.Limit > 0 && objsReturned >= opts.Limit
+ })
+ }
+
// Fast exit track to check if we are listing an object with
// a trailing slash, this will avoid to list the object content.
if HasSuffix(opts.BaseDir, SlashSeparator) {
@@ -406,6 +416,222 @@ func (s *xlStorage) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writ
return scanDir(opts.BaseDir)
}
+// walkDirExtended implements WalkDir for the extended object naming mode.
+//
+// On disk a regular object lives in a directory whose final component carries
+// the terminal marker (`obj1` -> `obj1◊/xl.meta`), prefix directories are plain
+// escaped names (`obj1/` stays `obj1/`) and directory objects keep the legacy
+// `__XLDIR__` suffix. The terminal marker sorts after '/' in byte order, so
+// disk order is not S3 key order: every level is decoded, sorted in the user
+// domain and emitted with recursion inline, which keeps the whole stream
+// ascending. Objects and the prefix of the same name therefore both appear,
+// which is the point of the mode.
+func (s *xlStorage) walkDirExtended(ctx context.Context, opts WalkDirOptions, volumeDir string, legacyFS bool, send func(metaCacheEntry) error, limitReached func() bool) error {
+ userBase := opts.BaseDir
+
+ // Translate BaseDir into the disk domain.
+ var (
+ diskBase string
+ objectDir bool
+ )
+ switch {
+ case userBase == "":
+ case strings.HasSuffix(userBase, SlashSeparator):
+ diskBase = escapeExtendedNaming(userBase)
+ default:
+ // Callers that address one object (healing, heal scans) pass the key
+ // without a trailing slash. Prefer the object directory when it
+ // exists, mirroring the directory-oriented safe-mode walk.
+ candidate := encodeExtendedObjectPath(userBase)
+ if _, sterr := Lstat(pathJoin(volumeDir, candidate)); sterr == nil {
+ diskBase = candidate
+ objectDir = true
+ } else {
+ diskBase = escapeExtendedNaming(userBase)
+ }
+ }
+
+ // A trailing-slash prefix may also be stored as a directory object; emit
+ // it first. Both representations can coexist with children of their own.
+ if !objectDir && strings.HasSuffix(userBase, SlashSeparator) {
+ dirObj := escapeExtendedNaming(strings.TrimSuffix(userBase, SlashSeparator)) + globalDirSuffix
+ if s.walkReadMu != nil {
+ s.walkReadMu.Lock()
+ }
+ metadata, merr := s.readMetadata(ctx, pathJoin(volumeDir, dirObj, xlStorageFormatFile))
+ if s.walkReadMu != nil {
+ s.walkReadMu.Unlock()
+ }
+ diskHealthCheckOK(ctx, merr)
+ if merr == nil && !limitReached() {
+ if err := send(metaCacheEntry{name: userBase, metadata: metadata}); err != nil {
+ return err
+ }
+ }
+ }
+
+ // The base itself is an object directory: emit it and stop, matching the
+ // safe-mode behavior for a directory-oriented base.
+ if objectDir {
+ if limitReached() {
+ return nil
+ }
+ if s.walkReadMu != nil {
+ s.walkReadMu.Lock()
+ }
+ metadata, merr := s.readMetadata(ctx, pathJoin(volumeDir, diskBase, xlStorageFormatFile))
+ if s.walkReadMu != nil {
+ s.walkReadMu.Unlock()
+ }
+ diskHealthCheckOK(ctx, merr)
+ if merr != nil {
+ if osIsNotExist(merr) || isSysErrIsDir(merr) {
+ return nil
+ }
+ return merr
+ }
+ return send(metaCacheEntry{name: userBase, metadata: metadata})
+ }
+
+ // A non-slash base that resolved to a prefix directory still prefixes the
+ // emitted names with "dir/" to match pathJoin semantics.
+ userRoot := userBase
+ if userRoot != "" && !strings.HasSuffix(userRoot, SlashSeparator) {
+ userRoot += SlashSeparator
+ }
+
+ type extendedWalkEntry struct {
+ disk string // disk-relative entry name without the trailing '/'
+ user string // user-domain name; prefixes end with '/'
+ isObject bool
+ }
+
+ var scanDir func(userCurrent, diskCurrent string) error
+ scanDir = func(userCurrent, diskCurrent string) error {
+ if s.walkMu != nil {
+ s.walkMu.Lock()
+ }
+ entries, lerr := s.ListDir(ctx, "", opts.Bucket, diskCurrent, -1)
+ if s.walkMu != nil {
+ s.walkMu.Unlock()
+ }
+ if lerr != nil {
+ if lerr != errVolumeNotFound && lerr != errFileNotFound {
+ internalLogOnceIf(ctx, lerr, "metacache-walk-scan-dir")
+ }
+ if opts.ReportNotFound && lerr == errFileNotFound && diskCurrent == diskBase {
+ lerr = errFileNotFound
+ } else {
+ lerr = nil
+ }
+ diskHealthCheckOK(ctx, lerr)
+ return lerr
+ }
+ diskHealthCheckOK(ctx, nil)
+
+ items := make([]extendedWalkEntry, 0, len(entries))
+ for _, entry := range entries {
+ if !strings.HasSuffix(entry, SlashSeparator) {
+ // Only directories represent objects or prefixes in this mode.
+ continue
+ }
+ diskName := strings.TrimSuffix(entry, SlashSeparator)
+ switch {
+ case isExtendedObjectDirName(diskName):
+ items = append(items, extendedWalkEntry{
+ disk: diskName,
+ user: decodeExtendedObjectPath(diskName),
+ isObject: true,
+ })
+ case strings.HasSuffix(diskName, globalDirSuffix):
+ stem := strings.TrimSuffix(diskName, globalDirSuffix)
+ items = append(items, extendedWalkEntry{
+ disk: diskName,
+ user: unescapeExtendedNaming(stem) + SlashSeparator,
+ isObject: true,
+ })
+ default:
+ items = append(items, extendedWalkEntry{
+ disk: diskName,
+ user: decodeExtendedObjectPath(diskName) + SlashSeparator,
+ })
+ }
+ }
+
+ // Emit in S3 key order. For the same name a directory object precedes
+ // the prefix directory, matching the merge layer's object-over-directory
+ // rule.
+ sort.SliceStable(items, func(i, j int) bool {
+ if items[i].user != items[j].user {
+ return items[i].user < items[j].user
+ }
+ return items[i].isObject && !items[j].isObject
+ })
+
+ firstLevel := diskCurrent == diskBase
+ for _, item := range items {
+ if limitReached() {
+ return nil
+ }
+ if contextCanceled(ctx) {
+ return ctx.Err()
+ }
+ if firstLevel && len(opts.FilterPrefix) > 0 && !strings.HasPrefix(item.user, opts.FilterPrefix) {
+ continue
+ }
+ // Conservative forwarding: keep everything that may still be at or
+ // beyond the marker; Marker filtering happens downstream.
+ if forward := opts.ForwardTo; len(forward) > 0 && item.user < forward && !strings.HasPrefix(forward, item.user) {
+ continue
+ }
+
+ diskPath := pathJoin(volumeDir, diskCurrent, item.disk)
+ if !item.isObject {
+ if isDirEmpty(diskPath, legacyFS) {
+ // Emptied concurrently; the safe walk skips such directories too.
+ continue
+ }
+ name := userCurrent + item.user
+ if err := send(metaCacheEntry{name: name}); err != nil {
+ return err
+ }
+ if opts.Recursive {
+ if err := scanDir(name, item.disk+SlashSeparator); err != nil {
+ if errors.Is(err, context.Canceled) {
+ return err
+ }
+ internalLogIf(ctx, err)
+ }
+ }
+ continue
+ }
+
+ if s.walkReadMu != nil {
+ s.walkReadMu.Lock()
+ }
+ metadata, merr := s.readMetadata(ctx, pathJoin(diskPath, xlStorageFormatFile))
+ if s.walkReadMu != nil {
+ s.walkReadMu.Unlock()
+ }
+ diskHealthCheckOK(ctx, merr)
+ if merr != nil {
+ // A marker-named directory without metadata is a torn write;
+ // skip it the way the safe walk skips an unreadable object.
+ if osIsNotExist(merr) || isSysErrIsDir(merr) {
+ continue
+ }
+ return merr
+ }
+ if err := send(metaCacheEntry{name: userCurrent + item.user, metadata: metadata}); err != nil {
+ return err
+ }
+ }
+ return nil
+ }
+
+ return scanDir(userRoot, diskBase)
+}
+
func (p *xlStorageDiskIDCheck) WalkDir(ctx context.Context, opts WalkDirOptions, wr io.Writer) (err error) {
if err := p.checkID(opts.DiskID); err != nil {
return err
diff --git a/cmd/object-naming.go b/cmd/object-naming.go
new file mode 100644
index 0000000000000..fb983f16926f1
--- /dev/null
+++ b/cmd/object-naming.go
@@ -0,0 +1,274 @@
+// Copyright (c) 2015-2026 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package cmd
+
+import (
+ "fmt"
+ "strings"
+)
+
+// Object naming modes. The mode is chosen when a cluster is formatted for the
+// first time and can never be changed afterwards; it is persisted in
+// format.json on every drive.
+const (
+ objectNamingSafe = "safe"
+ objectNamingExtended = "extended"
+)
+
+// Unicode markers used by the extended object naming mode.
+//
+// A user key is stored under an escaped form and, for regular objects, the
+// final path component carries the terminal marker. Because the marker and
+// every escape character are escaped inside user keys, an unescaped terminal
+// marker on disk can only have been appended by the server. This restores the
+// invariant that a directory holding an `xl.meta` never doubles as the parent
+// directory of another key, so objects and their own prefixes can coexist:
+//
+// obj1 -> /obj1◊/xl.meta
+// obj1/obj2 -> /obj1/obj2◊/xl.meta
+// obj1/ -> /obj1__XLDIR__/xl.meta (directory objects keep __XLDIR__)
+const (
+ objectNamingEscape = '␣' // ␣ escape character
+ objectNamingTerminal = '◊' // ◊ terminal marker for object names
+ objectNamingDirSuffix = globalDirSuffix
+)
+
+// extendedNamingEscapes maps reserved user characters to the digit stored
+// after the escape character.
+var extendedNamingEscapes = map[rune]rune{
+ '␣': '0', // ␣ escape character itself
+ '⫽': '1', // ⫽ extra slash
+ '◖': '2', // ◖ leading slash
+ '◗': '3', // ◗ trailing slash
+ '◊': '4', // ◊ terminal marker
+ '◉': '5', // ◉
+ '◈': '6', // ◈
+ '◇': '7', // ◇
+ '◆': '8', // ◆
+ '⬡': '9', // ⬡
+}
+
+// extendedNamingUnescapes is the reverse of extendedNamingEscapes.
+var extendedNamingUnescapes = func() map[rune]rune {
+ m := make(map[rune]rune, len(extendedNamingEscapes))
+ for r, d := range extendedNamingEscapes {
+ m[d] = r
+ }
+ return m
+}()
+
+// isObjectNamingModeValid reports whether mode names a supported object naming
+// mode. The empty string selects the default (safe) mode.
+func isObjectNamingModeValid(mode string) bool {
+ switch mode {
+ case "", objectNamingSafe, objectNamingExtended:
+ return true
+ }
+ return false
+}
+
+// normalizeObjectNamingMode returns the canonical mode string for a value
+// passed through the --object-naming flag or MINIO_OBJECT_NAMING.
+func normalizeObjectNamingMode(mode string) string {
+ if mode == "" {
+ return objectNamingSafe
+ }
+ return mode
+}
+
+// escapeExtendedNaming escapes every reserved character in s, leaving all
+// other characters (including '/') untouched.
+func escapeExtendedNaming(s string) string {
+ // Fast path: most keys contain no reserved characters.
+ idx := strings.IndexFunc(s, func(r rune) bool {
+ _, ok := extendedNamingEscapes[r]
+ return ok
+ })
+ if idx < 0 {
+ return s
+ }
+ var b strings.Builder
+ b.Grow(len(s) + 8)
+ b.WriteString(s[:idx])
+ for _, r := range s[idx:] {
+ if d, ok := extendedNamingEscapes[r]; ok {
+ b.WriteRune(objectNamingEscape)
+ b.WriteRune(d)
+ continue
+ }
+ b.WriteRune(r)
+ }
+ return b.String()
+}
+
+// unescapeExtendedNaming reverses escapeExtendedNaming. It never strips the
+// terminal marker; callers must do that first.
+func unescapeExtendedNaming(s string) string {
+ if !strings.ContainsRune(s, objectNamingEscape) {
+ return s
+ }
+ var b strings.Builder
+ b.Grow(len(s))
+ runes := []rune(s)
+ for i := 0; i < len(runes); i++ {
+ r := runes[i]
+ if r == objectNamingEscape && i+1 < len(runes) {
+ if orig, ok := extendedNamingUnescapes[runes[i+1]]; ok {
+ b.WriteRune(orig)
+ i++
+ continue
+ }
+ }
+ b.WriteRune(r)
+ }
+ return b.String()
+}
+
+// encodeExtendedObjectPath converts a user-domain object path to its on-disk
+// form.
+//
+// - Paths ending with the directory-object suffix have the suffix preserved
+// outside the escaped name and carry no terminal marker.
+// - Paths ending with '/' are treated as prefixes and are escaped only.
+// - All other paths are object names: escaped, with the terminal marker
+// appended to the final component.
+func encodeExtendedObjectPath(object string) string {
+ if strings.HasSuffix(object, objectNamingDirSuffix) {
+ stem := strings.TrimSuffix(object, objectNamingDirSuffix)
+ return escapeExtendedNaming(stem) + objectNamingDirSuffix
+ }
+ if strings.HasSuffix(object, SlashSeparator) {
+ return escapeExtendedNaming(object)
+ }
+ return escapeExtendedNaming(object) + string(objectNamingTerminal)
+}
+
+// decodeExtendedObjectPath reverses encodeExtendedObjectPath for a
+// disk-relative name. Directory-object names keep their __XLDIR__ suffix, which
+// callers pass through decodeDirObject as before.
+func decodeExtendedObjectPath(diskName string) string {
+ if strings.HasSuffix(diskName, objectNamingDirSuffix) {
+ stem := strings.TrimSuffix(diskName, objectNamingDirSuffix)
+ return unescapeExtendedNaming(stem) + objectNamingDirSuffix
+ }
+ if strings.HasSuffix(diskName, string(objectNamingTerminal)) {
+ diskName = strings.TrimSuffix(diskName, string(objectNamingTerminal))
+ }
+ return unescapeExtendedNaming(diskName)
+}
+
+// isExtendedObjectDirName reports whether a disk entry (without trailing '/')
+// is a regular object directory, i.e. its final component carries the
+// terminal marker.
+func isExtendedObjectDirName(name string) bool {
+ return strings.HasSuffix(name, string(objectNamingTerminal))
+}
+
+// encodeObjectName converts a user-domain object name into the form used
+// inside the storage stack (set/erasure/storage layers). In safe mode this is
+// the historical encodeDirObject (directory objects only). In extended mode
+// every object is escaped and regular objects carry the terminal marker, so a
+// directory holding an xl.meta can never double as another key's parent.
+//
+// The already-normalized directory-object form ("name__XLDIR__") is accepted
+// as well and maps to the same disk name, so callers that carry that form (the
+// scanner does, exactly as in safe mode) do not need to translate.
+func encodeObjectName(object string) string {
+ if !globalObjectNamingExtended {
+ return encodeDirObject(object)
+ }
+ if strings.HasSuffix(object, SlashSeparator) {
+ // Directory objects keep the legacy __XLDIR__ representation. The
+ // suffix itself is preserved outside the escaped name.
+ return escapeExtendedNaming(strings.TrimSuffix(object, SlashSeparator)) + objectNamingDirSuffix
+ }
+ if strings.HasSuffix(object, globalDirSuffix) {
+ stem := strings.TrimSuffix(object, globalDirSuffix)
+ return escapeExtendedNaming(stem) + objectNamingDirSuffix
+ }
+ return escapeExtendedNaming(object) + string(objectNamingTerminal)
+}
+
+// decodeObjectName reverses encodeObjectName for a stack-domain name.
+func decodeObjectName(object string) string {
+ if !globalObjectNamingExtended {
+ return decodeDirObject(object)
+ }
+ if strings.HasSuffix(object, objectNamingDirSuffix) {
+ return unescapeExtendedNaming(strings.TrimSuffix(object, objectNamingDirSuffix)) + SlashSeparator
+ }
+ if strings.HasSuffix(object, string(objectNamingTerminal)) {
+ return unescapeExtendedNaming(strings.TrimSuffix(object, string(objectNamingTerminal)))
+ }
+ return unescapeExtendedNaming(object)
+}
+
+// encodeObjectNameForBucket applies the storage-stack name transformation for
+// an object key in bucket. Meta buckets (.minio.sys, its tmp and multipart
+// parts) keep their historical raw layout: those keys are generated by the
+// server and contain none of the reserved characters, and keeping them raw
+// lets internal tooling and older binaries read them.
+func encodeObjectNameForBucket(bucket, object string) string {
+ if isMinioMetaBucketName(bucket) {
+ return object
+ }
+ return encodeObjectName(object)
+}
+
+// decodeObjectNameForBucket reverses encodeObjectNameForBucket.
+func decodeObjectNameForBucket(bucket, object string) string {
+ if isMinioMetaBucketName(bucket) {
+ return object
+ }
+ return decodeObjectName(object)
+}
+
+// encodeObjectPrefixName converts a user-domain object prefix (a key that may
+// or may not end with '/') into the form used to address the prefix directory
+// tree in the storage stack. Unlike encodeObjectName it never produces a
+// directory-object suffix or a terminal marker, because a prefix is not an
+// object.
+func encodeObjectPrefixName(prefix string) string {
+ if !globalObjectNamingExtended {
+ return prefix
+ }
+ return escapeExtendedNaming(prefix)
+}
+
+// decodeExtendedDiskPath decodes every component of a disk-relative path.
+// Terminal markers are stripped (only object directory components carry one)
+// and reserved characters are unescaped. Internal names (xl.meta, part files,
+// UUID data directories) contain none of the reserved characters, so decoding
+// them is a no-op. Directory-object suffixes are kept, as in safe mode.
+func decodeExtendedDiskPath(rel string) string {
+ if !globalObjectNamingExtended {
+ return rel
+ }
+ parts := strings.Split(rel, SlashSeparator)
+ for i, p := range parts {
+ p = strings.TrimSuffix(p, string(objectNamingTerminal))
+ parts[i] = unescapeExtendedNaming(p)
+ }
+ return strings.Join(parts, SlashSeparator)
+}
+
+// objectNamingModeMismatchError builds the startup error for inconsistent
+// object naming modes across drives or pools.
+func objectNamingModeMismatchError(found, expected string, source fmt.Stringer) error {
+ return fmt.Errorf("drive %s has object naming mode %q, expected %q: object naming cannot be changed after the cluster was formatted", source, normalizeObjectNamingMode(found), normalizeObjectNamingMode(expected))
+}
diff --git a/cmd/object-naming_test.go b/cmd/object-naming_test.go
new file mode 100644
index 0000000000000..981002f5b0c97
--- /dev/null
+++ b/cmd/object-naming_test.go
@@ -0,0 +1,554 @@
+// Copyright (c) 2015-2026 MinIO, Inc.
+//
+// This file is part of MinIO Object Storage stack
+//
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU Affero General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+//
+// This program is distributed in the hope that it will be useful
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU Affero General Public License for more details.
+//
+// You should have received a copy of the GNU Affero General Public License
+// along with this program. If not, see .
+
+package cmd
+
+import (
+ "bytes"
+ "context"
+ "io"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/minio/madmin-go/v3"
+)
+
+// TestObjectNamingEncodeDecode pins the on-disk naming transformation,
+// including every example from the extended naming documentation.
+func TestObjectNamingEncodeDecode(t *testing.T) {
+ previous := globalObjectNamingExtended
+ globalObjectNamingExtended = true
+ defer func() { globalObjectNamingExtended = previous }()
+
+ // v1 keeps the S3 key restrictions unchanged: "//" and leading "/" are
+ // rejected at the API boundary before reaching the encoder, and a trailing
+ // slash still uses the legacy __XLDIR__ representation. The escape
+ // machinery exists to keep every legal key unambiguous on disk.
+ encodeCases := []struct{ in, want string }{
+ {"obj1", "obj1◊"},
+ {"obj1/obj2", "obj1/obj2◊"},
+ {"data◊file.txt", "data␣4file.txt◊"},
+ {"a␣b", "a␣0b◊"},
+ {"x⫽y", "x␣1y◊"},
+ {"q◖r", "q␣2r◊"},
+ {"q◗r", "q␣3r◊"},
+ {"r◉s", "r␣5s◊"},
+ {"r◈s", "r␣6s◊"},
+ {"r◇s", "r␣7s◊"},
+ {"r◆s", "r␣8s◊"},
+ {"r⬡s", "r␣9s◊"},
+ }
+ for _, tc := range encodeCases {
+ if got := encodeObjectName(tc.in); got != tc.want {
+ t.Errorf("encodeObjectName(%q) = %q, want %q", tc.in, got, tc.want)
+ }
+ if got := decodeObjectName(tc.want); got != tc.in {
+ t.Errorf("decodeObjectName(%q) = %q, want %q", tc.want, got, tc.in)
+ }
+ }
+
+ // A trailing-slash key is a directory object: it keeps the legacy
+ // __XLDIR__ form (the documented v1 choice) and round-trips.
+ if got := encodeObjectName("myobject/"); got != "myobject__XLDIR__" {
+ t.Errorf("directory object encoding = %q", got)
+ }
+ for _, name := range []string{"obj1/", "a␣b/", "x◊/"} {
+ enc := encodeObjectName(name)
+ if enc != escapeExtendedNaming(name[:len(name)-1])+globalDirSuffix {
+ t.Fatalf("directory object %q encoded as %q", name, enc)
+ }
+ if got := decodeObjectName(enc); got != name {
+ t.Errorf("decodeObjectName(%q) = %q, want %q", enc, got, name)
+ }
+ }
+
+ // The pre-normalized directory-object form is accepted too (the scanner
+ // passes it exactly as in safe mode).
+ if got := encodeObjectName("obj1__XLDIR__"); got != "obj1__XLDIR__" {
+ t.Errorf("encodeObjectName(pre-normalized dir object) = %q", got)
+ }
+
+ // Prefixes are escaped without a terminal marker.
+ if got := encodeObjectPrefixName("a◊/b/"); got != "a␣4/b/" {
+ t.Errorf("encodeObjectPrefixName = %q", got)
+ }
+
+ // Meta buckets are never transformed.
+ if got := encodeObjectNameForBucket(minioMetaBucket, "config/x.json"); got != "config/x.json" {
+ t.Errorf("meta bucket object was transformed: %q", got)
+ }
+
+ // Safe mode keeps the historical behavior.
+ globalObjectNamingExtended = false
+ if got := encodeObjectName("obj1"); got != "obj1" {
+ t.Errorf("safe mode encodeObjectName = %q", got)
+ }
+ if got := encodeObjectName("obj1/"); got != "obj1__XLDIR__" {
+ t.Errorf("safe mode directory object = %q", got)
+ }
+ if got := decodeObjectName("obj1__XLDIR__"); got != "obj1/" {
+ t.Errorf("safe mode decodeObjectName = %q", got)
+ }
+}
+
+// TestObjectNamingPrefixEscapeIsOrderPreserving verifies the per-character
+// escape keeps the relative order of ordinary (non-reserved) characters, which
+// the walk relies on for FilterPrefix and ForwardTo.
+func TestObjectNamingPrefixEscapeIsOrderPreserving(t *testing.T) {
+ previous := globalObjectNamingExtended
+ globalObjectNamingExtended = true
+ defer func() { globalObjectNamingExtended = previous }()
+
+ samples := []string{
+ "a", "ab", "a/b", "a/bc", "a0", "a b", "a-b", "a.b", "a~", "a中", "a z",
+ }
+ for i := range samples {
+ for j := range samples {
+ if samples[i] < samples[j] {
+ if escapeExtendedNaming(samples[i]) >= escapeExtendedNaming(samples[j]) {
+ t.Fatalf("escape reordered %q and %q", samples[i], samples[j])
+ }
+ }
+ if samples[i] == samples[j] && escapeExtendedNaming(samples[i]) != escapeExtendedNaming(samples[j]) {
+ t.Fatalf("escape not injective for %q", samples[i])
+ }
+ }
+ }
+}
+
+// newExtendedNamingPool starts an erasure pool formatted with the extended
+// object naming mode and returns it with a fresh bucket.
+func newExtendedNamingPool(t *testing.T, nDisks int, bucketOpts MakeBucketOptions) (*erasureServerPools, []string, string) {
+ t.Helper()
+
+ previousMode := globalObjectNaming
+ previousExtended := globalObjectNamingExtended
+ globalObjectNaming = objectNamingExtended
+ globalObjectNamingExtended = false
+ t.Cleanup(func() {
+ globalObjectNaming = previousMode
+ globalObjectNamingExtended = previousExtended
+ })
+
+ ctx, cancel := context.WithCancel(t.Context())
+ fsDirs, err := getRandomDisks(nDisks)
+ if err != nil {
+ cancel()
+ t.Fatal(err)
+ }
+ obj, _, err := initObjectLayer(ctx, mustGetPoolEndpoints(0, fsDirs...))
+ if err != nil {
+ cancel()
+ removeRoots(fsDirs)
+ t.Fatal(err)
+ }
+ z := obj.(*erasureServerPools)
+ previous := newObjectLayerFn()
+ setObjectLayer(z)
+ t.Cleanup(func() {
+ cancel()
+ z.Shutdown(context.Background())
+ removeRoots(fsDirs)
+ setObjectLayer(previous)
+ })
+ if !globalObjectNamingExtended {
+ t.Fatal("extended object naming was not enabled by the format")
+ }
+
+ bucket := getRandomBucketName()
+ if err := z.MakeBucket(t.Context(), bucket, bucketOpts); err != nil {
+ t.Fatal(err)
+ }
+ return z, fsDirs, bucket
+}
+
+func putTestObject(t *testing.T, z *erasureServerPools, bucket, name string) {
+ t.Helper()
+ body := []byte("body of " + name)
+ if _, err := z.PutObject(t.Context(), bucket, name,
+ mustGetPutObjReader(t, bytes.NewReader(body), int64(len(body)), "", ""), ObjectOptions{}); err != nil {
+ t.Fatalf("PUT %q: %v", name, err)
+ }
+}
+
+func readTestObject(t *testing.T, z *erasureServerPools, bucket, name string) string {
+ t.Helper()
+ gr, err := z.GetObjectNInfo(t.Context(), bucket, name, nil, nil, ObjectOptions{})
+ if err != nil {
+ t.Fatalf("GET %q: %v", name, err)
+ }
+ defer gr.Close()
+ buf, err := io.ReadAll(gr)
+ if err != nil {
+ t.Fatalf("read %q: %v", name, err)
+ }
+ return string(buf)
+}
+
+func listObjectNames(t *testing.T, z *erasureServerPools, bucket, prefix, delimiter string) (objs, prefixes []string) {
+ t.Helper()
+ loi, err := z.ListObjects(t.Context(), bucket, prefix, "", delimiter, 10000)
+ if err != nil {
+ t.Fatalf("LIST prefix=%q delimiter=%q: %v", prefix, delimiter, err)
+ }
+ for _, obj := range loi.Objects {
+ objs = append(objs, obj.Name)
+ }
+ return objs, loi.Prefixes
+}
+
+// TestExtendedNamingConflictingObjects is the end-to-end matrix from the
+// conflict analysis: an object and its own prefix must both be listed, read,
+// and deleted independently.
+func TestExtendedNamingConflictingObjects(t *testing.T) {
+ z, fsDirs, bucket := newExtendedNamingPool(t, 16, MakeBucketOptions{})
+ ctx := t.Context()
+
+ putTestObject(t, z, bucket, "obj1")
+ putTestObject(t, z, bucket, "obj1/obj2")
+
+ // The disk layout must keep object and prefix directories apart.
+ for _, want := range []string{
+ filepath.Join(bucket, "obj1◊", xlStorageFormatFile),
+ filepath.Join(bucket, "obj1", "obj2◊", xlStorageFormatFile),
+ } {
+ p := filepath.Join(fsDirs[0], want)
+ if _, err := os.Lstat(p); err != nil {
+ t.Fatalf("expected object directory %q: %v", p, err)
+ }
+ }
+
+ // Both keys read normally.
+ if got := readTestObject(t, z, bucket, "obj1"); got != "body of obj1" {
+ t.Fatalf("GET obj1 = %q", got)
+ }
+ if got := readTestObject(t, z, bucket, "obj1/obj2"); got != "body of obj1/obj2" {
+ t.Fatalf("GET obj1/obj2 = %q", got)
+ }
+
+ // Root, delimiter '/': object and prefix both appear.
+ objs, prefixes := listObjectNames(t, z, bucket, "", "/")
+ if len(objs) != 1 || objs[0] != "obj1" {
+ t.Fatalf("root delimiter list objects = %v", objs)
+ }
+ if len(prefixes) != 1 || prefixes[0] != "obj1/" {
+ t.Fatalf("root delimiter list prefixes = %v", prefixes)
+ }
+
+ // Recursive: both keys, in S3 order.
+ objs, prefixes = listObjectNames(t, z, bucket, "", "")
+ if len(prefixes) != 0 {
+ t.Fatalf("recursive list returned prefixes: %v", prefixes)
+ }
+ if len(objs) != 2 || objs[0] != "obj1" || objs[1] != "obj1/obj2" {
+ t.Fatalf("recursive list = %v", objs)
+ }
+
+ // Listing the prefix that is also an object name.
+ objs, _ = listObjectNames(t, z, bucket, "obj1/", "")
+ if len(objs) != 1 || objs[0] != "obj1/obj2" {
+ t.Fatalf("prefix obj1/ list = %v", objs)
+ }
+ objs, _ = listObjectNames(t, z, bucket, "obj1/obj2", "")
+ if len(objs) != 1 || objs[0] != "obj1/obj2" {
+ t.Fatalf("exact key list = %v", objs)
+ }
+
+ // Resuming from a marker that is the prefix itself must not lose or repeat
+ // the child.
+ loi, err := z.ListObjects(ctx, bucket, "", "obj1/", "", 1000)
+ if err != nil {
+ t.Fatalf("LIST after prefix marker: %v", err)
+ }
+ if len(loi.Objects) != 1 || loi.Objects[0].Name != "obj1/obj2" {
+ t.Fatalf("LIST after prefix marker = %v", loi.Objects)
+ }
+
+ // Pagination across the conflict point: one key per page, no duplicates,
+ // no losses.
+ var paged []string
+ marker := ""
+ for {
+ loi, err := z.ListObjects(ctx, bucket, "", marker, "", 1)
+ if err != nil {
+ t.Fatalf("paged LIST: %v", err)
+ }
+ for _, obj := range loi.Objects {
+ if len(paged) > 0 && paged[len(paged)-1] >= obj.Name {
+ t.Fatalf("paged list out of order: %v then %q", paged, obj.Name)
+ }
+ paged = append(paged, obj.Name)
+ }
+ if !loi.IsTruncated {
+ break
+ }
+ marker = loi.NextMarker
+ if len(paged) > 10 {
+ t.Fatal("paged list did not terminate")
+ }
+ }
+ if len(paged) != 2 || paged[0] != "obj1" || paged[1] != "obj1/obj2" {
+ t.Fatalf("paged list = %v", paged)
+ }
+
+ // Deleting the parent object must not touch the child.
+ if _, err := z.DeleteObject(ctx, bucket, "obj1", ObjectOptions{}); err != nil {
+ t.Fatalf("DELETE obj1: %v", err)
+ }
+ if _, err := z.GetObjectInfo(ctx, bucket, "obj1", ObjectOptions{}); !isErrObjectNotFound(err) {
+ t.Fatalf("obj1 still readable after delete: %v", err)
+ }
+ if got := readTestObject(t, z, bucket, "obj1/obj2"); got != "body of obj1/obj2" {
+ t.Fatalf("child damaged by parent delete: %q", got)
+ }
+ objs, prefixes = listObjectNames(t, z, bucket, "", "/")
+ if len(objs) != 0 || len(prefixes) != 1 || prefixes[0] != "obj1/" {
+ t.Fatalf("after parent delete: objects=%v prefixes=%v", objs, prefixes)
+ }
+
+ // Healing addresses both keys through the same transformation.
+ if _, err := z.HealObject(ctx, bucket, "obj1/obj2", "", madmin.HealOpts{ScanMode: madmin.HealNormalScan}); err != nil {
+ t.Fatalf("heal child: %v", err)
+ }
+
+ // Overwriting the parent again keeps the child intact.
+ putTestObject(t, z, bucket, "obj1")
+ if got := readTestObject(t, z, bucket, "obj1/obj2"); got != "body of obj1/obj2" {
+ t.Fatalf("child damaged by parent overwrite: %q", got)
+ }
+ objs, _ = listObjectNames(t, z, bucket, "", "")
+ if len(objs) != 2 || objs[0] != "obj1" || objs[1] != "obj1/obj2" {
+ t.Fatalf("recursive list after overwrite = %v", objs)
+ }
+}
+
+// TestExtendedNamingVersioned runs the same conflict through a versioned
+// bucket, including delete markers and ListObjectVersions.
+func TestExtendedNamingVersioned(t *testing.T) {
+ z, _, bucket := newExtendedNamingPool(t, 16, MakeBucketOptions{VersioningEnabled: true})
+ ctx := t.Context()
+
+ putTestObject(t, z, bucket, "obj1")
+ putTestObject(t, z, bucket, "obj1/obj2")
+
+ loi, err := z.ListObjectVersions(ctx, bucket, "", "", "", "", 10000)
+ if err != nil {
+ t.Fatalf("ListObjectVersions: %v", err)
+ }
+ names := map[string]bool{}
+ for _, obj := range loi.Objects {
+ names[obj.Name] = true
+ }
+ if !names["obj1"] || !names["obj1/obj2"] {
+ t.Fatalf("ListObjectVersions missing conflict keys: %v", loi.Objects)
+ }
+
+ // A delete marker on the parent hides only the parent.
+ if _, err := z.DeleteObject(ctx, bucket, "obj1", ObjectOptions{}); err != nil {
+ t.Fatalf("DELETE obj1: %v", err)
+ }
+ objs, _ := listObjectNames(t, z, bucket, "", "")
+ if len(objs) != 1 || objs[0] != "obj1/obj2" {
+ t.Fatalf("recursive list after marker = %v", objs)
+ }
+ if got := readTestObject(t, z, bucket, "obj1/obj2"); got != "body of obj1/obj2" {
+ t.Fatalf("child unreadable after parent marker: %q", got)
+ }
+}
+
+// TestExtendedNamingDirObjectCoexist covers an object, its directory-object
+// twin and a child under the same prefix.
+func TestExtendedNamingDirObjectCoexist(t *testing.T) {
+ z, _, bucket := newExtendedNamingPool(t, 16, MakeBucketOptions{})
+
+ putTestObject(t, z, bucket, "X")
+ putTestObject(t, z, bucket, "X/")
+ putTestObject(t, z, bucket, "X/child")
+
+ objs, _ := listObjectNames(t, z, bucket, "", "")
+ want := []string{"X", "X/", "X/child"}
+ if len(objs) != len(want) {
+ t.Fatalf("recursive list = %v, want %v", objs, want)
+ }
+ for i := range want {
+ if objs[i] != want[i] {
+ t.Fatalf("recursive list = %v, want %v", objs, want)
+ }
+ }
+
+ if got := readTestObject(t, z, bucket, "X"); got != "body of X" {
+ t.Fatalf("GET X = %q", got)
+ }
+ if got := readTestObject(t, z, bucket, "X/"); got != "body of X/" {
+ t.Fatalf("GET X/ = %q", got)
+ }
+ if got := readTestObject(t, z, bucket, "X/child"); got != "body of X/child" {
+ t.Fatalf("GET X/child = %q", got)
+ }
+}
+
+// TestExtendedNamingEscapedKeyOrder lists keys that contain reserved
+// characters and prefixes, checking the stream stays in user-key order.
+func TestExtendedNamingEscapedKeyOrder(t *testing.T) {
+ z, _, bucket := newExtendedNamingPool(t, 16, MakeBucketOptions{})
+
+ keys := []string{"a", "a/b", "a0", "a◊b", "a␣c", "b"}
+ for _, k := range keys {
+ putTestObject(t, z, bucket, k)
+ }
+ // S3 orders keys by UTF-8 bytes: U+2423 (␣) sorts before U+25CA (◊).
+ objs, _ := listObjectNames(t, z, bucket, "", "")
+ want := []string{"a", "a/b", "a0", "a␣c", "a◊b", "b"}
+ if len(objs) != len(want) {
+ t.Fatalf("recursive list = %v, want %v", objs, want)
+ }
+ for i := range want {
+ if objs[i] != want[i] {
+ t.Fatalf("recursive list = %v, want %v", objs, want)
+ }
+ }
+ // Every key reads back under its own name.
+ for _, k := range keys {
+ if got := readTestObject(t, z, bucket, k); got != "body of "+k {
+ t.Fatalf("GET %q = %q", k, got)
+ }
+ }
+}
+
+// TestExtendedNamingMultipart completes a multipart upload onto a key that
+// conflicts with a prefix in the same bucket.
+func TestExtendedNamingMultipart(t *testing.T) {
+ z, _, bucket := newExtendedNamingPool(t, 16, MakeBucketOptions{})
+ ctx := t.Context()
+
+ putTestObject(t, z, bucket, "obj1/obj2")
+
+ res, err := z.NewMultipartUpload(ctx, bucket, "obj1", ObjectOptions{})
+ if err != nil {
+ t.Fatalf("NewMultipartUpload: %v", err)
+ }
+ data := bytes.Repeat([]byte("m"), 5<<20) // one part above the 5 MiB minimum
+ part, err := z.PutObjectPart(ctx, bucket, "obj1", res.UploadID, 1,
+ mustGetPutObjReader(t, bytes.NewReader(data), int64(len(data)), "", ""), ObjectOptions{})
+ if err != nil {
+ t.Fatalf("PutObjectPart: %v", err)
+ }
+ parts := []CompletePart{{ETag: part.ETag, PartNumber: part.PartNumber}}
+ if _, err := z.CompleteMultipartUpload(ctx, bucket, "obj1", res.UploadID, parts, ObjectOptions{}); err != nil {
+ t.Fatalf("CompleteMultipartUpload: %v", err)
+ }
+
+ gr, err := z.GetObjectNInfo(ctx, bucket, "obj1", nil, nil, ObjectOptions{})
+ if err != nil {
+ t.Fatalf("GET obj1: %v", err)
+ }
+ gr.Close()
+
+ putTestObject(t, z, bucket, "obj1/tail")
+ objs, _ := listObjectNames(t, z, bucket, "", "")
+ want := []string{"obj1", "obj1/obj2", "obj1/tail"}
+ if len(objs) != len(want) {
+ t.Fatalf("recursive list = %v, want %v", objs, want)
+ }
+ for i := range want {
+ if objs[i] != want[i] {
+ t.Fatalf("recursive list = %v, want %v", objs, want)
+ }
+ }
+}
+
+// TestExtendedNamingFormatPersistence verifies the mode is recorded on every
+// drive and that a conflicting request is rejected at startup.
+func TestExtendedNamingFormatPersistence(t *testing.T) {
+ z, fsDirs, _ := newExtendedNamingPool(t, 16, MakeBucketOptions{})
+ _ = z
+
+ data, err := os.ReadFile(filepath.Join(fsDirs[0], minioMetaBucket, formatConfigFile))
+ if err != nil {
+ t.Fatalf("read format.json: %v", err)
+ }
+ if !bytes.Contains(data, []byte(`"objectNaming":"extended"`)) {
+ t.Fatalf("format.json does not record extended naming: %s", data)
+ }
+
+ // Re-opening the same drives while requesting safe mode must fail.
+ globalObjectNaming = objectNamingSafe
+ ctx, cancel := context.WithCancel(t.Context())
+ defer cancel()
+ _, _, err = initObjectLayer(ctx, mustGetPoolEndpoints(0, fsDirs...))
+ if err == nil || !strings.Contains(err.Error(), "object naming") {
+ t.Fatalf("expected object naming mismatch error, got %v", err)
+ }
+}
+
+// TestSafeNamingConflictUnchanged locks in the documented behavior of the
+// default mode: the conflicting subtree stays invisible to LIST.
+func TestSafeNamingConflictUnchanged(t *testing.T) {
+ // The formats are written at first start, so a fresh cluster is required.
+ previousMode := globalObjectNaming
+ previousExtended := globalObjectNamingExtended
+ globalObjectNaming = ""
+ globalObjectNamingExtended = false
+ t.Cleanup(func() {
+ globalObjectNaming = previousMode
+ globalObjectNamingExtended = previousExtended
+ })
+
+ ctx, cancel := context.WithCancel(t.Context())
+ fsDirs, err := getRandomDisks(16)
+ if err != nil {
+ cancel()
+ t.Fatal(err)
+ }
+ obj, _, err := initObjectLayer(ctx, mustGetPoolEndpoints(0, fsDirs...))
+ if err != nil {
+ cancel()
+ removeRoots(fsDirs)
+ t.Fatal(err)
+ }
+ zs := obj.(*erasureServerPools)
+ previous := newObjectLayerFn()
+ setObjectLayer(zs)
+ t.Cleanup(func() {
+ cancel()
+ zs.Shutdown(context.Background())
+ removeRoots(fsDirs)
+ setObjectLayer(previous)
+ })
+ safeBucket := getRandomBucketName()
+ if err := zs.MakeBucket(ctx, safeBucket, MakeBucketOptions{}); err != nil {
+ t.Fatal(err)
+ }
+
+ putTestObject(t, zs, safeBucket, "obj1")
+ putTestObject(t, zs, safeBucket, "obj1/obj2")
+
+ objs, _ := listObjectNames(t, zs, safeBucket, "", "")
+ if len(objs) != 1 || objs[0] != "obj1" {
+ t.Fatalf("safe mode recursive list = %v, want [obj1]", objs)
+ }
+ objs, _ = listObjectNames(t, zs, safeBucket, "obj1/", "")
+ if len(objs) != 0 {
+ t.Fatalf("safe mode prefix obj1/ list = %v, want empty", objs)
+ }
+ // Exact-key reads still work in safe mode.
+ if got := readTestObject(t, zs, safeBucket, "obj1/obj2"); got != "body of obj1/obj2" {
+ t.Fatalf("safe mode GET obj1/obj2 = %q", got)
+ }
+}
diff --git a/cmd/server-main.go b/cmd/server-main.go
index c8a3a19ca276b..5f9c50a3f82d9 100644
--- a/cmd/server-main.go
+++ b/cmd/server-main.go
@@ -119,6 +119,11 @@ var ServerFlags = []cli.Flag{
Hidden: true,
EnvVar: "MINIO_INTERFACE",
},
+ cli.StringFlag{
+ Name: "object-naming",
+ Usage: "on-disk object naming mode chosen at first format: \"safe\" (default) or \"extended\"; cannot be changed after the cluster is formatted",
+ EnvVar: "MINIO_OBJECT_NAMING",
+ },
cli.DurationFlag{
Name: "dns-cache-ttl",
Usage: "custom DNS cache TTL",
diff --git a/docs/minio-limits.md b/docs/minio-limits.md
index ff7ab7ca916c3..80eb50214bf44 100644
--- a/docs/minio-limits.md
+++ b/docs/minio-limits.md
@@ -62,7 +62,7 @@ We found the following APIs to be redundant or less useful outside of AWS S3. If
Silo recommends using Linux operating system for production workloads.
-- Objects must not have conflicting objects as parent objects, applications using this behavior should change their behavior and use non-conflicting unique keys, for example situations such as following conflicting key patterns are not supported.
+- In the default (safe) object naming mode, objects must not have conflicting objects as parent objects. Applications using this behavior should change their behavior and use non-conflicting unique keys; for example, situations such as the following conflicting key patterns are not supported, and LIST operations at the collision return an empty result set.
```
PUT /a/b/1.txt
@@ -73,3 +73,24 @@ PUT /a/b
PUT /a/b
PUT /a/b/1.txt
```
+
+ A cluster formatted with `--object-naming=extended` (or
+ `MINIO_OBJECT_NAMING=extended`) supports these conflicting keys: an object
+ and a prefix of the same name (`obj1` and `obj1/obj2`) are stored in
+ physically separate directories, so both appear in LIST results and exact-key
+ reads, writes and deletes behave independently. Be aware of the following:
+
+ - The mode is chosen when the cluster is formatted for the first time, is
+ recorded in every drive's `format.json`, and cannot be changed afterwards.
+ Every node must use the same mode; drives or pools that disagree refuse to
+ start. Existing safe-mode clusters cannot be migrated in place.
+ - Object names on disk are escaped and carry a terminal marker (for example
+ `data◊file.txt` is stored as `data␣4file.txt◊`). Tools that read raw drive
+ contents, such as `mc admin inspect`, must address the encoded name.
+ - Escape sequences make some names slightly longer on disk. The 1024-byte
+ object name limit and the 255-byte per-segment limit still apply to the
+ encoded form.
+ - Replication and data movement between extended and safe mode clusters is
+ not supported.
+ - Keys ending with `__XLDIR__` remain reserved for directory objects, as in
+ safe mode.
diff --git a/internal/config/constants.go b/internal/config/constants.go
index bd00ae347a4dc..b58be41acc180 100644
--- a/internal/config/constants.go
+++ b/internal/config/constants.go
@@ -54,6 +54,10 @@ const (
EnvSiteName = "MINIO_SITE_NAME"
EnvSiteRegion = "MINIO_SITE_REGION"
+ // EnvObjectNaming selects the on-disk object naming mode at first format:
+ // "safe" (default) or "extended".
+ EnvObjectNaming = "MINIO_OBJECT_NAMING"
+
EnvMinIOSubnetLicense = "MINIO_SUBNET_LICENSE" // Deprecated Dec 2021
EnvMinIOSubnetAPIKey = "MINIO_SUBNET_API_KEY"
EnvMinIOSubnetProxy = "MINIO_SUBNET_PROXY"