Merge pull request #178 from pgsty/codex/multipool-correctness-20260911

fix(storage): reconcile multi-pool writes and conditional deletes

cmd/erasure-healing.go

@@ -1068,6 +1068,11 @@ func (er erasureObjects) HealObject(ctx context.Context, bucket, object, version
 		newReqInfo = logger.NewReqInfo("", "", globalDeploymentID(), "", "Heal", bucket, object)
 	}
 	healCtx := logger.SetReqInfo(GlobalContext, newReqInfo)
+	if opts.NoLock {
+		// The caller owns the namespace lock. Stop if that lock's context is
+		// canceled instead of continuing metadata writes after losing it.
+		healCtx = logger.SetReqInfo(ctx, newReqInfo)
+	}
 
 	// Healing directories handle it separately.
 	if HasSuffix(object, SlashSeparator) {

cmd/erasure-multipart.go

@@ -1155,17 +1155,10 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 		return oi, toObjectErr(err, bucket, object, uploadID)
 	}
 
-	// A trusted SSE-C replica completion re-orders the Object Lock carried in the
-	// upload metadata against the version it is about to replace, read on this
-	// erasure set under the write lock held above, so a hold or retention that
-	// reached the version after this upload was initiated is not rolled back at
-	// completion (issue #120). Scoped to SSE-C uploads, the only ones this issue
-	// routes through completion.
-	//
-	// Scope: correct for a single erasure set. A multi-pool deployment (duplicate
-	// versions across pools, ModTime ties, cross-pool lock authority) is out of
-	// scope and tracked in pgsty/silo#133.
-	if opts.ReplicaLockReconcile && crypto.SSEC.IsEncrypted(fi.Metadata) {
+	// Reconcile against the upload's persisted version, under the object lock.
+	// Multi-pool callers supply a resolver spanning all pools, including those
+	// draining their contents; the upload itself stays in its original pool.
+	if opts.ReplicaLockReconcile {
 		// A persisted upload records the null version as an empty VersionID; look
 		// it up as the null version so the reconcile reads the addressed version's
 		// stored lock, not the latest version's.
@@ -1173,7 +1166,11 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 		if lookupVersionID == "" {
 			lookupVersionID = nullVersionID
 		}
-		curr, gerr := er.getObjectInfo(ctx, bucket, object, ObjectOptions{
+		getObjectInfo := er.getObjectInfo
+		if opts.replicaObjectInfo != nil {
+			getObjectInfo = opts.replicaObjectInfo
+		}
+		curr, gerr := getObjectInfo(ctx, bucket, object, ObjectOptions{
 			VersionID:        lookupVersionID,
 			Versioned:        opts.Versioned,
 			VersionSuspended: opts.VersionSuspended,
@@ -1182,6 +1179,7 @@ func (er erasureObjects) CompleteMultipartUpload(ctx context.Context, bucket str
 		switch {
 		case gerr == nil:
 			reconcileStoredObjectLock(fi.Metadata, storedObjectLockState(curr.UserDefined))
+			reconcileStoredObjectTags(fi.Metadata, curr.UserDefined)
 		case isErrVersionNotFound(gerr) || isErrObjectNotFound(gerr):
 			// No existing version to order against: keep the upload's own accepted
 			// lock, including a pre-upgrade upload that persisted values without

cmd/erasure-object.go

@@ -133,6 +133,11 @@ func (er erasureObjects) CopyObject(ctx context.Context, srcBucket, srcObject, d
 		return fi.ToObjectInfo(srcBucket, srcObject, srcOpts.Versioned || srcOpts.VersionSuspended), toObjectErr(errMethodNotAllowed, srcBucket, srcObject)
 	}
 
+	if dstOpts.ReplicaLockReconcile {
+		reconcileStoredObjectLock(srcInfo.UserDefined, storedObjectLockState(fi.Metadata))
+		reconcileStoredObjectTags(srcInfo.UserDefined, fi.Metadata)
+	}
+
 	filterOnlineDisksInplace(fi, metaArr, onlineDisks)
 
 	versionID := srcInfo.VersionID
@@ -1280,7 +1285,11 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 			opts.NoLock = true
 		}
 
-		obj, err := er.getObjectInfo(ctx, bucket, object, opts)
+		getObjectInfo := er.getObjectInfo
+		if opts.replicaObjectInfo != nil {
+			getObjectInfo = opts.replicaObjectInfo
+		}
+		obj, err := getObjectInfo(ctx, bucket, object, opts)
 		// A destination read that fails for a reason other than not-found must not
 		// be taken as a passed precondition or as absent lock state.
 		if err != nil && !isErrVersionNotFound(err) && !isErrObjectNotFound(err) {
@@ -1297,17 +1306,12 @@ func (er erasureObjects) putObject(ctx context.Context, bucket string, object st
 			}
 		}
 
-		// Order this trusted SSE-C replica's Object Lock against the addressed
-		// version's stored state, read on this erasure set under the write lock,
-		// so a value that lost the ordering cannot overwrite a newer one committed
-		// after the handler decided (issue #120). Only reconcile against an
-		// existing version; on not-found the write's own accepted lock is kept.
-		//
-		// Scope: correct for a single erasure set. A multi-pool deployment
-		// (duplicate versions across pools, ModTime ties, cross-pool lock
-		// authority) is out of scope and tracked in pgsty/silo#133.
+		// Re-read under the write lock, using the pools-layer resolver when
+		// present. A missing version keeps the incoming accepted state; an
+		// existing version contributes independently ordered lock and tags.
 		if opts.ReplicaLockReconcile && err == nil {
 			reconcileStoredObjectLock(opts.UserDefined, storedObjectLockState(obj.UserDefined))
+			reconcileStoredObjectTags(opts.UserDefined, obj.UserDefined)
 		}
 	}
 

cmd/erasure-server-pool-consistency.go

@@ -0,0 +1,350 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// This file is part of Silo 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 <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"context"
+	"maps"
+	"sort"
+	"strings"
+	"time"
+
+	madmin "github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/bucket/lifecycle"
+	xhttp "github.com/minio/minio/internal/http"
+)
+
+// objectPoolInfos reads the addressed version in every pool, including pools
+// draining their contents. The caller must hold the pools-layer object lock.
+// An unreadable pool may contain a newer version or metadata; it is not absence.
+func (z *erasureServerPools) objectPoolInfos(ctx context.Context, bucket, object string, opts ObjectOptions) ([]PoolObjInfo, error) {
+	opts.NoLock = true
+	opts.CheckPrecondFn = nil
+	var copies []PoolObjInfo
+	for i, pool := range z.serverPools {
+		oi, err := pool.GetObjectInfo(ctx, bucket, object, opts)
+		if err == nil || (oi.DeleteMarker && (isErrObjectNotFound(err) || isErrMethodNotAllowed(err))) {
+			copies = append(copies, PoolObjInfo{Index: i, ObjInfo: oi})
+			continue
+		}
+		if !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+			return nil, err
+		}
+	}
+	sort.Slice(copies, func(i, j int) bool {
+		a, b := copies[i], copies[j]
+		if a.ObjInfo.ModTime.Equal(b.ObjInfo.ModTime) {
+			return a.Index < b.Index
+		}
+		return a.ObjInfo.ModTime.After(b.ObjInfo.ModTime)
+	})
+	if len(copies) == 0 {
+		if opts.VersionID != "" {
+			return nil, VersionNotFound{Bucket: bucket, Object: decodeDirObject(object), VersionID: opts.VersionID}
+		}
+		return nil, ObjectNotFound{Bucket: bucket, Object: decodeDirObject(object)}
+	}
+	return copies, nil
+}
+
+// Healing also commits object metadata. Both queued healing and drive healing
+// must use the same namespace as pooled writes, rather than racing them under
+// a destination set's independent namespace.
+func (z *erasureServerPools) healObjectInPool(ctx context.Context, er *erasureObjects, bucket, object, versionID string, opts madmin.HealOpts) (madmin.HealResultItem, error) {
+	if !z.SinglePool() && !opts.NoLock {
+		lk := z.NewNSLock(bucket, object)
+		lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
+		if err != nil {
+			return madmin.HealResultItem{}, err
+		}
+		ctx = lkctx.Context()
+		defer lk.Unlock(lkctx)
+		opts.NoLock = true
+	}
+	return er.HealObject(ctx, bucket, object, versionID, opts)
+}
+
+// mergePoolLockState resolves each independently ordered field. An ordered
+// removal is a value in its own right; an absent unordered field is not one.
+func mergePoolLockState(copies []PoolObjInfo) objectLockState {
+	state := storedObjectLockState(copies[0].ObjInfo.UserDefined)
+	for _, copy := range copies[1:] {
+		next := storedObjectLockState(copy.ObjInfo.UserDefined)
+		retentionTime, _ := time.Parse(time.RFC3339Nano, next.retentionTimestamp)
+		if state.retentionIsOlderThan(retentionTime) || (state.retentionTimestamp == "" && state.mode == "" && next.mode != "") {
+			state.mode, state.retainUntil, state.retentionTimestamp = next.mode, next.retainUntil, next.retentionTimestamp
+		}
+		holdTime, _ := time.Parse(time.RFC3339Nano, next.legalHoldTimestamp)
+		if state.legalHoldIsOlderThan(holdTime) || (state.legalHoldTimestamp == "" && state.legalHold == "" && next.legalHold != "") {
+			state.legalHold, state.legalHoldTimestamp = next.legalHold, next.legalHoldTimestamp
+		}
+	}
+	return state
+}
+
+func replaceObjectLockMetadata(metadata map[string]string, state objectLockState) {
+	for _, key := range []string{
+		strings.ToLower(xhttp.AmzObjectLockMode), strings.ToLower(xhttp.AmzObjectLockRetainUntilDate),
+		strings.ToLower(xhttp.AmzObjectLockLegalHold),
+		ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp,
+		ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp,
+	} {
+		// Metadata updates use maps.Copy at the set layer. Empty values also
+		// clear a previous value there, including timestamp-only removals.
+		if _, exists := metadata[key]; exists {
+			metadata[key] = ""
+		}
+	}
+	state.restoreRetention(metadata)
+	state.restoreLegalHold(metadata)
+}
+
+func mergedPoolObjectInfo(copies []PoolObjInfo) ObjectInfo {
+	oi := copies[0].ObjInfo
+	oi.UserDefined = maps.Clone(oi.UserDefined)
+	if oi.UserDefined == nil {
+		oi.UserDefined = make(map[string]string)
+	}
+	replaceObjectLockMetadata(oi.UserDefined, mergePoolLockState(copies))
+	for _, copy := range copies[1:] {
+		ts := copy.ObjInfo.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp]
+		stamp, _ := time.Parse(time.RFC3339Nano, ts)
+		if olderThan(oi.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp], stamp) {
+			oi.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp] = ts
+			oi.UserDefined[xhttp.AmzObjectTagging] = copy.ObjInfo.UserDefined[xhttp.AmzObjectTagging]
+		}
+	}
+	oi.UserTags = oi.UserDefined[xhttp.AmzObjectTagging]
+	return oi
+}
+
+func (z *erasureServerPools) replicaObjectInfo(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
+	if opts.VersionID == "" {
+		opts.VersionID = nullVersionID
+	}
+	copies, err := z.objectPoolInfos(ctx, bucket, object, opts)
+	if err != nil {
+		return ObjectInfo{}, err
+	}
+	if copies[0].ObjInfo.DeleteMarker {
+		return copies[0].ObjInfo, MethodNotAllowed{Bucket: bucket, Object: decodeDirObject(object), VersionID: opts.VersionID}
+	}
+	return mergedPoolObjectInfo(copies), nil
+}
+
+// metadataPoolInfos resolves an unqualified metadata request to one logical
+// version before collecting its copies, rather than mixing different versions.
+func (z *erasureServerPools) metadataPoolInfos(ctx context.Context, bucket, object string, opts ObjectOptions) ([]PoolObjInfo, error) {
+	copies, err := z.objectPoolInfos(ctx, bucket, object, opts)
+	if err != nil {
+		return nil, err
+	}
+	if copies[0].ObjInfo.DeleteMarker {
+		return nil, MethodNotAllowed{Bucket: bucket, Object: decodeDirObject(object), VersionID: opts.VersionID}
+	}
+	if opts.VersionID == "" {
+		opts.VersionID = copies[0].ObjInfo.VersionID
+		if opts.VersionID == "" {
+			opts.VersionID = nullVersionID
+		}
+		return z.objectPoolInfos(ctx, bucket, object, opts)
+	}
+	return copies, nil
+}
+
+func (z *erasureServerPools) updatePoolMetadata(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
+	copies, err := z.metadataPoolInfos(ctx, bucket, object, opts)
+	if err != nil {
+		return ObjectInfo{}, err
+	}
+	updated := mergedPoolObjectInfo(copies)
+	before := maps.Clone(updated.UserDefined)
+	if opts.EvalMetadataFn != nil {
+		if _, err := opts.EvalMetadataFn(&updated, nil); err != nil {
+			return ObjectInfo{}, err
+		}
+	}
+	changes := make(map[string]string)
+	for key, value := range updated.UserDefined {
+		if prior, ok := before[key]; !ok || prior != value {
+			changes[key] = value
+		}
+	}
+	for key := range before {
+		if _, ok := updated.UserDefined[key]; !ok {
+			changes[key] = ""
+		}
+	}
+	state := storedObjectLockState(updated.UserDefined)
+	opts.VersionID = updated.VersionID
+	if opts.VersionID == "" {
+		opts.VersionID = nullVersionID
+	}
+	opts.NoLock = true
+	opts.EvalMetadataFn = func(oi *ObjectInfo, _ error) (ReplicateDecision, error) {
+		maps.Copy(oi.UserDefined, changes)
+		replaceObjectLockMetadata(oi.UserDefined, state)
+		for _, key := range []string{xhttp.AmzObjectTagging, ReservedMetadataPrefixLower + TaggingTimestamp} {
+			value, exists := updated.UserDefined[key]
+			if exists || oi.UserDefined[key] != "" {
+				oi.UserDefined[key] = value
+			}
+		}
+		return ReplicateDecision{}, nil
+	}
+	var primary ObjectInfo
+	for _, copy := range copies {
+		oi, err := z.serverPools[copy.Index].PutObjectMetadata(ctx, bucket, object, opts)
+		if err != nil {
+			return ObjectInfo{}, err
+		}
+		if copy.Index == copies[0].Index {
+			primary = oi
+		}
+	}
+	return primary, nil
+}
+
+func reconcileStoredObjectTags(metadata, stored map[string]string) {
+	key := ReservedMetadataPrefixLower + TaggingTimestamp
+	stamp, err := time.Parse(time.RFC3339Nano, stored[key])
+	if err != nil {
+		return
+	}
+	incoming, err := time.Parse(time.RFC3339Nano, metadata[key])
+	if err != nil || !stamp.Before(incoming) {
+		metadata[key] = stored[key]
+		metadata[xhttp.AmzObjectTagging] = stored[xhttp.AmzObjectTagging]
+	}
+}
+
+// A restored version still owns its tier reference even while IsRemote is
+// false. Only the last copy of a reference may schedule its contents for GC.
+func sharesTierObject(oi ObjectInfo, copies []PoolObjInfo) bool {
+	ref := oi.TransitionedObject
+	if ref.Status != lifecycle.TransitionComplete {
+		return false
+	}
+	for _, copy := range copies {
+		other := copy.ObjInfo.TransitionedObject
+		if other.Status == lifecycle.TransitionComplete && ref.Tier == other.Tier && ref.Name == other.Name && ref.VersionID == other.VersionID {
+			return true
+		}
+	}
+	return false
+}
+
+// retireReplicaCopies runs only after committing a replacement. Failures are
+// returned to the caller, so a stale copy cannot be hidden behind a successful
+// response. Data movement owns its source cleanup and does not use this helper.
+func (z *erasureServerPools) retireReplicaCopies(ctx context.Context, bucket, object string, keep int, oi ObjectInfo) error {
+	versionID := oi.VersionID
+	if versionID == "" {
+		versionID = nullVersionID
+	}
+	copies, err := z.objectPoolInfos(ctx, bucket, object, ObjectOptions{VersionID: versionID})
+	if err != nil {
+		return err
+	}
+	var retained []PoolObjInfo
+	for _, copy := range copies {
+		if copy.Index == keep {
+			retained = []PoolObjInfo{copy}
+			break
+		}
+	}
+	if len(retained) == 0 {
+		return VersionNotFound{Bucket: bucket, Object: decodeDirObject(object), VersionID: versionID}
+	}
+	for i, copy := range copies {
+		if copy.Index == keep {
+			continue
+		}
+		_, err := z.serverPools[copy.Index].DeleteObject(ctx, bucket, object,
+			ObjectOptions{
+				VersionID: versionID, NoLock: true, NoAuditLog: true,
+				SkipFreeVersion: sharesTierObject(copy.ObjInfo, retained) || sharesTierObject(copy.ObjInfo, copies[i+1:]),
+			})
+		if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+			return err
+		}
+	}
+	return nil
+}
+
+// deleteObjectConditional evaluates the condition once against the logical
+// version, then removes all its copies under the same lock as pooled writers.
+func (z *erasureServerPools) deleteObjectConditional(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
+	copies, err := z.objectPoolInfos(ctx, bucket, object, opts)
+	if err != nil {
+		return ObjectInfo{}, err
+	}
+	primary := copies[0]
+	if opts.CheckPrecondFn(primary.ObjInfo) {
+		return ObjectInfo{}, PreConditionFailed{}
+	}
+	opts.CheckPrecondFn = nil
+	opts.NoLock = true
+	if opts.EvalRetentionBypassFn != nil || opts.EvalMetadataFn != nil {
+		versions, err := z.metadataPoolInfos(ctx, bucket, object, opts)
+		if err != nil {
+			return ObjectInfo{}, err
+		}
+		logical := mergedPoolObjectInfo(versions)
+		if opts.EvalRetentionBypassFn != nil {
+			if err := opts.EvalRetentionBypassFn(logical, nil); err != nil {
+				return ObjectInfo{}, err
+			}
+			opts.EvalRetentionBypassFn = nil
+		}
+		if opts.EvalMetadataFn != nil {
+			decision, err := opts.EvalMetadataFn(&logical, nil)
+			if err != nil {
+				return ObjectInfo{}, err
+			}
+			if decision.ReplicateAny() {
+				opts.SetDeleteReplicationState(decision, opts.VersionID)
+			}
+			opts.EvalMetadataFn = nil
+		}
+	}
+	if opts.VersionID == "" && (opts.Versioned || opts.VersionSuspended) {
+		// 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.replicationDecision = opts.DeleteReplication.ReplicateDecisionStr
+		return oi, err
+	}
+	// Retire non-authoritative copies first. If cleanup fails, retain the
+	// authoritative version and report the error instead of acknowledging a
+	// deletion that would expose an older copy.
+	for i := 1; i < len(copies); i++ {
+		candidate := copies[i]
+		deleteOpts := opts
+		deleteOpts.SkipFreeVersion = opts.SkipFreeVersion || sharesTierObject(candidate.ObjInfo, copies[:1]) || sharesTierObject(candidate.ObjInfo, copies[i+1:])
+		_, err := z.serverPools[candidate.Index].DeleteObject(ctx, bucket, object, deleteOpts)
+		if err != nil && !isErrObjectNotFound(err) && !isErrVersionNotFound(err) {
+			return ObjectInfo{}, err
+		}
+	}
+	oi, err := z.serverPools[primary.Index].DeleteObject(ctx, bucket, object, opts)
+	oi.Name = decodeDirObject(object)
+	oi.replicationDecision = opts.DeleteReplication.ReplicateDecisionStr
+	return oi, err
+}

cmd/erasure-server-pool-consistency_test.go

@@ -0,0 +1,682 @@
+// Copyright (c) 2026 Feng Ruohang
+//
+// This file is part of Silo 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 <http://www.gnu.org/licenses/>.
+
+package cmd
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"maps"
+	"strings"
+	"sync"
+	"testing"
+	"time"
+
+	madmin "github.com/minio/madmin-go/v3"
+	xhttp "github.com/minio/minio/internal/http"
+)
+
+func consistencyPools(t *testing.T) (*erasureServerPools, string) {
+	t.Helper()
+	ctx, cancel := context.WithCancel(t.Context())
+	obj, dirs, err := prepareErasurePoolsWithContext(ctx)
+	if err != nil {
+		cancel()
+		t.Fatal(err)
+	}
+	z := obj.(*erasureServerPools)
+	previous := newObjectLayerFn()
+	setObjectLayer(z)
+	t.Cleanup(func() { cancel(); z.Shutdown(context.Background()); removeRoots(dirs); setObjectLayer(previous) })
+	bucket := "pool-consistency"
+	if err := z.MakeBucket(t.Context(), bucket, MakeBucketOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	return z, bucket
+}
+
+func putConsistencyObject(t *testing.T, z *erasureServerPools, bucket, object string, pool int, body string, opts ObjectOptions) ObjectInfo {
+	t.Helper()
+	oi, err := z.serverPools[pool].PutObject(t.Context(), bucket, object,
+		mustGetPutObjReader(t, bytes.NewBufferString(body), int64(len(body)), "", ""), opts)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return oi
+}
+
+func TestPoolsConditionalDeleteVersionSelection(t *testing.T) {
+	z, bucket := consistencyPools(t)
+	const object = "split-versions"
+	old := putConsistencyObject(t, z, bucket, object, 1, "old", ObjectOptions{Versioned: true, MTime: time.Now().Add(-time.Hour)})
+	latest := putConsistencyObject(t, z, bucket, object, 0, "latest", ObjectOptions{Versioned: true})
+	_, err := z.DeleteObject(t.Context(), bucket, object, ObjectOptions{
+		Versioned: true, VersionID: old.VersionID, HasIfMatch: true,
+		CheckPrecondFn: func(oi ObjectInfo) bool { return oi.ETag != old.ETag },
+	})
+	if err != nil {
+		t.Fatalf("delete addressed version in pool 1: %v", err)
+	}
+	if _, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: old.VersionID}); !isErrVersionNotFound(err) {
+		t.Errorf("addressed version survived: %v", err)
+	}
+	if oi, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); err != nil || oi.VersionID != latest.VersionID {
+		t.Errorf("delete changed the latest version: %+v, %v", oi, err)
+	}
+}
+
+func TestPoolsConditionalDeleteDuplicateVersion(t *testing.T) {
+	z, bucket := consistencyPools(t)
+	const object = "duplicate-version"
+	oi := putConsistencyObject(t, z, bucket, object, 0, "payload", ObjectOptions{Versioned: true})
+	putConsistencyObject(t, z, bucket, object, 1, "payload", ObjectOptions{Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime})
+	_, err := z.DeleteObject(t.Context(), bucket, object, ObjectOptions{
+		Versioned: true, VersionID: oi.VersionID, HasIfMatch: true,
+		CheckPrecondFn: func(current ObjectInfo) bool { return current.ETag != oi.ETag },
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID}); !isErrVersionNotFound(err) {
+		t.Fatalf("success left a readable duplicate version: %v", err)
+	}
+}
+
+func TestPoolsConditionalDeleteReportsOtherPoolFailure(t *testing.T) {
+	z, bucket := consistencyPools(t)
+	const object = "delete-failure"
+	putConsistencyObject(t, z, bucket, object, 1, "older", ObjectOptions{MTime: time.Now().Add(-time.Hour)})
+	oi := putConsistencyObject(t, z, bucket, object, 0, "latest", ObjectOptions{})
+	set := z.serverPools[1].getHashedSet(object)
+	getDisks := set.getDisks
+	faulty := append([]StorageAPI(nil), getDisks()...)
+	for i := range faulty {
+		faulty[i] = accessMoveDeleteFaultDisk{StorageAPI: faulty[i], bucket: bucket, object: object}
+	}
+	set.getDisks = func() []StorageAPI { return faulty }
+	defer func() { set.getDisks = getDisks }()
+	_, err := z.DeleteObject(t.Context(), bucket, object, ObjectOptions{
+		HasIfMatch: true, CheckPrecondFn: func(current ObjectInfo) bool { return current.ETag != oi.ETag },
+	})
+	if err == nil {
+		t.Fatal("delete reported success despite the other pool's write failure")
+	}
+}
+
+func TestPoolsConditionalDeleteSerializesPut(t *testing.T) {
+	testPoolsConditionalDeleteWriter(t, false)
+}
+
+func TestPoolsConditionalDeleteSerializesCompletion(t *testing.T) {
+	testPoolsConditionalDeleteWriter(t, true)
+}
+
+func testPoolsConditionalDeleteWriter(t *testing.T, multipart bool) {
+	z, bucket := consistencyPools(t)
+	const object = "concurrent-put"
+	initial := putConsistencyObject(t, z, bucket, object, 1, "before", ObjectOptions{})
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+	reader := mustGetPutObjReader(t, bytes.NewBufferString("after"), 5, "", "")
+	destination := 1
+	write := func() error {
+		_, err := z.PutObject(ctx, bucket, object, reader, ObjectOptions{DataMovement: true, SrcPoolIdx: 0, DstPoolIdx: &destination})
+		return err
+	}
+	if multipart {
+		mp, err := z.serverPools[1].NewMultipartUpload(ctx, bucket, object, ObjectOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		part, err := z.serverPools[1].PutObjectPart(ctx, bucket, object, mp.UploadID, 1, reader, ObjectOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+		write = func() error {
+			_, err := z.CompleteMultipartUpload(ctx, bucket, object, mp.UploadID,
+				[]CompletePart{{PartNumber: 1, ETag: part.ETag}}, ObjectOptions{})
+			return err
+		}
+	}
+	checked, resume := make(chan struct{}), make(chan struct{})
+	var resumeOnce sync.Once
+	release := func() { resumeOnce.Do(func() { close(resume) }) }
+	defer release()
+	deleted := make(chan error, 1)
+	go func() {
+		_, err := z.DeleteObject(ctx, bucket, object, ObjectOptions{
+			HasIfMatch: true,
+			CheckPrecondFn: func(oi ObjectInfo) bool {
+				close(checked)
+				<-resume
+				return oi.ETag != initial.ETag
+			},
+		})
+		deleted <- err
+	}()
+	select {
+	case <-checked:
+	case err := <-deleted:
+		t.Fatalf("delete did not evaluate the precondition: %v", err)
+	case <-ctx.Done():
+		t.Fatal(ctx.Err())
+	}
+	written := make(chan error, 1)
+	go func() { written <- write() }()
+	var early bool
+	select {
+	case err := <-written:
+		early = true
+		if err != nil {
+			t.Errorf("concurrent PUT: %v", err)
+		}
+	case <-time.After(250 * time.Millisecond):
+	}
+	release()
+	if err := <-deleted; err != nil {
+		t.Fatal(err)
+	}
+	if !early {
+		if err := <-written; err != nil {
+			t.Fatal(err)
+		}
+	}
+	if early {
+		t.Error("PUT committed while DELETE was between its comparison and removal")
+	}
+	if _, err := z.GetObjectInfo(ctx, bucket, object, ObjectOptions{}); err != nil {
+		t.Errorf("matching the old ETag removed the concurrent replacement: %v", err)
+	}
+}
+
+type consistencyGateReader struct {
+	io.Reader
+	entered, resume chan struct{}
+	once            sync.Once
+	release         sync.Once
+}
+
+func (r *consistencyGateReader) Read(p []byte) (int, error) {
+	r.once.Do(func() { close(r.entered); <-r.resume })
+	return r.Reader.Read(p)
+}
+
+func TestPoolsReplicaSerializesMetadataAndHealing(t *testing.T) {
+	for _, heal := range []bool{false, true} {
+		t.Run(fmt.Sprintf("heal=%t", heal), func(t *testing.T) {
+			z, bucket := consistencyPools(t)
+			const object = "metadata-race"
+			old, recent := "2026-09-09T09:00:00Z", "2026-09-09T10:00:00Z"
+			meta := poolLockMetadata("GOVERNANCE", "OFF", old, old)
+			oi := putConsistencyObject(t, z, bucket, object, 1, "data", ObjectOptions{Versioned: true, UserDefined: meta})
+			z.poolMetaMutex.Lock()
+			z.poolMeta.Pools[0].Decommission = &PoolDecommissionInfo{}
+			z.poolMetaMutex.Unlock()
+			ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+			defer cancel()
+			gate := &consistencyGateReader{Reader: strings.NewReader("data"), entered: make(chan struct{}), resume: make(chan struct{})}
+			release := func() { gate.release.Do(func() { close(gate.resume) }) }
+			defer release()
+			reader := mustGetPutObjReader(t, gate, 4, "", "")
+			written := make(chan error, 1)
+			go func() {
+				_, err := z.PutObject(ctx, bucket, object, reader, ObjectOptions{
+					Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime,
+					ReplicaLockReconcile: true, UserDefined: maps.Clone(meta),
+				})
+				written <- err
+			}()
+			select {
+			case <-gate.entered:
+			case err := <-written:
+				t.Fatalf("write failed before consuming the body: %v", err)
+			case <-ctx.Done():
+				t.Fatal(ctx.Err())
+			}
+			mutated := make(chan error, 1)
+			go func() {
+				var err error
+				if heal {
+					_, err = z.healObjectInPool(ctx, z.serverPools[1].getHashedSet(object), bucket, object, oi.VersionID, madmin.HealOpts{})
+				} else {
+					_, err = z.PutObjectMetadata(ctx, bucket, object, ObjectOptions{
+						VersionID: oi.VersionID, MTime: oi.ModTime,
+						EvalMetadataFn: func(current *ObjectInfo, _ error) (ReplicateDecision, error) {
+							current.UserDefined[strings.ToLower(xhttp.AmzObjectLockLegalHold)] = "ON"
+							current.UserDefined[ReservedMetadataPrefixLower+ObjectLockLegalHoldTimestamp] = recent
+							return ReplicateDecision{}, nil
+						},
+					})
+				}
+				mutated <- err
+			}()
+			var early bool
+			select {
+			case err := <-mutated:
+				early = true
+				t.Errorf("mutation completed during the paused replica write: %v", err)
+			case <-time.After(250 * time.Millisecond):
+			}
+			release()
+			if err := <-written; err != nil {
+				t.Fatal(err)
+			}
+			if !early {
+				if err := <-mutated; err != nil {
+					t.Fatal(err)
+				}
+			}
+			got, err := z.GetObjectInfo(ctx, bucket, object, ObjectOptions{VersionID: oi.VersionID})
+			if err != nil {
+				t.Fatal(err)
+			}
+			if !heal && storedObjectLockState(got.UserDefined).legalHold != "ON" {
+				t.Error("replica write rolled back the newer metadata update")
+			}
+		})
+	}
+}
+
+func TestPoolsConditionalDeletePreservesVersionHistory(t *testing.T) {
+	z, bucket := consistencyPools(t)
+	const object = "conditional-marker"
+	older := putConsistencyObject(t, z, bucket, object, 1, "older", ObjectOptions{Versioned: true, MTime: time.Now().Add(-time.Hour)})
+	latest := putConsistencyObject(t, z, bucket, object, 0, "latest", ObjectOptions{Versioned: true})
+	_, err := z.DeleteObject(t.Context(), bucket, object, ObjectOptions{
+		Versioned:      true,
+		CheckPrecondFn: func(oi ObjectInfo) bool { return oi.ETag != older.ETag },
+	})
+	if !isErrPreconditionFailed(err) {
+		t.Fatalf("wrong latest ETag: %v", err)
+	}
+	marker, err := z.DeleteObject(t.Context(), bucket, object, ObjectOptions{
+		Versioned:      true,
+		CheckPrecondFn: func(oi ObjectInfo) bool { return oi.ETag != latest.ETag },
+	})
+	if err != nil || !marker.DeleteMarker {
+		t.Fatalf("logical delete did not create a marker: %+v, %v", marker, err)
+	}
+	if got, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); !isErrObjectNotFound(err) || !got.DeleteMarker {
+		t.Errorf("delete marker did not hide the split object: %+v, %v", got, err)
+	}
+	for _, version := range []string{older.VersionID, latest.VersionID} {
+		if _, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: version}); err != nil {
+			t.Errorf("conditional marker removed history %s: %v", version, err)
+		}
+	}
+}
+
+func poolLockMetadata(retention, hold, retentionTime, holdTime string) map[string]string {
+	return map[string]string{
+		strings.ToLower(xhttp.AmzObjectLockMode):                   retention,
+		strings.ToLower(xhttp.AmzObjectLockRetainUntilDate):        "2030-01-01T00:00:00Z",
+		strings.ToLower(xhttp.AmzObjectLockLegalHold):              hold,
+		ReservedMetadataPrefixLower + ObjectLockRetentionTimestamp: retentionTime,
+		ReservedMetadataPrefixLower + ObjectLockLegalHoldTimestamp: holdTime,
+	}
+}
+
+func TestPoolsReplicaIndependentLockWinners(t *testing.T) {
+	for _, tc := range []struct {
+		name                 string
+		multipart, rebalance bool
+	}{
+		{"put-decommission", false, false},
+		{"put-rebalance", false, true},
+		{"multipart-decommission", true, false},
+		{"multipart-rebalance", true, true},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			z, bucket := consistencyPools(t)
+			const object = "split-lock-state"
+			old, recent := "2026-09-09T09:00:00Z", "2026-09-09T10:00:00Z"
+			first := poolLockMetadata("GOVERNANCE", "OFF", recent, old)
+			second := poolLockMetadata("", "ON", old, recent)
+			oi := putConsistencyObject(t, z, bucket, object, 0, "data", ObjectOptions{Versioned: true, UserDefined: first})
+			putConsistencyObject(t, z, bucket, object, 1, "data", ObjectOptions{
+				Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime, UserDefined: second,
+			})
+			opts := ObjectOptions{
+				Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime,
+				ReplicaLockReconcile: true, UserDefined: poolLockMetadata("", "OFF", old, old),
+			}
+			var uploadID string
+			var parts []CompletePart
+			if tc.multipart {
+				mp, err := z.serverPools[1].NewMultipartUpload(t.Context(), bucket, object, opts)
+				if err != nil {
+					t.Fatal(err)
+				}
+				uploadID = mp.UploadID
+				part, err := z.serverPools[1].PutObjectPart(t.Context(), bucket, object, uploadID, 1,
+					mustGetPutObjReader(t, bytes.NewBufferString("data"), 4, "", ""), ObjectOptions{})
+				if err != nil {
+					t.Fatal(err)
+				}
+				parts = []CompletePart{{PartNumber: 1, ETag: part.ETag}}
+			}
+			// The upload and both versions predate the routing change. Retention's
+			// winner remains in the draining pool; legal hold's winner is in pool 1.
+			if tc.rebalance {
+				z.rebalMu.Lock()
+				z.rebalMeta = &rebalanceMeta{PoolStats: []*rebalanceStats{
+					{Participating: true, Info: rebalanceInfo{Status: rebalStarted}}, {},
+				}}
+				z.rebalMu.Unlock()
+			} else {
+				z.poolMetaMutex.Lock()
+				z.poolMeta.Pools[0].Decommission = &PoolDecommissionInfo{}
+				z.poolMetaMutex.Unlock()
+			}
+			var err error
+			if tc.multipart {
+				_, err = z.CompleteMultipartUpload(t.Context(), bucket, object, uploadID, parts,
+					ObjectOptions{Versioned: true, MTime: oi.ModTime, ReplicaLockReconcile: true})
+			} else {
+				_, err = z.PutObject(t.Context(), bucket, object,
+					mustGetPutObjReader(t, bytes.NewBufferString("data"), 4, "", ""), opts)
+			}
+			if err != nil {
+				t.Fatal(err)
+			}
+			got, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID})
+			if err != nil {
+				t.Fatal(err)
+			}
+			state := storedObjectLockState(got.UserDefined)
+			if state.mode != "GOVERNANCE" || state.retentionTimestamp != recent || state.legalHold != "ON" || state.legalHoldTimestamp != recent {
+				t.Errorf("pooled read lost independently ordered lock state: %+v", state)
+			}
+			if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID}); !isErrVersionNotFound(err) {
+				t.Errorf("successful replacement left its stale competing copy: %v", err)
+			}
+		})
+	}
+}
+
+func TestPoolsReplicaSoleDrainingOwner(t *testing.T) {
+	for _, rebalance := range []bool{false, true} {
+		for _, null := range []bool{false, true} {
+			t.Run(fmt.Sprintf("rebalance=%t/null=%t", rebalance, null), func(t *testing.T) {
+				z, bucket := consistencyPools(t)
+				const object = "sole-owner"
+				old, recent := "2026-09-09T09:00:00Z", "2026-09-09T10:00:00Z"
+				meta := poolLockMetadata("", "ON", recent, recent)
+				delete(meta, strings.ToLower(xhttp.AmzObjectLockRetainUntilDate))
+				oi := putConsistencyObject(t, z, bucket, object, 0, "data", ObjectOptions{Versioned: !null, UserDefined: meta})
+				versionID := oi.VersionID
+				if null {
+					versionID = nullVersionID
+					// A different latest version must not donate its lock to the null version.
+					putConsistencyObject(t, z, bucket, object, 1, "other-version", ObjectOptions{Versioned: true})
+				}
+				if rebalance {
+					z.rebalMu.Lock()
+					z.rebalMeta = &rebalanceMeta{PoolStats: []*rebalanceStats{
+						{Participating: true, Info: rebalanceInfo{Status: rebalStarted}}, {},
+					}}
+					z.rebalMu.Unlock()
+				} else {
+					z.poolMetaMutex.Lock()
+					z.poolMeta.Pools[0].Decommission = &PoolDecommissionInfo{}
+					z.poolMetaMutex.Unlock()
+				}
+				_, err := z.PutObject(t.Context(), bucket, object,
+					mustGetPutObjReader(t, bytes.NewBufferString("data"), 4, "", ""), ObjectOptions{
+						Versioned: !null, VersionSuspended: null, VersionID: versionID, MTime: oi.ModTime,
+						ReplicaLockReconcile: true, UserDefined: poolLockMetadata("GOVERNANCE", "OFF", old, old),
+					})
+				if err != nil {
+					t.Fatal(err)
+				}
+				got, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: versionID})
+				if err != nil {
+					t.Fatal(err)
+				}
+				state := storedObjectLockState(got.UserDefined)
+				if state.mode != "" || state.retainUntil != "" || state.retentionTimestamp != recent || state.legalHold != "ON" {
+					t.Errorf("lost a removal or legal hold from the sole draining owner: %+v", state)
+				}
+			})
+		}
+	}
+}
+
+func TestPoolsMetadataUpdateUsesMergedVersion(t *testing.T) {
+	z, bucket := consistencyPools(t)
+	const object = "metadata-update"
+	old, recent := "2026-09-09T09:00:00Z", "2026-09-09T10:00:00Z"
+	oi := putConsistencyObject(t, z, bucket, object, 0, "data", ObjectOptions{
+		Versioned: true, UserDefined: poolLockMetadata("GOVERNANCE", "OFF", recent, old),
+	})
+	putConsistencyObject(t, z, bucket, object, 1, "data", ObjectOptions{
+		Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime, UserDefined: poolLockMetadata("", "ON", old, recent),
+	})
+	latest := putConsistencyObject(t, z, bucket, object, 1, "latest", ObjectOptions{Versioned: true})
+	called := 0
+	_, err := z.PutObjectMetadata(t.Context(), bucket, object, ObjectOptions{
+		VersionID: oi.VersionID, MTime: oi.ModTime,
+		EvalMetadataFn: func(current *ObjectInfo, _ error) (ReplicateDecision, error) {
+			called++
+			state := storedObjectLockState(current.UserDefined)
+			if state.mode != "GOVERNANCE" || state.legalHold != "ON" {
+				return ReplicateDecision{}, fmt.Errorf("metadata policy evaluated stale state: %+v", state)
+			}
+			current.UserDefined["custom-update"] = "preserved"
+			return ReplicateDecision{}, nil
+		},
+	})
+	if err != nil || called != 1 {
+		t.Fatalf("metadata update: %v; callback count %d", err, called)
+	}
+	for i, pool := range z.serverPools {
+		got, err := pool.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID})
+		if err != nil {
+			t.Fatal(err)
+		}
+		state := storedObjectLockState(got.UserDefined)
+		if state.mode != "GOVERNANCE" || state.legalHold != "ON" || got.UserDefined["custom-update"] != "preserved" {
+			t.Errorf("pool %d did not receive the merged update: %+v", i, got.UserDefined)
+		}
+	}
+	if current, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{}); err != nil || current.VersionID != latest.VersionID {
+		t.Errorf("updating an older version changed the latest: %+v, %v", current, err)
+	}
+}
+
+func TestPoolsReplicaMetadataCopyReconcilesLockAndTags(t *testing.T) {
+	z, bucket := consistencyPools(t)
+	const object = "replica-metadata-copy"
+	old, recent := "2026-09-09T09:00:00Z", "2026-09-09T10:00:00Z"
+	meta := poolLockMetadata("GOVERNANCE", "OFF", recent, old)
+	meta[xhttp.AmzObjectTagging] = "key=new"
+	meta[ReservedMetadataPrefixLower+TaggingTimestamp] = recent
+	oi := putConsistencyObject(t, z, bucket, object, 0, "data", ObjectOptions{Versioned: true, UserDefined: meta})
+	putConsistencyObject(t, z, bucket, object, 1, "data", ObjectOptions{
+		Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime, UserDefined: poolLockMetadata("", "ON", old, recent),
+	})
+	stale := oi
+	stale.metadataOnly = true
+	stale.UserDefined = maps.Clone(oi.UserDefined)
+	stale.UserDefined[xhttp.AmzObjectTagging] = "key=old"
+	stale.UserDefined[ReservedMetadataPrefixLower+TaggingTimestamp] = old
+	_, err := z.CopyObject(t.Context(), bucket, object, bucket, object, stale,
+		ObjectOptions{VersionID: oi.VersionID}, ObjectOptions{
+			Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime, ReplicaLockReconcile: true,
+		})
+	if err != nil {
+		t.Fatal(err)
+	}
+	got, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	state := storedObjectLockState(got.UserDefined)
+	if state.mode != "GOVERNANCE" || state.legalHold != "ON" || got.UserTags != "key=new" {
+		t.Errorf("metadata replication rolled back a newer field: %+v", got.UserDefined)
+	}
+}
+
+func TestPoolsReplicaCleanupFailureCanRetry(t *testing.T) {
+	z, bucket := consistencyPools(t)
+	const object = "replica-cleanup-failure"
+	old, recent := "2026-09-09T09:00:00Z", "2026-09-09T10:00:00Z"
+	oi := putConsistencyObject(t, z, bucket, object, 0, "data", ObjectOptions{
+		Versioned: true, UserDefined: poolLockMetadata("GOVERNANCE", "ON", recent, recent),
+	})
+	z.poolMetaMutex.Lock()
+	z.poolMeta.Pools[0].Decommission = &PoolDecommissionInfo{}
+	z.poolMetaMutex.Unlock()
+	set := z.serverPools[0].getHashedSet(object)
+	getDisks := set.getDisks
+	faulty := append([]StorageAPI(nil), getDisks()...)
+	for i := range faulty {
+		faulty[i] = accessMoveDeleteFaultDisk{StorageAPI: faulty[i], bucket: bucket, object: object, version: oi.VersionID}
+	}
+	set.getDisks = func() []StorageAPI { return faulty }
+	defer func() { set.getDisks = getDisks }()
+	write := func() error {
+		_, err := z.PutObject(t.Context(), bucket, object,
+			mustGetPutObjReader(t, bytes.NewBufferString("data"), 4, "", ""), ObjectOptions{
+				Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime, ReplicaLockReconcile: true,
+				UserDefined: poolLockMetadata("", "OFF", old, old),
+			})
+		return err
+	}
+	if err := write(); err == nil {
+		t.Fatal("replica replacement hid a competing-copy cleanup failure")
+	}
+	if got, err := z.serverPools[1].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID}); err != nil || storedObjectLockState(got.UserDefined).legalHold != "ON" {
+		t.Fatalf("cleanup failure lost the committed reconciled version: %+v, %v", got, err)
+	}
+	set.getDisks = getDisks
+	if err := write(); err != nil {
+		t.Fatalf("retry could not finish cleanup: %v", err)
+	}
+	if _, err := z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID}); !isErrVersionNotFound(err) {
+		t.Errorf("retry left the competing version: %v", err)
+	}
+}
+
+func TestPoolsRetiringCopyPreservesSharedTierObject(t *testing.T) {
+	for _, test := range []struct {
+		name              string
+		deleting          bool
+		failPrimaryDelete bool
+		differentRemote   bool
+		restored          bool
+	}{
+		{name: "metadata-copy"},
+		{name: "restored-metadata-copy", restored: true},
+		{name: "metadata-copy-distinct-reference", differentRemote: true},
+		{name: "failed-primary-delete", deleting: true, failPrimaryDelete: true},
+		{name: "failed-primary-delete-distinct-reference", deleting: true, failPrimaryDelete: true, differentRemote: true},
+		{name: "successful-delete", deleting: true},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			z, bucket := consistencyPools(t)
+			const object = "shared-tier-object"
+			metadata := map[string]string{
+				ReservedMetadataPrefixLower + TransitionStatus:       "complete",
+				ReservedMetadataPrefixLower + TransitionTier:         "TEST-TIER",
+				ReservedMetadataPrefixLower + TransitionedObjectName: "shared-remote-object",
+				ReservedMetadataPrefixLower + TransitionedVersionID:  "shared-remote-version",
+			}
+			if test.restored {
+				metadata[xhttp.AmzRestore] = completedRestoreObj(time.Now().Add(time.Hour)).String()
+			}
+			oi := putConsistencyObject(t, z, bucket, object, 0, "data", ObjectOptions{Versioned: true, UserDefined: metadata})
+			secondaryMetadata := maps.Clone(metadata)
+			if test.differentRemote {
+				secondaryMetadata[ReservedMetadataPrefixLower+TransitionedObjectName] = "other-remote-object"
+			}
+			putConsistencyObject(t, z, bucket, object, 1, "data", ObjectOptions{
+				Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime, UserDefined: secondaryMetadata,
+			})
+			current, err := z.GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID})
+			if err != nil || current.TransitionedObject.Status != "complete" || current.IsRemote() == test.restored {
+				t.Fatalf("fixture did not persist the tier reference: %+v, %v", current.TransitionedObject, err)
+			}
+			if test.failPrimaryDelete {
+				// The authoritative copy remains readable if its deletion fails.
+				// Retiring a secondary copy must not schedule its shared remote
+				// contents for garbage collection in that case.
+				set := z.serverPools[0].getHashedSet(object)
+				getDisks := set.getDisks
+				faulty := append([]StorageAPI(nil), getDisks()...)
+				for i := range faulty {
+					faulty[i] = accessMoveDeleteFaultDisk{StorageAPI: faulty[i], bucket: bucket, object: object, version: oi.VersionID}
+				}
+				set.getDisks = func() []StorageAPI { return faulty }
+				defer func() { set.getDisks = getDisks }()
+			}
+			if test.deleting {
+				_, err := z.DeleteObject(t.Context(), bucket, object, ObjectOptions{
+					Versioned: true, VersionID: oi.VersionID,
+					CheckPrecondFn: func(info ObjectInfo) bool { return info.ETag != current.ETag },
+				})
+				if (err != nil) != test.failPrimaryDelete {
+					t.Fatalf("unexpected authoritative delete result: %v", err)
+				}
+			} else {
+				current.metadataOnly = true
+				current.UserDefined["metadata-update"] = "new"
+				_, err := z.CopyObject(t.Context(), bucket, object, bucket, object, current,
+					ObjectOptions{VersionID: oi.VersionID}, ObjectOptions{
+						Versioned: true, VersionID: oi.VersionID, MTime: oi.ModTime, ReplicaLockReconcile: true,
+					})
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+			_, err = z.serverPools[0].GetObjectInfo(t.Context(), bucket, object, ObjectOptions{VersionID: oi.VersionID})
+			primaryDeleted := test.deleting && !test.failPrimaryDelete
+			if primaryDeleted {
+				if !isErrVersionNotFound(err) {
+					t.Fatalf("authoritative copy survived successful delete: %v", err)
+				}
+			} else if err != nil {
+				t.Fatalf("lost retained authoritative copy: %v", err)
+			}
+			for pool, wantFree := range []bool{primaryDeleted, test.differentRemote} {
+				for _, disk := range z.serverPools[pool].getHashedSet(object).getDisks() {
+					data, err := disk.ReadAll(t.Context(), bucket, pathJoin(object, xlStorageFormatFile))
+					if errors.Is(err, errFileNotFound) && !wantFree {
+						continue
+					}
+					if err != nil {
+						t.Fatal(err)
+					}
+					versions, err := getFileInfoVersions(data, bucket, object, false)
+					if err != nil {
+						t.Fatal(err)
+					}
+					wantCount := 0
+					if wantFree {
+						wantCount = 1
+					}
+					if len(versions.FreeVersions) != wantCount {
+						t.Fatalf("pool %d has %d tier GC markers, want %d", pool, len(versions.FreeVersions), wantCount)
+					}
+				}
+			}
+		})
+	}
+}

cmd/erasure-server-pool-decom_test.go

@@ -23,6 +23,10 @@ import (
 )
 
 func prepareErasurePools() (ObjectLayer, []string, error) {
+	return prepareErasurePoolsWithContext(context.Background())
+}
+
+func prepareErasurePoolsWithContext(ctx context.Context) (ObjectLayer, []string, error) {
 	nDisks := 32
 	fsDirs, err := getRandomDisks(nDisks)
 	if err != nil {
@@ -32,7 +36,7 @@ func prepareErasurePools() (ObjectLayer, []string, error) {
 	pools := mustGetPoolEndpoints(0, fsDirs[:16]...)
 	pools = append(pools, mustGetPoolEndpoints(1, fsDirs[16:]...)...)
 
-	objLayer, _, err := initObjectLayer(context.Background(), pools)
+	objLayer, _, err := initObjectLayer(ctx, pools)
 	if err != nil {
 		removeRoots(fsDirs)
 		return nil, nil, err

cmd/erasure-server-pool.go

@@ -645,6 +645,12 @@ func (z *erasureServerPools) getPoolIdxNoLock(ctx context.Context, bucket, objec
 // if none are found falls back to most available space pool, this function is
 // designed to be only used by PutObject, CopyObject (newObject creation) and NewMultipartUpload.
 func (z *erasureServerPools) getPoolIdx(ctx context.Context, bucket, object string, size int64, dstPoolIdx *int) (idx int, err error) {
+	return z.getWritePoolIdx(ctx, bucket, object, size, dstPoolIdx, false)
+}
+
+// getWritePoolIdx keeps the write-allocation policy when the caller already
+// holds the pools-layer object lock and must not reacquire a set read lock.
+func (z *erasureServerPools) getWritePoolIdx(ctx context.Context, bucket, object string, size int64, dstPoolIdx *int, noLock bool) (idx int, err error) {
 	if dstPoolIdx != nil {
 		if *dstPoolIdx < 0 || *dstPoolIdx >= len(z.serverPools) {
 			return -1, errInvalidArgument
@@ -656,6 +662,7 @@ func (z *erasureServerPools) getPoolIdx(ctx context.Context, bucket, object stri
 	}
 
 	pinfo, _, err := z.getPoolInfoExistingWithOpts(ctx, bucket, object, ObjectOptions{
+		NoLock:             noLock,
 		SkipDecommissioned: true,
 		SkipRebalancing:    true,
 	})
@@ -1164,7 +1171,18 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec
 		return z.serverPools[0].PutObject(ctx, bucket, object, data, opts)
 	}
 
-	idx, err := z.getPoolIdx(ctx, bucket, object, data.Size(), dataMovementDstPool(opts))
+	if !opts.NoLock {
+		lk := z.NewNSLock(bucket, object)
+		lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
+		if err != nil {
+			return ObjectInfo{}, err
+		}
+		ctx = lkctx.Context()
+		defer lk.Unlock(lkctx)
+	}
+	opts.NoLock = true
+
+	idx, err := z.getWritePoolIdx(ctx, bucket, object, data.Size(), dataMovementDstPool(opts), true)
 	if err != nil {
 		return ObjectInfo{}, err
 	}
@@ -1178,7 +1196,15 @@ func (z *erasureServerPools) PutObject(ctx context.Context, bucket string, objec
 		}
 	}
 
-	return z.serverPools[idx].PutObject(ctx, bucket, object, data, opts)
+	if opts.ReplicaLockReconcile || opts.DataMovement {
+		opts.ReplicaLockReconcile = true
+		opts.replicaObjectInfo = z.replicaObjectInfo
+	}
+	oi, err := z.serverPools[idx].PutObject(ctx, bucket, object, data, opts)
+	if err == nil && opts.ReplicaLockReconcile && !opts.DataMovement {
+		err = z.retireReplicaCopies(ctx, bucket, object, idx, oi)
+	}
+	return oi, err
 }
 
 func (z *erasureServerPools) deletePrefix(ctx context.Context, bucket string, prefix string) error {
@@ -1199,15 +1225,7 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
 		object = encodeDirObject(object)
 	}
 
-	// Acquire a write lock before deleting the object.
-	//
-	// NOTE: this lock is taken at the server-pool level. The conditional
-	// (If-Match) precondition below relies on this lock making the read-check-
-	// delete sequence atomic. That holds for a single erasure set: the write
-	// path (PutObject) locks at the destination set, which shares this lock's
-	// namespace only within one set. Multi-pool conditional-delete atomicity
-	// (concurrent writers across pools, cross-pool version selection) is a
-	// separate concern tracked as a follow-up.
+	// Serialize logical deletion with pooled writes and metadata updates.
 	lk := z.NewNSLock(bucket, object)
 	lkctx, err := lk.GetLock(ctx, globalDeleteOperationTimeout)
 	if err != nil {
@@ -1242,6 +1260,10 @@ func (z *erasureServerPools) DeleteObject(ctx context.Context, bucket string, ob
 		return objInfo, err
 	}
 
+	if !z.SinglePool() && opts.CheckPrecondFn != nil {
+		return z.deleteObjectConditional(ctx, bucket, object, opts)
+	}
+
 	gopts := opts
 	gopts.NoLock = true
 
@@ -1377,8 +1399,12 @@ func (z *erasureServerPools) deleteObjectFromAllPools(ctx context.Context, bucke
 	// the delete call tries to clean up other pools during DeleteObject call.
 	objInfo = dobjects[0]
 	objInfo.Name = decodeDirObject(object)
-	err = derrs[0]
-	return objInfo, err
+	for _, derr := range derrs {
+		if derr != nil && !isErrObjectNotFound(derr) && !isErrVersionNotFound(derr) {
+			return objInfo, derr
+		}
+	}
+	return objInfo, derrs[0]
 }
 
 func (z *erasureServerPools) DeleteObjects(ctx context.Context, bucket string, objects []ObjectToDelete, opts ObjectOptions) ([]DeletedObject, []error) {
@@ -1466,6 +1492,22 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
 		dstOpts.NoLock = true
 	}
 
+	if !z.SinglePool() && cpSrcDstSame && srcInfo.metadataOnly && dstOpts.ReplicaLockReconcile && srcOpts.VersionID == dstOpts.VersionID {
+		copies, err := z.metadataPoolInfos(ctx, dstBucket, dstObject, dstOpts)
+		if err != nil {
+			return ObjectInfo{}, err
+		}
+		stored := mergedPoolObjectInfo(copies)
+		reconcileStoredObjectLock(srcInfo.UserDefined, storedObjectLockState(stored.UserDefined))
+		reconcileStoredObjectTags(srcInfo.UserDefined, stored.UserDefined)
+		idx := copies[0].Index
+		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)
+		}
+		return oi, err
+	}
+
 	poolIdx, err := z.getPoolIdxNoLock(ctx, dstBucket, dstObject, srcInfo.Size, dataMovementDstPool(dstOpts))
 	if err != nil {
 		return objInfo, err
@@ -1477,6 +1519,17 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
 		}
 	}
 
+	if !z.SinglePool() && dstOpts.ReplicaLockReconcile {
+		dstOpts.replicaObjectInfo = z.replicaObjectInfo
+		if !dstOpts.DataMovement {
+			defer func() {
+				if err == nil {
+					err = z.retireReplicaCopies(ctx, dstBucket, dstObject, poolIdx, objInfo)
+				}
+			}()
+		}
+	}
+
 	// CopyObjectHandler predicts the outcome of this decision in
 	// copyRewritesObjectData(); keep the two in sync.
 	if cpSrcDstSame && srcInfo.metadataOnly {
@@ -1511,6 +1564,8 @@ func (z *erasureServerPools) CopyObject(ctx context.Context, srcBucket, srcObjec
 		EncryptFn:                  dstOpts.EncryptFn,
 		WantChecksum:               dstOpts.WantChecksum,
 		WantServerSideChecksumType: dstOpts.WantServerSideChecksumType,
+		ReplicaLockReconcile:       dstOpts.ReplicaLockReconcile,
+		replicaObjectInfo:          dstOpts.replicaObjectInfo,
 	}
 
 	return z.serverPools[poolIdx].PutObject(ctx, dstBucket, dstObject, srcInfo.PutObjReader, putOpts)
@@ -2158,6 +2213,19 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
 		}
 	}()
 
+	if !z.SinglePool() {
+		if !opts.NoLock {
+			lk := z.NewNSLock(bucket, encodeDirObject(object))
+			lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
+			if err != nil {
+				return objInfo, err
+			}
+			ctx = lkctx.Context()
+			defer lk.Unlock(lkctx)
+		}
+		opts.NoLock = true
+	}
+
 	// Hold write locks to verify uploaded parts, also disallows any
 	// parallel PutObjectPart() requests.
 	uploadIDLock := z.NewNSLock(bucket, pathJoin(object, uploadID))
@@ -2172,13 +2240,21 @@ func (z *erasureServerPools) CompleteMultipartUpload(ctx context.Context, bucket
 		return z.serverPools[0].CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
 	}
 
+	if opts.ReplicaLockReconcile || opts.DataMovement {
+		opts.ReplicaLockReconcile = true
+		opts.replicaObjectInfo = z.replicaObjectInfo
+	}
+
 	for idx, pool := range z.serverPools {
 		if z.IsSuspended(idx) {
 			continue
 		}
 		objInfo, err = pool.CompleteMultipartUpload(ctx, bucket, object, uploadID, uploadedParts, opts)
 		if err == nil {
-			return objInfo, nil
+			if opts.ReplicaLockReconcile && !opts.DataMovement {
+				err = z.retireReplicaCopies(ctx, bucket, encodeDirObject(object), idx, objInfo)
+			}
+			return objInfo, err
 		}
 		if _, ok := err.(InvalidUploadID); ok {
 			// upload id not found move to next pool
@@ -2705,7 +2781,7 @@ func (z *erasureServerPools) HealObject(ctx context.Context, bucket, object, ver
 		wg.Add(1)
 		go func(idx int, pool *erasureSets) {
 			defer wg.Done()
-			result, err := pool.HealObject(ctx, bucket, object, versionID, opts)
+			result, err := z.healObjectInPool(ctx, pool.getHashedSet(object), bucket, object, versionID, opts)
 			result.Object = decodeDirObject(result.Object)
 			errs[idx] = err
 			results[idx] = result
@@ -2974,15 +3050,7 @@ func (z *erasureServerPools) PutObjectMetadata(ctx context.Context, bucket, obje
 		defer lk.Unlock(lkctx)
 	}
 
-	opts.MetadataChg = true
-	opts.NoLock = true
-	// We don't know the size here set 1GiB at least.
-	idx, err := z.getPoolIdxExistingWithOpts(ctx, bucket, object, opts)
-	if err != nil {
-		return ObjectInfo{}, err
-	}
-
-	return z.serverPools[idx].PutObjectMetadata(ctx, bucket, object, opts)
+	return z.updatePoolMetadata(ctx, bucket, object, opts)
 }
 
 // PutObjectTags - replace or add tags to an existing object
@@ -3003,44 +3071,31 @@ func (z *erasureServerPools) PutObjectTags(ctx context.Context, bucket, object s
 		defer lk.Unlock(lkctx)
 	}
 
-	opts.MetadataChg = true
-	opts.NoLock = true
-
-	// We don't know the size here set 1GiB at least.
-	idx, err := z.getPoolIdxExistingWithOpts(ctx, bucket, object, opts)
+	copies, err := z.metadataPoolInfos(ctx, bucket, object, opts)
 	if err != nil {
 		return ObjectInfo{}, err
 	}
-
-	return z.serverPools[idx].PutObjectTags(ctx, bucket, object, tags, opts)
+	opts.NoLock = true
+	opts.VersionID = copies[0].ObjInfo.VersionID
+	if opts.VersionID == "" {
+		opts.VersionID = nullVersionID
+	}
+	var primary ObjectInfo
+	for _, copy := range copies {
+		oi, err := z.serverPools[copy.Index].PutObjectTags(ctx, bucket, object, tags, opts)
+		if err != nil {
+			return ObjectInfo{}, err
+		}
+		if copy.Index == copies[0].Index {
+			primary = oi
+		}
+	}
+	return primary, nil
 }
 
 // DeleteObjectTags - delete object tags from an existing object
 func (z *erasureServerPools) DeleteObjectTags(ctx context.Context, bucket, object string, opts ObjectOptions) (ObjectInfo, error) {
-	object = encodeDirObject(object)
-	if z.SinglePool() {
-		return z.serverPools[0].DeleteObjectTags(ctx, bucket, object, opts)
-	}
-
-	if !opts.NoLock {
-		// Lock the object before deleting tags.
-		lk := z.NewNSLock(bucket, object)
-		lkctx, err := lk.GetLock(ctx, globalOperationTimeout)
-		if err != nil {
-			return ObjectInfo{}, err
-		}
-		ctx = lkctx.Context()
-		defer lk.Unlock(lkctx)
-	}
-
-	opts.MetadataChg = true
-	opts.NoLock = true
-	idx, err := z.getPoolIdxExistingWithOpts(ctx, bucket, object, opts)
-	if err != nil {
-		return ObjectInfo{}, err
-	}
-
-	return z.serverPools[idx].DeleteObjectTags(ctx, bucket, object, opts)
+	return z.PutObjectTags(ctx, bucket, object, "", opts)
 }
 
 // GetObjectTags - get object tags from an existing object
@@ -3050,7 +3105,7 @@ func (z *erasureServerPools) GetObjectTags(ctx context.Context, bucket, object s
 		return z.serverPools[0].GetObjectTags(ctx, bucket, object, opts)
 	}
 
-	oi, _, err := z.getLatestObjectInfoWithIdx(ctx, bucket, object, opts)
+	oi, err := z.GetObjectInfo(ctx, bucket, object, opts)
 	if err != nil {
 		return nil, err
 	}

cmd/global-heal.go

@@ -166,6 +166,12 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 	if objAPI == nil {
 		return errServerNotInitialized
 	}
+	healInPool := er.HealObject
+	if z, ok := objAPI.(*erasureServerPools); ok && !z.SinglePool() {
+		healInPool = func(ctx context.Context, bucket, object, versionID string, opts madmin.HealOpts) (madmin.HealResultItem, error) {
+			return z.healObjectInPool(ctx, er, bucket, object, versionID, opts)
+		}
+	}
 
 	started := tracker.Started
 	if started.IsZero() || started.Equal(timeSentinel) {
@@ -419,7 +425,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 			var result healEntryResult
 			fivs, err := entry.fileInfoVersions(bucket)
 			if err != nil {
-				res, err := er.HealObject(ctx, bucket, encodedEntryName, "",
+				res, err := healInPool(ctx, bucket, encodedEntryName, "",
 					madmin.HealOpts{
 						ScanMode: scanMode,
 						Remove:   healDeleteDangling,
@@ -455,7 +461,7 @@ func (er *erasureObjects) healErasureSet(ctx context.Context, buckets []string,
 					continue
 				}
 
-				res, err := er.HealObject(ctx, bucket, encodedEntryName,
+				res, err := healInPool(ctx, bucket, encodedEntryName,
 					version.VersionID, madmin.HealOpts{
 						ScanMode: scanMode,
 						Remove:   healDeleteDangling,

cmd/object-api-interface.go

@@ -99,10 +99,13 @@ type ObjectOptions struct {
 	ReplicationSourceTaggingTimestamp   time.Time // set if MinIOSourceTaggingTimestamp received
 	ReplicationSourceLegalholdTimestamp time.Time // set if MinIOSourceObjectLegalholdTimestamp received
 	ReplicationSourceRetentionTimestamp time.Time // set if MinIOSourceObjectRetentionTimestamp received
-	ReplicaLockReconcile                bool      // set for a trusted SSE-C replica full write/completion: re-order Object Lock against the destination version read under the write lock (single erasure set; see pgsty/silo#133)
+	ReplicaLockReconcile                bool      // re-order a trusted replica write against the destination version under the object write lock
 	DeletePrefix                        bool      // set true to enforce a prefix deletion, only application for DeleteObject API,
 	DeletePrefixObject                  bool      // set true when object's erasure set is resolvable by object name (using getHashedSetIndex)
 
+	// Pools-layer version resolver; the caller holds the shared object lock.
+	replicaObjectInfo GetObjectInfoFn
+
 	Speedtest bool // object call specifically meant for SpeedTest code, set to 'true' when invoked by SpeedtestHandler.
 
 	// Use the maximum parity (N/2), used when saving server configuration files
@@ -136,8 +139,8 @@ type ObjectOptions struct {
 	// when looking up a version by fi.VersionID
 	InclFreeVersions bool
 	// SkipFreeVersion skips adding a free version when a tiered version is
-	// being 'replaced'
-	// Note: Used only when a tiered object is being expired.
+	// being replaced. Used when expiring tiered content or retiring a copy
+	// whose tier reference is still owned by another copy.
 	SkipFreeVersion bool
 
 	MetadataChg           bool                  // is true if it is a metadata update operation.

cmd/object-handlers.go

@@ -1832,6 +1832,7 @@ func (api objectAPIHandlers) CopyObjectHandler(w http.ResponseWriter, r *http.Re
 		applyReplicatedObjectLock(srcInfo.UserDefined, storedLock, replicaTrusted,
 			retentionMode, retentionDate, legalHold,
 			dstOpts.ReplicationSourceRetentionTimestamp, dstOpts.ReplicationSourceLegalholdTimestamp)
+		dstOpts.ReplicaLockReconcile = replicaTrusted && dstOpts.VersionID != ""
 
 		if replicaTrusted {
 			// An SSE-C key rotation snapshots every stored reserved key into
@@ -2414,10 +2415,8 @@ func (api objectAPIHandlers) PutObjectHandler(w http.ResponseWriter, r *http.Req
 		// The decision above orders against the version as read here; let the
 		// object layer re-run it against the version read under the write lock
 		// that guards the replacement, so a newer hold or retention committed in
-		// between is not rolled back (issue #120). Scoped to the SSE-C replica
-		// retransmit this issue enables, keyed on the incoming write's restored
-		// SSE-C seal, the same predicate as the duplicate-version exemption.
-		opts.ReplicaLockReconcile = isReplicaTrusted(ctx) && opts.VersionID != "" && crypto.SSEC.IsEncrypted(metadata)
+		// between is not rolled back, across all pools and encryption modes.
+		opts.ReplicaLockReconcile = isReplicaTrusted(ctx) && opts.VersionID != ""
 	}
 	if s3Err != ErrNone {
 		writeErrorResponse(ctx, w, errorCodes.ToAPIErr(s3Err), r.URL)

docs/investigations/issue-116/evidence-20260911.json

@@ -0,0 +1,534 @@
+{
+  "date": "2026-09-11",
+  "runs": [
+    {
+      "runid": "silo-v1-0806-9c53f14c",
+      "version": "0806",
+      "image": "sha256:29a498b24669cae1fed11c1a2fb2b3d73c68829a0a9c0b14e71b386671d38fac",
+      "nodes": 4,
+      "drives": 4,
+      "filesystem": "Linux tmpfs named volumes, held mounted across server restarts",
+      "drive_bytes": 268435456,
+      "status": "PASS",
+      "phases": [
+        {
+          "phase": "startup-admin",
+          "seconds_from_start": 4.157,
+          "first_online_by_coordinator": [
+            4.037,
+            4.078,
+            4.118,
+            3.737
+          ]
+        },
+        {
+          "phase": "startup-canary",
+          "attempts": 1,
+          "puts": 4,
+          "gets": 16,
+          "hard_deadline_seconds": 60,
+          "gate_seconds": 0.239,
+          "seconds_from_start": 4.396,
+          "first_put_seconds_after_admin_by_coordinator": [
+            0.096,
+            0.116,
+            0.127,
+            0.138
+          ],
+          "first_four_reads_seconds_after_admin_by_coordinator": [
+            0.162,
+            0.194,
+            0.217,
+            0.239
+          ],
+          "transient_error_count": 0,
+          "first_transient_errors": []
+        },
+        {
+          "phase": "full-restart-admin",
+          "seconds_from_start": 2.084,
+          "first_online_by_coordinator": [
+            1.937,
+            1.976,
+            2.016,
+            1.643
+          ]
+        },
+        {
+          "phase": "full-restart-canary",
+          "attempts": 45,
+          "puts": 4,
+          "gets": 16,
+          "hard_deadline_seconds": 60,
+          "gate_seconds": 14.449,
+          "seconds_from_start": 16.534,
+          "first_put_seconds_after_admin_by_coordinator": [
+            14.315,
+            13.573,
+            13.953,
+            0.05
+          ],
+          "first_four_reads_seconds_after_admin_by_coordinator": [
+            14.373,
+            14.396,
+            14.426,
+            14.449
+          ],
+          "transient_error_count": 129,
+          "first_transient_errors": [
+            {
+              "attempt": 1,
+              "operation": "put",
+              "node": 0,
+              "error": "An error occurred (SlowDownWrite) when calling the PutObject operation (reached max retries: 0): Resource requested is unwritable, please reduce your request rate"
+            },
+            {
+              "attempt": 1,
+              "operation": "put",
+              "node": 1,
+              "error": "An error occurred (SlowDownWrite) when calling the PutObject operation (reached max retries: 0): Resource requested is unwritable, please reduce your request rate"
+            },
+            {
+              "attempt": 1,
+              "operation": "put",
+              "node": 2,
+              "error": "An error occurred (SlowDownWrite) when calling the PutObject operation (reached max retries: 0): Resource requested is unwritable, please reduce your request rate"
+            },
+            {
+              "attempt": 2,
+              "operation": "put",
+              "node": 0,
+              "error": "An error occurred (SlowDownWrite) when calling the PutObject operation (reached max retries: 0): Resource requested is unwritable, please reduce your request rate"
+            }
+          ]
+        },
+        {
+          "phase": "readback-15s",
+          "seconds_after_canary": 17.596,
+          "objects": 55,
+          "reads": 220,
+          "errors": []
+        },
+        {
+          "phase": "readback-30s",
+          "seconds_after_canary": 31.357,
+          "objects": 55,
+          "reads": 220,
+          "errors": []
+        },
+        {
+          "phase": "readback-60s",
+          "seconds_after_canary": 61.566,
+          "objects": 55,
+          "reads": 220,
+          "errors": []
+        },
+        {
+          "phase": "one-node-outage",
+          "existing_read": true,
+          "put": true,
+          "readers": 3
+        },
+        {
+          "phase": "rejoin-admin",
+          "seconds_from_start": 1.382,
+          "first_online_by_coordinator": [
+            1.27,
+            1.307,
+            1.343,
+            1.382
+          ]
+        },
+        {
+          "phase": "rejoin-canary",
+          "attempts": 1,
+          "puts": 4,
+          "gets": 16,
+          "hard_deadline_seconds": 60,
+          "gate_seconds": 0.158,
+          "seconds_from_start": 1.541,
+          "first_put_seconds_after_admin_by_coordinator": [
+            0.013,
+            0.029,
+            0.039,
+            0.05
+          ],
+          "first_four_reads_seconds_after_admin_by_coordinator": [
+            0.077,
+            0.102,
+            0.132,
+            0.158
+          ],
+          "transient_error_count": 0,
+          "first_transient_errors": []
+        },
+        {
+          "phase": "final-readback",
+          "seconds_after_canary": 1.71,
+          "objects": 60,
+          "reads": 240,
+          "errors": []
+        }
+      ],
+      "acknowledged_objects": 60,
+      "version_ids_recorded": true,
+      "sha256_recorded": true
+    },
+    {
+      "runid": "silo-v1-0903-a457573c",
+      "version": "0903",
+      "image": "sha256:b616a0cf8cb281e7e6bb3c9b1fb53875b4016a2878223925541c18f82d6c5ca3",
+      "nodes": 4,
+      "drives": 4,
+      "filesystem": "Linux tmpfs named volumes, held mounted across server restarts",
+      "drive_bytes": 268435456,
+      "status": "PASS",
+      "phases": [
+        {
+          "phase": "startup-admin",
+          "seconds_from_start": 4.271,
+          "first_online_by_coordinator": [
+            4.158,
+            4.194,
+            3.827,
+            3.867
+          ]
+        },
+        {
+          "phase": "startup-canary",
+          "attempts": 1,
+          "puts": 4,
+          "gets": 16,
+          "hard_deadline_seconds": 60,
+          "gate_seconds": 0.2,
+          "seconds_from_start": 4.471,
+          "first_put_seconds_after_admin_by_coordinator": [
+            0.054,
+            0.068,
+            0.081,
+            0.092
+          ],
+          "first_four_reads_seconds_after_admin_by_coordinator": [
+            0.118,
+            0.146,
+            0.167,
+            0.2
+          ],
+          "transient_error_count": 0,
+          "first_transient_errors": []
+        },
+        {
+          "phase": "full-restart-admin",
+          "seconds_from_start": 2.284,
+          "first_online_by_coordinator": [
+            2.143,
+            2.193,
+            0.849,
+            0.902
+          ]
+        },
+        {
+          "phase": "full-restart-canary",
+          "attempts": 1,
+          "puts": 4,
+          "gets": 16,
+          "hard_deadline_seconds": 60,
+          "gate_seconds": 0.191,
+          "seconds_from_start": 2.476,
+          "first_put_seconds_after_admin_by_coordinator": [
+            0.016,
+            0.029,
+            0.044,
+            0.057
+          ],
+          "first_four_reads_seconds_after_admin_by_coordinator": [
+            0.093,
+            0.124,
+            0.158,
+            0.192
+          ],
+          "transient_error_count": 0,
+          "first_transient_errors": []
+        },
+        {
+          "phase": "readback-15s",
+          "seconds_after_canary": 15.235,
+          "objects": 8,
+          "reads": 32,
+          "errors": []
+        },
+        {
+          "phase": "readback-30s",
+          "seconds_after_canary": 30.356,
+          "objects": 8,
+          "reads": 32,
+          "errors": []
+        },
+        {
+          "phase": "readback-60s",
+          "seconds_after_canary": 60.192,
+          "objects": 8,
+          "reads": 32,
+          "errors": []
+        },
+        {
+          "phase": "one-node-outage",
+          "existing_read": true,
+          "put": true,
+          "readers": 3
+        },
+        {
+          "phase": "rejoin-admin",
+          "seconds_from_start": 0.903,
+          "first_online_by_coordinator": [
+            0.791,
+            0.829,
+            0.865,
+            0.903
+          ]
+        },
+        {
+          "phase": "rejoin-canary",
+          "attempts": 1,
+          "puts": 4,
+          "gets": 16,
+          "hard_deadline_seconds": 60,
+          "gate_seconds": 0.141,
+          "seconds_from_start": 1.044,
+          "first_put_seconds_after_admin_by_coordinator": [
+            0.012,
+            0.022,
+            0.035,
+            0.045
+          ],
+          "first_four_reads_seconds_after_admin_by_coordinator": [
+            0.069,
+            0.096,
+            0.12,
+            0.141
+          ],
+          "transient_error_count": 0,
+          "first_transient_errors": []
+        },
+        {
+          "phase": "final-readback",
+          "seconds_after_canary": 0.301,
+          "objects": 13,
+          "reads": 52,
+          "errors": []
+        }
+      ],
+      "acknowledged_objects": 13,
+      "version_ids_recorded": true,
+      "sha256_recorded": true
+    },
+    {
+      "runid": "silo-v1-current-6a14eb18",
+      "version": "current",
+      "image": "pgsty/d12a:build",
+      "image_id": "sha256:307af7711e2e04ab75759cb42a1eef45c43c4404894c0e30dd19f742b107b922",
+      "platform": "linux/arm64",
+      "nodes": 4,
+      "drives": 4,
+      "filesystem": "Linux tmpfs named volumes, held mounted across server restarts",
+      "drive_bytes": 268435456,
+      "status": "PASS",
+      "binary_sha256": "1e4cd7b78ecfa1b0cf28f1961d48a220a712c6be1fd3a01b60ee99aec62f04a4",
+      "source_revision": "b32f2d9dd01a383a9991d34ffe248063463a031d+pool-consistency",
+      "phases": [
+        {
+          "phase": "startup-admin",
+          "seconds_from_start": 7.374,
+          "first_online_by_coordinator": [
+            7.207,
+            6.677,
+            6.74,
+            6.8
+          ]
+        },
+        {
+          "phase": "startup-canary",
+          "attempts": 1,
+          "puts": 4,
+          "gets": 16,
+          "hard_deadline_seconds": 60,
+          "gate_seconds": 0.479,
+          "seconds_from_start": 7.854,
+          "first_put_seconds_after_admin_by_coordinator": [
+            0.266,
+            0.279,
+            0.292,
+            0.305
+          ],
+          "first_four_reads_seconds_after_admin_by_coordinator": [
+            0.336,
+            0.378,
+            0.431,
+            0.479
+          ],
+          "transient_error_count": 0,
+          "first_transient_errors": []
+        },
+        {
+          "phase": "full-restart-admin",
+          "seconds_from_start": 2.461,
+          "first_online_by_coordinator": [
+            2.264,
+            1.414,
+            2.396,
+            2.461
+          ]
+        },
+        {
+          "phase": "full-restart-canary",
+          "attempts": 36,
+          "puts": 4,
+          "gets": 16,
+          "hard_deadline_seconds": 60,
+          "gate_seconds": 14.489,
+          "seconds_from_start": 16.951,
+          "first_put_seconds_after_admin_by_coordinator": [
+            0.013,
+            0.025,
+            14.301,
+            0.046
+          ],
+          "first_four_reads_seconds_after_admin_by_coordinator": [
+            14.358,
+            14.394,
+            14.435,
+            14.489
+          ],
+          "transient_error_count": 104,
+          "first_transient_errors": [
+            {
+              "attempt": 1,
+              "operation": "put",
+              "node": 2,
+              "error": "An error occurred (SlowDownWrite) when calling the PutObject operation (reached max retries: 0): Resource requested is unwritable, please reduce your request rate"
+            },
+            {
+              "attempt": 1,
+              "operation": "get",
+              "node": 2,
+              "key": "full-restart-attempt-1-node-0",
+              "error": "An error occurred (SlowDownRead) when calling the GetObject operation (reached max retries: 0): Resource requested is unreadable, please reduce your request rate"
+            },
+            {
+              "attempt": 1,
+              "operation": "get",
+              "node": 2,
+              "key": "full-restart-attempt-1-node-3",
+              "error": "An error occurred (SlowDownRead) when calling the GetObject operation (reached max retries: 0): Resource requested is unreadable, please reduce your request rate"
+            },
+            {
+              "attempt": 2,
+              "operation": "put",
+              "node": 2,
+              "error": "An error occurred (SlowDownWrite) when calling the PutObject operation (reached max retries: 0): Resource requested is unwritable, please reduce your request rate"
+            }
+          ]
+        },
+        {
+          "phase": "readback-15s",
+          "started_seconds_after_canary": 15.0,
+          "seconds_after_canary": 22.006,
+          "objects": 113,
+          "reads": 452,
+          "errors": []
+        },
+        {
+          "phase": "readback-30s",
+          "started_seconds_after_canary": 30.003,
+          "seconds_after_canary": 36.571,
+          "objects": 113,
+          "reads": 452,
+          "errors": []
+        },
+        {
+          "phase": "readback-60s",
+          "started_seconds_after_canary": 60.0,
+          "seconds_after_canary": 65.743,
+          "objects": 113,
+          "reads": 452,
+          "errors": []
+        },
+        {
+          "phase": "one-node-outage",
+          "existing_read": true,
+          "put": true,
+          "readers": 3
+        },
+        {
+          "phase": "rejoin-admin",
+          "seconds_from_start": 2.49,
+          "first_online_by_coordinator": [
+            1.165,
+            2.417,
+            1.646,
+            2.087
+          ]
+        },
+        {
+          "phase": "rejoin-canary",
+          "attempts": 37,
+          "puts": 4,
+          "gets": 16,
+          "hard_deadline_seconds": 60,
+          "gate_seconds": 13.977,
+          "seconds_from_start": 16.468,
+          "first_put_seconds_after_admin_by_coordinator": [
+            0.013,
+            0.024,
+            0.035,
+            13.889
+          ],
+          "first_four_reads_seconds_after_admin_by_coordinator": [
+            13.909,
+            13.927,
+            13.953,
+            13.977
+          ],
+          "transient_error_count": 36,
+          "first_transient_errors": [
+            {
+              "attempt": 1,
+              "operation": "put",
+              "node": 3,
+              "error": "An error occurred (SlowDownWrite) when calling the PutObject operation (reached max retries: 0): Resource requested is unwritable, please reduce your request rate"
+            },
+            {
+              "attempt": 2,
+              "operation": "put",
+              "node": 3,
+              "error": "An error occurred (SlowDownWrite) when calling the PutObject operation (reached max retries: 0): Resource requested is unwritable, please reduce your request rate"
+            },
+            {
+              "attempt": 3,
+              "operation": "put",
+              "node": 3,
+              "error": "An error occurred (SlowDownWrite) when calling the PutObject operation (reached max retries: 0): Resource requested is unwritable, please reduce your request rate"
+            },
+            {
+              "attempt": 4,
+              "operation": "put",
+              "node": 3,
+              "error": "An error occurred (SlowDownWrite) when calling the PutObject operation (reached max retries: 0): Resource requested is unwritable, please reduce your request rate"
+            }
+          ]
+        },
+        {
+          "phase": "final-readback",
+          "started_seconds_after_canary": 0.0,
+          "seconds_after_canary": 6.507,
+          "objects": 226,
+          "reads": 904,
+          "errors": []
+        }
+      ],
+      "acknowledged_objects": 226,
+      "version_ids_recorded": true,
+      "sha256_recorded": true
+    }
+  ]
+}

docs/investigations/issue-116/run-linux.py

@@ -0,0 +1,327 @@
+#!/usr/bin/env python3
+"""Bounded four-container restart/readback acceptance for pgsty/silo#116."""
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import secrets
+import shutil
+import signal
+import socket
+import subprocess
+import time
+import uuid
+from concurrent.futures import ThreadPoolExecutor
+
+import boto3
+from botocore.config import Config
+
+ROOT = None
+MCLI = None
+CURRENT_BINARY = None
+SOURCE_REVISION = None
+IMAGES = {
+    '0806': 'pgsty/silo:RELEASE.2026-08-06T00-00-00Z',
+    '0903': 'pgsty/silo:RELEASE.2026-09-03T13-18-01Z',
+    'current': 'pgsty/d12a:build',
+}
+
+
+def docker(*args, timeout=40, check=True):
+    return subprocess.run(['docker', *args], check=check, capture_output=True, text=True, timeout=timeout)
+
+
+class Deadline(BaseException):
+    pass
+
+
+def bounded(seconds, operation):
+    def expired(*_):
+        raise Deadline(f'hard deadline of {seconds}s exceeded')
+    old = signal.signal(signal.SIGALRM, expired)
+    signal.setitimer(signal.ITIMER_REAL, seconds)
+    try:
+        return operation(time.monotonic() + seconds)
+    finally:
+        signal.setitimer(signal.ITIMER_REAL, 0)
+        signal.signal(signal.SIGALRM, old)
+
+
+def remaining(deadline):
+    value = deadline - time.monotonic()
+    if value <= 0:
+        raise Deadline('absolute deadline exceeded')
+    return value
+
+
+def run(version):
+    image_info = json.loads(docker('image', 'inspect', IMAGES[version]).stdout)[0]
+    runid = f'silo-v1-{version}-{uuid.uuid4().hex[:8]}'
+    out = ROOT / runid
+    out.mkdir(mode=0o700)
+    user, password = 'local116', secrets.token_urlsafe(24)
+    envfile = out / 'credentials.env'
+    envfile.write_text(f'MINIO_ROOT_USER={user}\nMINIO_ROOT_PASSWORD={password}\nMINIO_CI_CD=1\nMINIO_BROWSER=off\nGOMAXPROCS=2\n')
+    envfile.chmod(0o600)
+    env = {k: v for k, v in os.environ.items() if not k.startswith(('MINIO_', 'SILO_', 'MC_'))
+           and k.lower() not in {'http_proxy', 'https_proxy', 'all_proxy', 'no_proxy'}}
+    nodes = [f'{runid}-n{i}' for i in range(4)]
+    sockets = [socket.socket() for _ in nodes]
+    for sock in sockets:
+        sock.bind(('127.0.0.1', 0))
+    ports = [sock.getsockname()[1] for sock in sockets]
+    for sock in sockets:
+        sock.close()
+    volumes = [n + '-data' for n in nodes]
+    endpoints, ledger = [], []
+    result = {'runid': runid, 'version': version, 'image': IMAGES[version],
+              'image_id': image_info['Id'], 'platform': image_info['Os']+'/'+image_info['Architecture'],
+              'nodes': 4,
+              'drives': 4, 'filesystem': 'Linux tmpfs named volumes, held mounted across server restarts',
+              'drive_bytes': 268435456, 'phases': [], 'status': 'RUNNING'}
+    if version == 'current':
+        result['binary_sha256'] = hashlib.sha256(CURRENT_BINARY.read_bytes()).hexdigest()
+        result['source_revision'] = SOURCE_REVISION
+    bucket = 'canary-' + uuid.uuid4().hex[:10]
+
+    def save(event=None):
+        if event is not None:
+            result['phases'].append(event)
+            compact = {k: (len(v) if k in ('transient_errors', 'errors') else v) for k, v in event.items()}
+            print(json.dumps({'version': version, **compact}), flush=True)
+        (out / 'result.json').write_text(json.dumps(result, indent=2) + '\n')
+        (out / 'acknowledged.json').write_text(json.dumps(ledger, indent=2) + '\n')
+
+    def parallel(fn, values):
+        with ThreadPoolExecutor(max_workers=4) as pool:
+            return list(pool.map(fn, values))
+
+    def client(i, deadline):
+        timeout = remaining(deadline)
+        return boto3.client('s3', endpoint_url=endpoints[i], aws_access_key_id=user,
+                            aws_secret_access_key=password, region_name='us-east-1',
+                            config=Config(proxies={}, signature_version='s3v4', s3={'addressing_style': 'path'},
+                                          retries={'total_max_attempts': 1}, connect_timeout=timeout,
+                                          read_timeout=timeout, request_checksum_calculation='when_required',
+                                          response_checksum_validation='when_required'))
+
+    def admin_gate(origin, phase):
+        def check(deadline):
+            first = [None] * 4
+            while True:
+                states = []
+                for i in range(4):
+                    try:
+                        p = subprocess.run([str(MCLI), '--config-dir', str(out / 'mcli'), '--json',
+                                            'admin', 'info', f'n{i}'], env=env, capture_output=True,
+                                           text=True, timeout=min(5, remaining(deadline)))
+                        info = json.loads(p.stdout)['info']
+                        servers = info['servers']
+                        ok = len(servers) == 4 and all(s.get('state') == 'online' and s.get('drives')
+                             and all(d.get('state') == 'ok' for d in s['drives'])
+                             and all(v == 'online' for v in s.get('network', {}).values()) for s in servers)
+                        if ok:
+                            (out / f'{phase}-admin-{i}.json').write_text(json.dumps(info, indent=2) + '\n')
+                            if first[i] is None:
+                                first[i] = round(time.monotonic() - origin, 3)
+                        states.append(ok)
+                    except (Exception,):
+                        states.append(False)
+                if all(states):
+                    event = {'phase': phase + '-admin', 'seconds_from_start': round(time.monotonic()-origin, 3),
+                             'first_online_by_coordinator': first}
+                    save(event)
+                    return time.monotonic()
+                time.sleep(min(.25, remaining(deadline)))
+        return bounded(90, check)
+
+    def canary(phase, origin, admin_time, setup=False):
+        def check(deadline):
+            started = time.monotonic()
+            first_put, first_reads = [None] * 4, [None] * 4
+            errors, attempt, setup_done = [], 0, not setup
+            result['active_canary'] = {'phase': phase, 'errors': errors}
+            while True:
+                attempt += 1
+                remaining(deadline)
+                if not setup_done:
+                    try:
+                        try:
+                            client(0, deadline).create_bucket(Bucket=bucket)
+                        except Exception as e:
+                            if 'BucketAlreadyOwnedByYou' not in str(e):
+                                raise
+                        client(0, deadline).put_bucket_versioning(Bucket=bucket, VersioningConfiguration={'Status': 'Enabled'})
+                        setup_done = True
+                    except Exception as e:
+                        errors.append({'attempt': attempt, 'operation': 'setup', 'error': str(e)[:250]})
+                        time.sleep(min(.25, remaining(deadline)))
+                        continue
+                acked = []
+                for i in range(4):
+                    key = f'{phase}-attempt-{attempt}-node-{i}'
+                    payload = (key + '\n').encode() * 16384
+                    try:
+                        vid = client(i, deadline).put_object(Bucket=bucket, Key=key, Body=payload)['VersionId']
+                        entry = {'phase': phase, 'key': key, 'version': vid, 'bytes': len(payload),
+                                 'sha256': hashlib.sha256(payload).hexdigest(), 'writer': i,
+                                 'ack_seconds_from_start': round(time.monotonic()-origin, 3)}
+                        ledger.append(entry)
+                        save()  # Persist every acknowledged write, including failed rounds.
+                        acked.append(entry)
+                        if first_put[i] is None:
+                            first_put[i] = round(time.monotonic()-admin_time, 3)
+                    except Exception as e:
+                        errors.append({'attempt': attempt, 'operation': 'put', 'node': i, 'error': str(e)[:250]})
+                reads = 0
+                for i in range(4):
+                    own_reads = 0
+                    for entry in acked:
+                        try:
+                            verify(i, entry, deadline)
+                            reads += 1
+                            own_reads += 1
+                        except Exception as e:
+                            errors.append({'attempt': attempt, 'operation': 'get', 'node': i,
+                                           'key': entry['key'], 'error': str(e)[:250]})
+                    if own_reads == 4 and first_reads[i] is None:
+                        first_reads[i] = round(time.monotonic()-admin_time, 3)
+                remaining(deadline)
+                if len(acked) == 4 and reads == 16:
+                    result.pop('active_canary', None)
+                    save({'phase': phase + '-canary', 'attempts': attempt, 'puts': 4, 'gets': 16,
+                          'hard_deadline_seconds': 60, 'gate_seconds': round(time.monotonic()-started, 3),
+                          'seconds_from_start': round(time.monotonic()-origin, 3),
+                          'first_put_seconds_after_admin_by_coordinator': first_put,
+                          'first_four_reads_seconds_after_admin_by_coordinator': first_reads, 'transient_errors': errors})
+                    return time.monotonic()
+                (out / f'{phase}-canary-errors.json').write_text(json.dumps(errors, indent=2) + '\n')
+                if attempt == 1:
+                    print(json.dumps({'version': version, 'phase': phase, 'first_attempt_errors': errors}), flush=True)
+                time.sleep(min(.25, remaining(deadline)))
+        return bounded(60, check)
+
+    def verify(i, entry, deadline):
+        got = client(i, deadline).get_object(Bucket=bucket, Key=entry['key'], VersionId=entry['version'])
+        try:
+            data = got['Body'].read()
+        finally:
+            got['Body'].close()
+        assert len(data) == entry['bytes'] and hashlib.sha256(data).hexdigest() == entry['sha256'], entry['key']
+        assert got.get('VersionId') == entry['version'], entry['key']
+        remaining(deadline)
+
+    def readback(label, origin):
+        def check(deadline):
+            started = time.monotonic()
+            errors = []
+            for entry in ledger:
+                for i in range(4):
+                    try:
+                        verify(i, entry, deadline)
+                    except Exception as e:
+                        errors.append({'key': entry['key'], 'node': i, 'error': str(e)[:250]})
+            save({'phase': label, 'started_seconds_after_canary': round(started-origin, 3),
+                  'seconds_after_canary': round(time.monotonic()-origin, 3),
+                  'objects': len(ledger), 'reads': len(ledger)*4, 'errors': errors})
+            return errors
+        return bounded(60, check)
+
+    try:
+        docker('network', 'create', runid)
+        for volume in volumes:
+            docker('volume', 'create', '--driver', 'local', '--opt', 'type=tmpfs',
+                   '--opt', 'device=tmpfs', '--opt', 'o=size=256m', volume)
+        mounts = [arg for i, v in enumerate(volumes) for arg in ('--mount', f'type=volume,source={v},target=/keep/{i}')]
+        docker('run', '-d', '--pull=never', '--network', 'none', '--name', runid + '-keeper',
+               *mounts, '--entrypoint', 'sleep', 'alpine:3.23', '1800')
+        urls = [f'http://{n}:9000/data' for n in nodes]
+        def create(i):
+            extra = []
+            if version == 'current':
+                extra = ['--mount', f'type=bind,source={CURRENT_BINARY},target=/lab/silo,readonly',
+                         '--entrypoint', '/lab/silo']
+            docker('create', '--pull=never', '--name', nodes[i], '--network', runid,
+                   '--hostname', nodes[i], '--cpus', '2', '--memory', '3g', '--env-file', str(envfile),
+                   '--mount', f'type=volume,source={volumes[i]},target=/data',
+                   '-p', f'127.0.0.1:{ports[i]}:9000', *extra, image_info['Id'], 'server', '--address', ':9000',
+                   '--console-address', ':9001', *urls)
+        parallel(create, range(4))
+        start = time.monotonic()
+        parallel(lambda n: docker('start', n), nodes)
+        for i, n in enumerate(nodes):
+            endpoint = 'http://' + docker('port', n, '9000/tcp').stdout.strip()
+            endpoints.append(endpoint)
+            env[f'MC_HOST_n{i}'] = endpoint.replace('http://', f'http://{user}:{password}@')
+        admin = admin_gate(start, 'startup')
+        canary('startup', start, admin, setup=True)
+        parallel(lambda n: docker('stop', '-t', '10', n), nodes)
+        start = time.monotonic()
+        parallel(lambda n: docker('start', n), nodes)
+        admin = admin_gate(start, 'full-restart')
+        gate = canary('full-restart', start, admin)
+        read_errors = []
+        for after in (15, 30, 60):
+            time.sleep(max(0, gate + after - time.monotonic()))
+            read_errors.extend(readback(f'readback-{after}s', gate))
+        assert not read_errors, 'acknowledged object readback failure; see result.json'
+        docker('stop', '-t', '10', nodes[3])
+        def outage(deadline):
+            verify(0, ledger[0], deadline)
+            payload = b'acknowledged with one Linux node offline' * 16384
+            key = 'one-node-outage'
+            vid = client(0, deadline).put_object(Bucket=bucket, Key=key, Body=payload)['VersionId']
+            entry = {'phase': 'outage', 'key': key, 'version': vid, 'bytes': len(payload),
+                     'sha256': hashlib.sha256(payload).hexdigest(), 'writer': 0}
+            ledger.append(entry)
+            save()
+            for i in range(3):
+                verify(i, entry, deadline)
+            save({'phase': 'one-node-outage', 'existing_read': True, 'put': True, 'readers': 3})
+        bounded(60, outage)
+        start = time.monotonic()
+        docker('start', nodes[3])
+        admin = admin_gate(start, 'rejoin')
+        gate = canary('rejoin', start, admin)
+        assert not readback('final-readback', gate)
+        result['status'] = 'PASS'
+    except BaseException as e:
+        result['status'] = 'FAIL'
+        result['error'] = f'{type(e).__name__}: {e}'
+        raise
+    finally:
+        save()
+        for n in nodes:
+            log = docker('logs', n, check=False)
+            (out / (n + '.log')).write_text(log.stdout + log.stderr)
+            docker('rm', '-f', n, check=False)
+        docker('rm', '-f', runid + '-keeper', check=False)
+        for v in volumes:
+            docker('volume', 'rm', v, check=False)
+        docker('network', 'rm', runid, check=False)
+        envfile.unlink(missing_ok=True)
+        print(json.dumps({'version': version, 'status': result['status'], 'evidence': str(out)}), flush=True)
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('versions', nargs='+', choices=list(IMAGES))
+    parser.add_argument('--output', type=Path, required=True, help='directory for retained evidence')
+    parser.add_argument('--mcli', default=shutil.which('mcli'), help='native mcli executable')
+    parser.add_argument('--current-binary', type=Path, help='Linux binary matching the Docker architecture')
+    parser.add_argument('--source-revision', help='Git revision of --current-binary')
+    parser.add_argument('--current-image', default=IMAGES['current'], help='cached Linux base image for current binary')
+    args = parser.parse_args()
+    if not args.mcli or not Path(args.mcli).is_file():
+        parser.error('--mcli must point to an executable file')
+    if 'current' in args.versions and (not args.current_binary or not args.current_binary.is_file()):
+        parser.error('current requires --current-binary')
+    ROOT = args.output.resolve()
+    ROOT.mkdir(mode=0o700, parents=True, exist_ok=True)
+    MCLI = Path(args.mcli).resolve()
+    CURRENT_BINARY = args.current_binary.resolve() if args.current_binary else None
+    SOURCE_REVISION = args.source_revision
+    IMAGES['current'] = args.current_image
+    for version in args.versions:
+        run(version)

docs/investigations/release-readiness-20260911.md

@@ -0,0 +1,143 @@
+# Remaining release correctness work, 2026-09-11
+
+This records the three work items agreed after the branch/PR consolidation:
+OIDC #154, Linux restart/readback #116, and the related multi-pool defects
+#133/#144. The maintained target is SILO with the PGSTY Console, mcli and
+silo-pkg dependencies in the repository's current `go.mod`.
+
+## Multi-pool writes and conditional deletion (#133, #144)
+
+The pools layer now holds its object write lock across PUT and multipart
+completion, as it already does for metadata updates and DELETE. Multipart
+completion acquires the object lock before the upload lock. Queued healing and
+drive healing use this namespace too; healing under a caller-owned lock retains
+that lock's cancellation context. The destination-pool allocation policy is
+unchanged.
+
+Trusted replica writes resolve the addressed version in every pool, including
+draining and rebalancing pools. Retention, legal hold and tags retain their
+independent ordering timestamps. A timestamp-only removal survives a stale
+retransmit. Completion resolves the version persisted in the upload, including
+the null version, rather than taking a later latest version's metadata.
+
+Successful replica replacement retires competing copies of that exact version,
+so an equal-ModTime copy in an earlier pool cannot shadow the reconciled result.
+Metadata updates evaluate their callback once against the merged version and
+update all its copies. Replica metadata COPY rechecks ordering under the lock.
+Cleanup failures propagate; a replacement may already have committed when
+cleanup fails, and a retry can finish cleanup. Data movement keeps ownership of
+its source cleanup.
+
+Retiring copies preserves any remote-tier reference still held by a surviving
+copy, including temporarily restored objects. Only the last copy of that tier
+reference schedules remote contents for garbage collection. This also protects
+the authoritative copy if the final conditional deletion fails; unrelated tier
+contents remain eligible for cleanup.
+
+Conditional DELETE evaluates its precondition against the logical latest or
+explicitly addressed version. It checks all pools before mutation, removes
+secondary copies before the authoritative one, and returns cleanup errors.
+Versioned DELETE without a version ID creates a delete marker and preserves
+version history. Retention and replication callbacks evaluate the reconciled
+logical version. The existing all-pool delete helper also propagates errors
+from non-first pools.
+
+The deterministic two-pool, 32-drive fixtures cover version selection,
+duplicate removal, delete failure propagation, PUT/DELETE and completion/DELETE
+interleavings, independent lock winners, draining/rebalancing owners, null
+versions, metadata COPY, metadata/healing serialization and cleanup retry.
+Tiered-copy tests inspect the persisted garbage-collection markers after
+metadata COPY, restored COPY, successful deletion and failed primary deletion,
+with distinct remote references as cleanup controls.
+The original branch reproduced the wrong-version lookup, surviving duplicate,
+suppressed delete error, PUT/DELETE race, and PUT/completion lock-state failures
+before the fixes were applied.
+
+Before the final tier-reference guard, local validation passed the full `cmd`
+suite (255.720 s), all `internal` tests, `go vet ./...`, generated-file checks
+and the branding/entrypoint checks.
+Focused race checks cover the pooled interleavings, SSE-C lock regressions,
+conditional deletion, access-tier movement and TLS defaults. The two-pool and
+related replica/delete/movement tests also passed as a Linux/arm64 test binary
+in an isolated container with an 8 GiB `/tmp` tmpfs. Its initial 1 GiB tmpfs
+was insufficient for the existing single-drive test fixtures' free-space guard.
+The final tier-reference guard passed all six new cases in that Linux fixture;
+the retained restart binary below predates this guard and has no remote tier
+configured.
+
+## Linux restart/readback (#116)
+
+The [runner](issue-116/run-linux.py) creates four Linux/arm64 server containers
+on one Docker Desktop Linux VM, with separate network identities and one drive
+per node (EC 2+2). A holder container keeps four 256 MiB Linux tmpfs named
+volumes mounted across full server stops. Docker's ordinary filesystem had
+only 3.8 GiB free out of 2 TiB and correctly hit the server's free-space limit;
+the test uses the separate tmpfs filesystems without changing that threshold.
+This is the four-containers-on-one-Linux-host option explicitly accepted in
+the [#116 V1 plan](https://github.com/pgsty/silo/issues/116).
+
+This covers process/container restart and TCP peer reconnection on one Linux
+host. It does not establish independent-host, host-reboot or physical-media
+durability. Every resource created by the runner is removed in its cleanup.
+The [retained summary](issue-116/evidence-20260911.json) records image identities,
+per-coordinator observations and readback counts.
+
+| Binary | Full restart to all admin views online | Admin gate to complete 4 PUT / 16 GET round | Timed readbacks | Final readback |
+|---|---:|---:|---:|---:|
+| 0806 release | 2.084 s | 14.449 s | 660 / 660 | 240 / 240 |
+| 0903 release | 2.284 s | 0.191 s | 96 / 96 | 52 / 52 |
+| Current fix candidate | 2.461 s | 14.489 s | 1356 / 1356 | 904 / 904 |
+
+Each canary has one hard 60-second deadline covering setup, requests, response
+body reads and sleeps, with SDK retries disabled. Every acknowledged PUT uses
+a unique versioned key and immediately records its VersionId, size and SHA-256.
+The timed checks reread these same objects through all four coordinators at
+15, 30 and 60 seconds **after the data canary succeeds**. They do not replace
+early acknowledgements with later writes. The final check also includes writes
+made during the one-node outage and subsequent rejoin canary.
+
+All three runs passed the bounded canary, scheduled readbacks, one-node-outage
+read/write and rejoin readback. The candidate also showed a rejoin window:
+admin online at 2.490 s, complete data canary 13.977 s later. These are individual
+observations, not a latency guarantee or a comparison proving one version
+faster. Admin/health readiness still must be followed by a data-path check.
+
+To repeat with cached images and a native mcli executable:
+
+```sh
+uv run --with boto3==1.43.92 docs/investigations/issue-116/run-linux.py \
+  0806 0903 --output /tmp/silo-linux-acceptance --mcli /path/to/mcli
+
+CGO_ENABLED=0 GOOS=linux GOARCH=arm64 go build -tags kqueue -o /tmp/silo-linux .
+uv run --with boto3==1.43.92 docs/investigations/issue-116/run-linux.py \
+  current --output /tmp/silo-linux-candidate --mcli /path/to/mcli \
+  --current-binary /tmp/silo-linux --source-revision "$(git rev-parse HEAD)"
+```
+
+The runner uses `--pull=never`; preload the two release images, `alpine:3.23`,
+and the Linux base image selected with `--current-image`. Match the candidate
+binary architecture to that image. Detailed local evidence, including the
+acknowledgement ledger and full logs, remains in the requested output directory.
+
+## OIDC (#154)
+
+The TLS implementation fix was already merged in `48e1846525cc`: transports
+honor Go's key-exchange defaults, so `GODEBUG=tlsmlkem=0` can opt out of ML-KEM
+for an ingress that rejects it. It does not disable certificate validation or
+force an automatic protocol downgrade. See the existing
+[Go 1.27 investigation](go127-stack.md) and [issue investigation](issue-154.md).
+
+A Linux build of main `b32f2d9dd01a` passed fresh isolated fixture checks:
+
+- ML-KEM-intolerant IdP with `tlsmlkem=0`: discovery, IAM, Console login (204),
+  authenticated bucket listing (200).
+- Normal TLS 1.3 IdP without that override: the same complete login chain.
+- Add OIDC through the real administration API with the compatibility setting.
+- Invalid JWT signature and audience: no session cookie and bucket access 403.
+- Untrusted CA: discovery fails and cluster readiness remains 503.
+
+The affected customer's discovery URL and ingress configuration are still
+unavailable. This establishes the supported local fix and its negative
+controls, not the root cause or recovery of that hidden deployment. #154 stays
+open for an affected-environment retest; no release date or published artifact
+is implied by these checks.