fix: close IAM loading and batch recovery review gaps

Signed-off-by: Feng Ruohang <rh@vonng.com>

cmd/admin-handlers.go

@@ -791,7 +791,7 @@ func (a adminAPIHandlers) MetricsHandler(w http.ResponseWriter, r *http.Request)
 
 	enc := json.NewEncoder(w)
 	for n > 0 {
-		var m madmin.RealtimeMetrics
+		var m realtimeMetrics
 		mLocal := collectLocalMetrics(types, collectMetricsOpts{
 			hosts: hostMap,
 			disks: diskMap,
@@ -2675,7 +2675,7 @@ func fetchHealthInfo(healthCtx context.Context, objectAPI ObjectLayer, query *ur
 	// collect all realtime metrics except disk
 	// disk metrics are already included under drive info of each server
 	getRealtimeMetrics := func() *madmin.RealtimeMetrics {
-		var m madmin.RealtimeMetrics
+		var m realtimeMetrics
 		types := madmin.MetricsAll &^ madmin.MetricsDisk
 		mLocal := collectLocalMetrics(types, collectMetricsOpts{})
 		m.Merge(&mLocal)
@@ -2690,7 +2690,8 @@ func fetchHealthInfo(healthCtx context.Context, objectAPI ObjectLayer, query *ur
 			m.ByHost[anonAddr(host)] = metrics
 			delete(m.ByHost, host)
 		}
-		return &m
+		legacy := m.legacy()
+		return &legacy
 	}
 
 	anonymizeCmdLine := func(cmdLine string) string {

cmd/batch-handlers.go

@@ -462,6 +462,7 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 	if done, err := ri.startAuthorized(ctx, api, job); done || err != nil {
 		return err
 	}
+	ri.resetReplicationAttempt()
 	globalBatchJobsMetrics.save(job.ID, ri)
 
 	retryAttempts := job.Replicate.Flags.Retry.Attempts
@@ -576,9 +577,10 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 		s3Type := r.Target.Type == BatchJobReplicateResourceS3 || r.Source.Type == BatchJobReplicateResourceS3
 		minioSrc := r.Source.Type == BatchJobReplicateResourceMinIO
 		ctx, cancelCause := context.WithCancelCause(ctx)
-		// refuse ends the job for its submitter: record why before stopping.
-		refuse := func(err error) {
-			ri.refuse(err)
+		stop := func(err error) {
+			if isBatchJobAuthzError(err) {
+				ri.refuse(err)
+			}
 			cancelCause(err)
 		}
 		// Each run lists the source from the start and counts every locked
@@ -615,6 +617,15 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 				// Refused or canceled: schedule nothing more.
 				continue
 			}
+			if obj.Err != nil {
+				switch minio.ToErrorResponse(obj.Err).StatusCode {
+				case http.StatusUnauthorized, http.StatusForbidden:
+					stop(batchJobAuthzError{reason: "source listing denied: " + obj.Err.Error()})
+				default:
+					stop(batchListingError(obj.Err))
+				}
+				continue
+			}
 			oi := toObjectInfo(r.Source.Bucket, obj.Key, obj)
 			if !minioSrc {
 				// Check if metadata filter was requested and it is expected to have
@@ -663,8 +674,11 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 				stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts)
 				success := true
 				if err := r.ReplicateFromSource(ctx, api, core, oi, retry, job.Submitter); err != nil {
-					if isBatchJobAuthzError(err) {
-						refuse(err)
+					if isBatchJobStopped(err) {
+						stop(err)
+						return
+					}
+					if ctx.Err() != nil {
 						return
 					}
 					// A locked destination version is left as it is: the
@@ -706,6 +720,14 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 			}
 			return nil
 		}
+		if cause := context.Cause(ctx); cause != nil {
+			cancelCause(nil)
+			if isBatchJobUnavailable(cause) {
+				ri.resetReplicationAttempt()
+				return ri.savePaused(ctx, api, job, cause)
+			}
+			return cause
+		}
 
 		ri.RetryAttempts = attempts
 		ri.Complete = ri.ObjectsFailed == 0 && ri.DeleteMarkersFailed == 0
@@ -721,14 +743,7 @@ func (r *BatchJobReplicateV1) StartFromSource(ctx context.Context, api ObjectLay
 
 		cancelCause(nil)
 		if ri.Failed {
-			ri.ObjectsFailed = 0
-			ri.DeleteMarkersFailed = 0
-			ri.Bucket = ""
-			ri.Object = ""
-			ri.Objects = 0
-			ri.DeleteMarkers = 0
-			ri.BytesFailed = 0
-			ri.BytesTransferred = 0
+			ri.resetReplicationAttempt()
 			retry = true // indicate we are retrying..
 			time.Sleep(delay + time.Duration(rnd.Float64()*float64(delay)))
 			continue
@@ -1337,6 +1352,21 @@ func (ri *batchJobInfo) trackCurrentBucketBatch(bucket string, batch []ObjectInf
 	}
 }
 
+func (ri *batchJobInfo) resetReplicationAttempt() {
+	ri.mu.Lock()
+	defer ri.mu.Unlock()
+	ri.Complete, ri.Failed = false, false
+	ri.ObjectsFailed = 0
+	ri.DeleteMarkersFailed = 0
+	ri.Bucket = ""
+	ri.Object = ""
+	ri.Objects = 0
+	ri.DeleteMarkers = 0
+	ri.BytesFailed = 0
+	ri.BytesTransferred = 0
+	ri.VersionsSkippedLocked = 0
+}
+
 // Start start the batch replication job, resumes if there was a pending job via "job.ID"
 func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job BatchJobRequest) error {
 	ri := &batchJobInfo{
@@ -1353,8 +1383,8 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 	if done, err := ri.startAuthorized(ctx, api, job); done || err != nil {
 		return err
 	}
+	ri.resetReplicationAttempt()
 	globalBatchJobsMetrics.save(job.ID, ri)
-	lastObject := ri.Object
 
 	retryAttempts := job.Replicate.Flags.Retry.Attempts
 	if retryAttempts <= 0 {
@@ -1461,9 +1491,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 			slowCh = make(chan itemOrErr[ObjectInfo], 100)
 		)
 		ctx, cancelCause := context.WithCancelCause(ctx)
-		// refuse ends the job for its submitter: record why before stopping.
-		refuse := func(err error) {
-			ri.refuse(err)
+		stop := func(err error) {
+			if isBatchJobAuthzError(err) {
+				ri.refuse(err)
+			}
 			cancelCause(err)
 		}
 
@@ -1488,8 +1519,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 				writeFn := func(batch []ObjectInfo) {
 					if len(batch) > 0 {
 						if err := r.writeAsArchive(ctx, api, cl, batch, r.Target.Prefix, job.Submitter); err != nil {
-							if isBatchJobAuthzError(err) {
-								refuse(err)
+							if isBatchJobStopped(err) {
+								stop(err)
+								return
+							}
+							if ctx.Err() != nil {
 								return
 							}
 							batchLogOnceIf(ctx, err, job.ID+"writeAsArchive")
@@ -1505,7 +1539,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 					}
 				}
 				for obj := range walkCh {
-					if obj.Item.DeleteMarker || !obj.Item.VersionPurgeStatus.Empty() || obj.Item.Size >= int64(smallerThan) {
+					if context.Cause(ctx) != nil {
+						continue
+					}
+					if obj.Err != nil || obj.Item.DeleteMarker || !obj.Item.VersionPurgeStatus.Empty() || obj.Item.Size >= int64(smallerThan) {
 						slowCh <- obj
 						continue
 					}
@@ -1518,7 +1555,9 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 					writeFn(batch)
 					batch = batch[:0]
 				}
-				writeFn(batch)
+				if context.Cause(ctx) == nil {
+					writeFn(batch)
+				}
 				xioutil.SafeClose(slowCh)
 			}()
 		} else {
@@ -1551,11 +1590,10 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 			for _, prefix := range prefixes {
 				prefixWalkCh := make(chan itemOrErr[ObjectInfo], 100)
 				if err := api.Walk(ctx, r.Source.Bucket, prefix, prefixWalkCh, WalkOptions{
-					Marker:   lastObject,
 					Filter:   selectObj,
 					AskDisks: walkQuorum,
 				}); err != nil {
-					cancelCause(err)
+					cancelCause(batchListingError(err))
 					xioutil.SafeClose(walkCh)
 					return
 				}
@@ -1575,7 +1613,7 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 				continue
 			}
 			if res.Err != nil {
-				ri.Failed = true
+				cancelCause(batchListingError(res.Err))
 				batchLogOnceIf(ctx, res.Err, job.ID+"res.Err")
 				continue
 			}
@@ -1594,8 +1632,11 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 				stopFn := globalBatchJobsMetrics.trace(batchJobMetricReplication, job.ID, attempts)
 				success := true
 				if err := r.ReplicateToTarget(ctx, api, c, result, retry, job.Submitter); err != nil {
-					if isBatchJobAuthzError(err) {
-						refuse(err)
+					if isBatchJobStopped(err) {
+						stop(err)
+						return
+					}
+					if ctx.Err() != nil {
 						return
 					}
 					if minio.ToErrorResponse(err).Code == "PreconditionFailed" {
@@ -1632,9 +1673,13 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 			}
 			return nil
 		}
-		// Do not need to retry if we can't list objects on source.
-		if context.Cause(ctx) != nil {
-			return context.Cause(ctx)
+		if cause := context.Cause(ctx); cause != nil {
+			cancelCause(nil)
+			if isBatchJobUnavailable(cause) {
+				ri.resetReplicationAttempt()
+				return ri.savePaused(ctx, api, job, cause)
+			}
+			return cause
 		}
 		ri.RetryAttempts = attempts
 		ri.Complete = ri.ObjectsFailed == 0 && ri.DeleteMarkersFailed == 0
@@ -1650,14 +1695,7 @@ func (r *BatchJobReplicateV1) Start(ctx context.Context, api ObjectLayer, job Ba
 
 		cancelCause(nil)
 		if ri.Failed {
-			ri.ObjectsFailed = 0
-			ri.DeleteMarkersFailed = 0
-			ri.Bucket = ""
-			ri.Object = ""
-			ri.Objects = 0
-			ri.DeleteMarkers = 0
-			ri.BytesFailed = 0
-			ri.BytesTransferred = 0
+			ri.resetReplicationAttempt()
 			retry = true // indicate we are retrying..
 			time.Sleep(delay + time.Duration(rnd.Float64()*float64(delay)))
 			continue
@@ -2627,20 +2665,20 @@ func (ri *batchJobInfo) metric() madmin.JobMetric {
 	return m
 }
 
-func (m *batchJobMetrics) report(jobID string) (metrics *madmin.BatchJobMetrics) {
-	metrics = &madmin.BatchJobMetrics{CollectedAt: time.Now(), Jobs: make(map[string]madmin.JobMetric)}
+func (m *batchJobMetrics) report(jobID string) (metrics *batchJobMetricsReport) {
+	metrics = &batchJobMetricsReport{CollectedAt: time.Now(), Jobs: make(map[string]batchJobStatusMetric)}
 	m.RLock()
 	defer m.RUnlock()
 
 	if jobID != "" {
 		if job, ok := m.metrics[jobID]; ok {
-			metrics.Jobs[jobID] = job.metric()
+			metrics.Jobs[jobID] = job.statusMetric()
 		}
 		return metrics
 	}
 
 	for id, job := range m.metrics {
-		metrics.Jobs[id] = job.metric()
+		metrics.Jobs[id] = job.statusMetric()
 	}
 	return metrics
 }

cmd/batch-progress.go

@@ -170,7 +170,7 @@ type batchJobStatus struct {
 
 type batchJobStatusMetric struct {
 	madmin.JobMetric
-	VersionsSkippedLocked int64  `json:"versionsSkippedLocked,omitempty"`
+	VersionsSkippedLocked *int64 `json:"versionsSkippedLocked,omitempty"`
 	KeysRefused           int64  `json:"keysRefused,omitempty"`
 	FailureReason         string `json:"failureReason,omitempty"`
 	PausedReason          string `json:"pausedReason,omitempty"`
@@ -179,9 +179,10 @@ type batchJobStatusMetric struct {
 func (ri *batchJobInfo) statusMetric() batchJobStatusMetric {
 	ri.mu.RLock()
 	defer ri.mu.RUnlock()
+	versionsSkipped := ri.VersionsSkippedLocked
 	return batchJobStatusMetric{
 		JobMetric:             ri.metric(),
-		VersionsSkippedLocked: ri.VersionsSkippedLocked,
+		VersionsSkippedLocked: &versionsSkipped,
 		KeysRefused:           ri.KeysRefused,
 		FailureReason:         ri.FailureReason,
 		PausedReason:          ri.PausedReason,

cmd/batch-replicate-retry_test.go

@@ -0,0 +1,343 @@
+package cmd
+
+import (
+	"archive/tar"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	polplugin "github.com/minio/minio/internal/config/policy/plugin"
+	xnet "github.com/pgsty/silo-pkg/v3/net"
+	"github.com/pgsty/silo-pkg/v3/policy"
+)
+
+type batchReplicationWriteObserver struct {
+	ObjectLayer
+	written func(string, string)
+}
+
+type batchReplicationWalkFailure struct {
+	ObjectLayer
+	streamed bool
+}
+
+func (failure batchReplicationWalkFailure) Walk(ctx context.Context, _, _ string, results chan<- itemOrErr[ObjectInfo], _ WalkOptions) error {
+	if !failure.streamed {
+		return errErasureReadQuorum
+	}
+	go func() {
+		defer close(results)
+		select {
+		case results <- itemOrErr[ObjectInfo]{Err: errErasureReadQuorum}:
+		case <-ctx.Done():
+		}
+	}()
+	return nil
+}
+
+func TestBatchReplicationListingUnavailable(t *testing.T) {
+	for _, snowball := range []bool{false, true} {
+		for _, streamed := range []bool{false, true} {
+			t.Run(fmt.Sprintf("snowball=%t/streamed=%t", snowball, streamed), func(t *testing.T) {
+				fixture := newBatchAuthzFixture(t)
+				job := BatchJobRequest{
+					ID: "replicate-listing" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+					Replicate: &BatchJobReplicateV1{
+						Source: BatchJobReplicateSource{
+							Type: BatchJobReplicateResourceMinIO, Bucket: fixture.bucket,
+							Snowball: BatchJobSnowball{
+								Disable: ptr(!snowball), InMemory: ptr(true), Compress: ptr(false), SkipErrs: ptr(false),
+								Batch: ptr(1), SmallerThan: ptr("1MiB"),
+							},
+						},
+						Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://target.invalid"},
+					},
+				}
+				api := batchReplicationWalkFailure{ObjectLayer: fixture.obj, streamed: streamed}
+				if err := job.Replicate.Start(fixture.ctx, api, job); !isBatchJobUnavailable(err) {
+					t.Fatalf("listing failure did not pause: %v", err)
+				}
+				report := fixture.jobInfo(t, &job)
+				if report.PausedReason == "" || report.Failed || report.Complete || report.Object != "" || report.ObjectsFailed != 0 {
+					t.Fatalf("listing failure saved an invalid outcome: %+v", report)
+				}
+			})
+		}
+	}
+}
+
+func (observer batchReplicationWriteObserver) PutObject(ctx context.Context, bucket, object string, reader *PutObjReader, opts ObjectOptions) (ObjectInfo, error) {
+	info, err := observer.ObjectLayer.PutObject(ctx, bucket, object, reader, opts)
+	if err == nil {
+		observer.written(bucket, object)
+	}
+	return info, err
+}
+
+func TestBatchReplicationRemoteListingDenied(t *testing.T) {
+	fixture := newBatchAuthzFixture(t)
+	previousTransport := remoteInstanceTransport
+	remoteInstanceTransport = atomic.Value{}
+	setRemoteInstanceTransport(batchLockTransport(func(request *http.Request) (*http.Response, error) {
+		status := http.StatusForbidden
+		body := `<Error><Code>AccessDenied</Code><Message>source listing denied</Message></Error>`
+		if request.URL.Query().Has("location") {
+			status = http.StatusOK
+			body = `<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/">us-east-1</LocationConstraint>`
+		}
+		return &http.Response{
+			StatusCode: status, Header: make(http.Header),
+			Body: io.NopCloser(strings.NewReader(body)), ContentLength: int64(len(body)), Request: request,
+		}, nil
+	}))
+	defer func() { remoteInstanceTransport = previousTransport }()
+	job := BatchJobRequest{
+		ID: "replicate-listing-denied" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+		Replicate: &BatchJobReplicateV1{
+			Source: BatchJobReplicateSource{
+				Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://source.invalid",
+				Creds: BatchJobReplicateCredentials{AccessKey: "test-access", SecretKey: "test-secret"},
+			},
+			Target: BatchJobReplicateTarget{Type: BatchJobReplicateResourceMinIO, Bucket: fixture.bucket},
+		},
+	}
+	if err := job.Replicate.StartFromSource(fixture.ctx, fixture.obj, job); err != nil {
+		t.Fatalf("explicit remote denial was not terminal: %v", err)
+	}
+	report := fixture.jobInfo(t, &job)
+	if !report.Failed || report.Complete || report.FailureReason == "" || report.PausedReason != "" {
+		t.Fatalf("remote denial saved an invalid outcome: %+v", report)
+	}
+}
+
+func TestBatchReplicationDependencyRecovery(t *testing.T) {
+	for _, mode := range []string{"push", "pull", "snowball"} {
+		for _, denied := range []bool{false, true} {
+			t.Run(fmt.Sprintf("%s/denied=%t", mode, denied), func(t *testing.T) {
+				t.Setenv("_MINIO_BATCH_REPLICATION_WORKERS", "2")
+				fixture := newBatchAuthzFixture(t)
+				keys := []string{"a-done", "b-pending", "c-later"}
+				versions := make(map[string]string)
+				for _, key := range keys {
+					versions[key] = mustGetUUID()
+					if mode != "pull" {
+						versions[key] = batchPutVersion(t, fixture.obj, fixture.bucket, key, nil).VersionID
+					}
+				}
+				laterDone := make(chan struct{})
+				signalLater := sync.OnceFunc(func() { close(laterDone) })
+				var writtenMu sync.Mutex
+				written := make(map[string]bool)
+				recordWrite := func(bucket, key string) {
+					if bucket != fixture.bucket && bucket != "remote" {
+						return
+					}
+					writtenMu.Lock()
+					written[key] = true
+					writtenMu.Unlock()
+					if key == "c-later" {
+						signalLater()
+					}
+				}
+				var outage atomic.Bool
+				outage.Store(true)
+				var decisions atomic.Int32
+				pluginServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+					var input struct {
+						Input policy.Args `json:"input"`
+					}
+					if err := json.NewDecoder(request.Body).Decode(&input); err != nil {
+						t.Error(err)
+						writer.WriteHeader(http.StatusBadRequest)
+						return
+					}
+					action := policy.Action(policy.GetObjectAction)
+					if mode == "pull" {
+						action = policy.PutObjectAction
+					}
+					if outage.Load() && input.Input.Action == action && input.Input.ObjectName == "b-pending" {
+						decisions.Add(1)
+						if mode != "snowball" {
+							select {
+							case <-laterDone:
+							case <-request.Context().Done():
+								return
+							}
+						}
+						if denied {
+							io.WriteString(writer, `{"result":false}`)
+							return
+						}
+						writer.WriteHeader(http.StatusServiceUnavailable)
+					}
+					io.WriteString(writer, `{"result":true}`)
+				}))
+				defer pluginServer.Close()
+				pluginURL, err := xnet.ParseHTTPURL(pluginServer.URL)
+				mustIAM(t, err)
+				previousPlugin := newGlobalAuthZPluginFn()
+				setGlobalAuthZPlugin(polplugin.New(polplugin.Args{
+					URL: pluginURL, CloseRespFn: func(reader io.ReadCloser) { reader.Close() },
+				}))
+				defer setGlobalAuthZPlugin(previousPlugin)
+				previousTransport := remoteInstanceTransport
+				remoteInstanceTransport = atomic.Value{}
+				setRemoteInstanceTransport(batchLockTransport(func(request *http.Request) (*http.Response, error) {
+					if err := request.Context().Err(); err != nil {
+						return nil, err
+					}
+					body := ""
+					header := make(http.Header)
+					query := request.URL.Query()
+					switch {
+					case query.Has("location"):
+						body = `<LocationConstraint xmlns="http://s3.amazonaws.com/doc/2006-03-01/">us-east-1</LocationConstraint>`
+					case query.Has("versions"):
+						body = `<ListVersionsResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><Name>remote</Name><IsTruncated>false</IsTruncated>`
+						for _, key := range keys {
+							body += fmt.Sprintf(`<Version><Key>%s</Key><VersionId>%s</VersionId><IsLatest>true</IsLatest><LastModified>2026-11-01T00:00:00.000Z</LastModified><ETag>"etag"</ETag><Size>7</Size><StorageClass>STANDARD</StorageClass></Version>`, key, versions[key])
+						}
+						body += `</ListVersionsResult>`
+					case request.Method == http.MethodGet:
+						body = "payload"
+						header.Set("Etag", `"etag"`)
+						header.Set("Last-Modified", UTCNow().Format(http.TimeFormat))
+						header.Set("X-Amz-Version-Id", query.Get("versionId"))
+					case request.Method == http.MethodPut:
+						if mode == "snowball" && request.Body != nil {
+							archive := tar.NewReader(request.Body)
+							for {
+								entry, readErr := archive.Next()
+								if readErr == io.EOF {
+									break
+								}
+								if readErr != nil {
+									return nil, readErr
+								}
+								recordWrite("remote", entry.Name)
+							}
+						} else if mode != "snowball" {
+							if _, err := io.Copy(io.Discard, request.Body); err != nil {
+								return nil, err
+							}
+							recordWrite("remote", strings.TrimPrefix(request.URL.Path, "/remote/"))
+						}
+						header.Set("Etag", `"etag"`)
+					default:
+						return nil, fmt.Errorf("unexpected replication request: %s %s", request.Method, request.URL)
+					}
+					header.Set("Content-Length", strconv.Itoa(len(body)))
+					return &http.Response{
+						StatusCode: http.StatusOK, Header: header,
+						Body: io.NopCloser(strings.NewReader(body)), ContentLength: int64(len(body)), Request: request,
+					}, nil
+				}))
+				defer func() { remoteInstanceTransport = previousTransport }()
+				job := &BatchJobRequest{
+					ID: "replicate-retry" + getKeySeparator() + "0", Started: UTCNow(), Submitter: batchRootSubmitter(t),
+					Replicate: &BatchJobReplicateV1{
+						Source: BatchJobReplicateSource{Type: BatchJobReplicateResourceMinIO, Bucket: fixture.bucket},
+						Target: BatchJobReplicateTarget{
+							Type: BatchJobReplicateResourceMinIO, Bucket: "remote", Endpoint: "https://target.invalid",
+							Creds: BatchJobReplicateCredentials{AccessKey: "test-access", SecretKey: "test-secret"},
+						},
+					},
+				}
+				if mode == "pull" {
+					job.Replicate.Source.Endpoint = job.Replicate.Target.Endpoint
+					job.Replicate.Source.Creds = job.Replicate.Target.Creds
+					job.Replicate.Source.Bucket = "remote"
+					job.Replicate.Target.Endpoint = ""
+					job.Replicate.Target.Bucket = fixture.bucket
+				}
+				if mode == "snowball" {
+					job.Replicate.Source.Snowball = BatchJobSnowball{
+						Disable: ptr(false), InMemory: ptr(true), Compress: ptr(false), SkipErrs: ptr(false),
+						Batch: ptr(1), SmallerThan: ptr("1MiB"),
+					}
+				}
+				job.Replicate.Flags.Retry.Attempts = 1
+				job.Replicate.Flags.Retry.Delay = time.Millisecond
+				interrupted := &batchJobInfo{
+					JobID: job.ID, JobType: string(job.Type()), StartTime: job.Started,
+					Object: "c-later", Objects: 2,
+				}
+				mustIAM(t, interrupted.updateAfter(fixture.ctx, fixture.obj, 0, *job))
+				pool := globalBatchJobPool
+				pool.objLayer = batchReplicationWriteObserver{ObjectLayer: fixture.obj, written: recordWrite}
+				ctx, cancel := context.WithCancel(fixture.ctx)
+				pool.ctx = ctx
+				mustIAM(t, pool.queueJob(job))
+				done := make(chan struct{})
+				go func() { defer close(done); pool.AddWorker() }()
+				defer func() {
+					cancel()
+					<-done
+					if t.Failed() {
+						t.Logf("last report: %+v", fixture.jobInfo(t, job))
+					}
+				}()
+				state := "paused"
+				if denied {
+					state = "terminal"
+				}
+				waitBatchState(t, pool, job.ID, state)
+				report := fixture.jobInfo(t, job)
+				if decisions.Load() == 0 {
+					t.Fatal("test did not reach per-object authorization")
+				}
+				if denied {
+					if !report.Failed || report.Complete || report.FailureReason == "" || report.PausedReason != "" {
+						t.Fatalf("denial was not terminal: %+v", report)
+					}
+					pool.retryDue(time.Now().Add(time.Hour))
+					if len(pool.jobCh) != 0 {
+						t.Fatal("denied job was retried")
+					}
+					return
+				}
+				if report.Failed || report.Complete || report.FailureReason != "" || report.PausedReason == "" ||
+					report.Object != "" || report.Objects != 0 || report.ObjectsFailed != 0 {
+					t.Fatalf("outage was not saved with a safe restart point: %+v", report)
+				}
+				pool.jmu.Lock()
+				owner := pool.jobOwners[job.ID]
+				sameOwner := owner.req == job && owner.retries == 1
+				pool.jmu.Unlock()
+				if !sameOwner {
+					t.Fatal("pause lost the job owner")
+				}
+				outage.Store(false)
+				pool.retryDue(time.Now().Add(time.Hour))
+				waitBatchState(t, pool, job.ID, "terminal")
+				report = fixture.jobInfo(t, job)
+				if !report.Complete || report.Failed || report.PausedReason != "" || report.Objects != int64(len(keys)) || report.ObjectsFailed != 0 {
+					t.Fatalf("retry did not finish with consistent counts: %+v", report)
+				}
+				writtenMu.Lock()
+				for _, key := range keys {
+					if !written[key] {
+						t.Errorf("retry skipped %s", key)
+					}
+				}
+				writtenMu.Unlock()
+				if mode == "pull" {
+					for _, key := range keys {
+						if !batchVersionExists(t, fixture.obj, fixture.bucket, key, versions[key]) {
+							t.Errorf("retry lost remote version of %s", key)
+						}
+					}
+				}
+			})
+		}
+	}
+}

cmd/iam-load-user-race_test.go

@@ -5,22 +5,44 @@ package cmd
 
 import (
 	"context"
+	"errors"
+	"slices"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/pgsty/silo-pkg/v3/policy"
 )
 
 // Hold a lazy user load after it has read storage. This makes the ordering
 // between an in-flight read and a completed credential edit deterministic.
 type iamLoadUserBlocker struct {
 	IAMStorageAPI
-	key     string
-	waiting chan struct{}
-	release chan struct{}
-	once    sync.Once
+	key          string
+	waiting      chan struct{}
+	release      chan struct{}
+	once         sync.Once
+	loads        atomic.Int32
+	retryWaiting chan struct{}
+	retryRelease chan struct{}
+	retryErr     error
 }
 
 func (s *iamLoadUserBlocker) loadUser(ctx context.Context, key string, userType IAMUserType, out map[string]UserIdentity) error {
+	if key == s.key && userType == regUser && s.loads.Add(1) == 2 {
+		if s.retryWaiting != nil {
+			close(s.retryWaiting)
+			select {
+			case <-s.retryRelease:
+			case <-ctx.Done():
+				return ctx.Err()
+			}
+		}
+		if s.retryErr != nil {
+			return s.retryErr
+		}
+	}
 	err := s.IAMStorageAPI.loadUser(ctx, key, userType, out)
 	if err == nil && key == s.key && userType == regUser {
 		s.once.Do(func() { close(s.waiting) })
@@ -33,6 +55,230 @@ func (s *iamLoadUserBlocker) loadUser(ctx context.Context, key string, userType
 	return err
 }
 
+func TestIAMLoadUserConcurrentUpdates(t *testing.T) {
+	for _, change := range []string{"unrelated-edit", "unrelated-load", "delete", "mapping", "policy"} {
+		t.Run(change, func(t *testing.T) {
+			ctx, sys, _ := prepareIAMRevisionFixture(t)
+			const policyName = "lazy-policy"
+			batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+			cold := iamAdminUser(ctx, t, sys, "lazy-cold", policyName)
+			other := iamAdminUser(ctx, t, sys, "lazy-other", "readwrite")
+			cache := sys.store.lock()
+			delete(cache.iamUsersMap, cold.AccessKey)
+			sys.store.unlock()
+			original := sys.store.IAMStorageAPI
+			blocker := &iamLoadUserBlocker{
+				IAMStorageAPI: original, key: cold.AccessKey,
+				waiting: make(chan struct{}), release: make(chan struct{}),
+			}
+			sys.store.IAMStorageAPI = blocker
+			unblock := sync.OnceFunc(func() { close(blocker.release) })
+			done := make(chan struct{})
+			t.Cleanup(func() {
+				unblock()
+				<-done
+				sys.store.IAMStorageAPI = original
+			})
+			var valid bool
+			var loadErr error
+			go func() {
+				defer close(done)
+				_, valid, loadErr = sys.CheckKey(ctx, cold.AccessKey)
+			}()
+			select {
+			case <-blocker.waiting:
+			case <-time.After(5 * time.Second):
+				t.Fatal("cold load did not read the identity")
+			}
+			switch change {
+			case "unrelated-edit":
+				mustIAM(t, sys.SetUserSecretKey(ctx, other.AccessKey, "unrelated-rotated-secret"))
+			case "unrelated-load":
+				mustIAM(t, sys.store.LoadUser(ctx, other.AccessKey))
+			case "delete":
+				mustIAM(t, sys.DeleteUser(ctx, cold.AccessKey, false))
+			case "mapping":
+				_, err := sys.PolicyDBSet(ctx, cold.AccessKey, "readonly", regUser, false)
+				mustIAM(t, err)
+			case "policy":
+				batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+			}
+			unblock()
+			select {
+			case <-done:
+			case <-time.After(5 * time.Second):
+				t.Fatal("cold load did not finish")
+			}
+			if loadErr != nil || valid != (change != "delete") {
+				t.Fatalf("valid=%v err=%v after %s", valid, loadErr, change)
+			}
+			if blocker.loads.Load() != 2 {
+				t.Fatalf("conflicting load read the identity %d times, want 2", blocker.loads.Load())
+			}
+			if change == "mapping" {
+				names, err := sys.store.PolicyDBGet(cold.AccessKey)
+				mustIAM(t, err)
+				if !slices.Equal(names, []string{"readonly"}) {
+					t.Fatalf("restored stale mapping: %v", names)
+				}
+			}
+			if change == "policy" {
+				current, err := sys.store.GetPolicy(policyName)
+				mustIAM(t, err)
+				if current.IsAllowed(policy.Args{Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) {
+					t.Fatal("restored stale policy")
+				}
+			}
+		})
+	}
+}
+
+func TestIAMLoadUserRetryKeepsCachedReadsAvailable(t *testing.T) {
+	for _, unavailable := range []bool{false, true} {
+		t.Run(map[bool]string{false: "loaded", true: "unavailable"}[unavailable], func(t *testing.T) {
+			ctx, sys, _ := prepareIAMRevisionFixture(t)
+			cold := iamAdminUser(ctx, t, sys, "retry-cold", "readwrite")
+			other := iamAdminUser(ctx, t, sys, "retry-cached", "readwrite")
+			cache := sys.store.lock()
+			delete(cache.iamUsersMap, cold.AccessKey)
+			sys.store.unlock()
+			original := sys.store.IAMStorageAPI
+			blocker := &iamLoadUserBlocker{
+				IAMStorageAPI: original, key: cold.AccessKey,
+				waiting: make(chan struct{}), release: make(chan struct{}),
+				retryWaiting: make(chan struct{}), retryRelease: make(chan struct{}),
+			}
+			if unavailable {
+				blocker.retryErr = errErasureReadQuorum
+			}
+			sys.store.IAMStorageAPI = blocker
+			unblock := sync.OnceFunc(func() { close(blocker.release) })
+			unblockRetry := sync.OnceFunc(func() { close(blocker.retryRelease) })
+			done := make(chan struct{})
+			t.Cleanup(func() {
+				unblock()
+				unblockRetry()
+				<-done
+				sys.store.IAMStorageAPI = original
+			})
+			var valid bool
+			var loadErr error
+			go func() {
+				defer close(done)
+				_, valid, loadErr = sys.CheckKey(ctx, cold.AccessKey)
+			}()
+			awaitIAMPolicyReload(t, blocker.waiting)
+			mustIAM(t, sys.SetUserSecretKey(ctx, other.AccessKey, "retry-rotated-secret"))
+			unblock()
+			awaitIAMPolicyReload(t, blocker.retryWaiting)
+			readDone := make(chan struct{})
+			go func() {
+				defer close(readDone)
+				user, ok, err := sys.CheckKey(ctx, other.AccessKey)
+				if err != nil || !ok || user.Credentials.SecretKey != "retry-rotated-secret" {
+					t.Errorf("cached read failed: valid=%v err=%v", ok, err)
+				}
+			}()
+			select {
+			case <-readDone:
+			case <-time.After(time.Second):
+				unblockRetry()
+				<-readDone
+				t.Fatal("cold-load retry blocked cached authentication")
+			}
+			unblockRetry()
+			<-done
+			if unavailable {
+				if valid || !errors.Is(loadErr, errIAMNotInitialized) {
+					t.Fatalf("retry outage was not retryable: valid=%v err=%v", valid, loadErr)
+				}
+			} else if !valid || loadErr != nil {
+				t.Fatalf("retry lost the cold identity: valid=%v err=%v", valid, loadErr)
+			}
+		})
+	}
+}
+
+func TestIAMLoadUserRefreshesAssociatedRecords(t *testing.T) {
+	t.Setenv("_MINIO_IAM_GROUP_REFRESH", "on")
+	for _, change := range []string{"policy", "mapping", "group", "parent"} {
+		t.Run(change, func(t *testing.T) {
+			ctx, sys, obj := prepareIAMRevisionFixture(t)
+			const policyName = "associated-policy"
+			const groupName = "associated-group"
+			batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Allow","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+			parent := iamAdminUser(ctx, t, sys, "associated-parent", policyName)
+			_, err := sys.AddUsersToGroup(ctx, groupName, []string{parent.AccessKey})
+			mustIAM(t, err)
+			key := parent.AccessKey
+			if change == "parent" {
+				key = iamAdminServiceAccount(ctx, t, sys, parent.AccessKey, "associated-service", "").AccessKey
+			}
+			gate := &iamPolicyRefreshReadGate{
+				ObjectLayer: obj, userPath: getGroupInfoPath(groupName),
+				refreshEntered: make(chan struct{}), refreshRelease: make(chan struct{}),
+			}
+			sys.store.IAMStorageAPI.(*IAMObjectStore).objAPI = gate
+			unblock := sync.OnceFunc(func() { close(gate.refreshRelease) })
+			done := make(chan struct{})
+			var loadErr error
+			t.Cleanup(func() {
+				unblock()
+				<-done
+			})
+			go func() {
+				defer close(done)
+				loadErr = sys.store.LoadUser(context.WithValue(ctx, iamPolicyRefreshContext{}, "refresh"), key)
+			}()
+			awaitIAMPolicyReload(t, gate.refreshEntered)
+			switch change {
+			case "policy":
+				batchAuthzPolicy(ctx, t, sys, policyName, `{"Effect":"Deny","Action":["s3:*"],"Resource":["arn:aws:s3:::bucket/*"]}`)
+			case "mapping":
+				_, err = sys.PolicyDBSet(ctx, parent.AccessKey, "readonly", regUser, false)
+				mustIAM(t, err)
+			case "group":
+				_, err = sys.RemoveUsersFromGroup(ctx, groupName, []string{parent.AccessKey})
+				mustIAM(t, err)
+			case "parent":
+				mustIAM(t, sys.SetUserSecretKey(ctx, parent.AccessKey, "associated-rotated-secret"))
+			}
+			unblock()
+			select {
+			case <-done:
+				mustIAM(t, loadErr)
+			case <-time.After(5 * time.Second):
+				t.Fatal("associated-record reload did not finish")
+			}
+			switch change {
+			case "policy":
+				current, err := sys.store.GetPolicy(policyName)
+				mustIAM(t, err)
+				if current.IsAllowed(policy.Args{Action: policy.PutObjectAction, BucketName: "bucket", ObjectName: "object"}) {
+					t.Fatal("partial snapshot restored old policy")
+				}
+			case "mapping":
+				mapping, ok := sys.store.GetMappedPolicy(parent.AccessKey, false)
+				if !ok || mapping.Policies != "readonly" {
+					t.Fatalf("partial snapshot restored old mapping: %+v", mapping)
+				}
+			case "group":
+				cache := sys.store.rlock()
+				member := cache.iamUserGroupMemberships[parent.AccessKey].Contains(groupName)
+				sys.store.runlock()
+				if member {
+					t.Fatal("partial snapshot restored old group membership")
+				}
+			case "parent":
+				current, valid, err := sys.CheckKey(ctx, parent.AccessKey)
+				if err != nil || !valid || current.Credentials.SecretKey != "associated-rotated-secret" {
+					t.Fatal("child reload restored old parent credential")
+				}
+			}
+		})
+	}
+}
+
 func TestIAMLoadUserDoesNotRestoreRotatedCredential(t *testing.T) {
 	ctx, sys, _ := prepareIAMRevisionFixture(t)
 	user := iamAdminUser(ctx, t, sys, "load-user-race", "readwrite")

cmd/iam-store.go

@@ -3225,15 +3225,17 @@ func (store *IAMStoreSys) forgetCachedIdentity(cache *iamCache, accessKey string
 
 // LoadUser - attempts to load user info from storage and updates cache.
 func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) error {
-	loadedAt := time.Now()
 	groupLoad := env.Get("_MINIO_IAM_GROUP_REFRESH", config.EnableOff) == config.EnableOn
 
-	newCachePopulate := func() (val any, err error) {
-		newCache := newIamCache()
+	newCachePopulate := func(ctx context.Context) (newCache *iamCache, err error) {
+		newCache = newIamCache()
 		newCache.revisions = store.revisionIndex()
 
 		// Check for service account first
-		store.loadUser(ctx, accessKey, svcUser, newCache.iamUsersMap)
+		err = store.loadUser(ctx, accessKey, svcUser, newCache.iamUsersMap)
+		if err != nil && !errors.Is(err, errNoSuchUser) && !errors.Is(err, errConfigNotFound) {
+			return nil, err
+		}
 
 		svc, found := newCache.iamUsersMap[accessKey]
 		if found {
@@ -3251,6 +3253,9 @@ func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) error
 
 		if !found {
 			err = store.loadUser(ctx, accessKey, regUser, newCache.iamUsersMap)
+			if err != nil && !errors.Is(err, errNoSuchUser) && !errors.Is(err, errConfigNotFound) {
+				return nil, err
+			}
 			if _, found = newCache.iamUsersMap[accessKey]; found {
 				// NOTE: we are not worried about loading errors from policies.
 				store.loadMappedPolicyWithRetry(ctx, accessKey, regUser, false, newCache.iamUserPolicyMap, 3)
@@ -3294,69 +3299,67 @@ func (store *IAMStoreSys) LoadUser(ctx context.Context, accessKey string) error
 		return newCache, err
 	}
 
-	var (
-		val any
-		err error
-	)
-	if store.group != nil {
-		val, err, _ = store.group.Do(accessKey, newCachePopulate)
-	} else {
-		val, err = newCachePopulate()
-	}
-
-	// Return error right away if any.
-	if err != nil {
-		if errors.Is(err, errNoSuchUser) || errors.Is(err, errConfigNotFound) {
-			return nil
+	load := func() (any, error) {
+		loadedAt := time.Now()
+		newCache, err := newCachePopulate(ctx)
+		if err != nil {
+			return nil, err
 		}
-		return err
+
+		cache := store.lock()
+		defer store.unlock()
+
+		if cache.updatedAt.After(loadedAt) {
+			err = store.withIAMStorage(ctx, func(ctx context.Context) error {
+				newCache, err = newCachePopulate(ctx)
+				return err
+			})
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		// We need to merge the new cache with the existing cache because the
+		// periodic IAM reload is partial. The periodic load here is to account.
+		newCache.iamGroupPolicyMap.Range(func(k string, v MappedPolicy) bool {
+			cache.iamGroupPolicyMap.Store(k, v)
+			return true
+		})
+
+		maps.Copy(cache.iamGroupsMap, newCache.iamGroupsMap)
+
+		maps.Copy(cache.iamPolicyDocsMap, newCache.iamPolicyDocsMap)
+
+		maps.Copy(cache.iamUserGroupMemberships, newCache.iamUserGroupMemberships)
+
+		newCache.iamUserPolicyMap.Range(func(k string, v MappedPolicy) bool {
+			cache.iamUserPolicyMap.Store(k, v)
+			return true
+		})
+
+		maps.Copy(cache.iamUsersMap, newCache.iamUsersMap)
+
+		maps.Copy(cache.iamSTSAccountsMap, newCache.iamSTSAccountsMap)
+
+		newCache.iamSTSPolicyMap.Range(func(k string, v MappedPolicy) bool {
+			cache.iamSTSPolicyMap.Store(k, v)
+			return true
+		})
+
+		cache.updatedAt = time.Now()
+		return nil, nil
 	}
 
-	newCache, ok := val.(*iamCache)
-	if !ok {
+	var err error
+	if store.group != nil {
+		_, err, _ = store.group.Do(accessKey, load)
+	} else {
+		_, err = load()
+	}
+	if errors.Is(err, errNoSuchUser) || errors.Is(err, errConfigNotFound) {
 		return nil
 	}
-
-	cache := store.lock()
-	defer store.unlock()
-
-	// A partial load may have started before a credential edit or deletion and
-	// finished after it. Do not merge that snapshot over the newer cache state;
-	// the next notification or load will supply the missing entry.
-	if cache.updatedAt.After(loadedAt) {
-		return nil
-	}
-
-	// We need to merge the new cache with the existing cache because the
-	// periodic IAM reload is partial. The periodic load here is to account.
-	newCache.iamGroupPolicyMap.Range(func(k string, v MappedPolicy) bool {
-		cache.iamGroupPolicyMap.Store(k, v)
-		return true
-	})
-
-	maps.Copy(cache.iamGroupsMap, newCache.iamGroupsMap)
-
-	maps.Copy(cache.iamPolicyDocsMap, newCache.iamPolicyDocsMap)
-
-	maps.Copy(cache.iamUserGroupMemberships, newCache.iamUserGroupMemberships)
-
-	newCache.iamUserPolicyMap.Range(func(k string, v MappedPolicy) bool {
-		cache.iamUserPolicyMap.Store(k, v)
-		return true
-	})
-
-	maps.Copy(cache.iamUsersMap, newCache.iamUsersMap)
-
-	maps.Copy(cache.iamSTSAccountsMap, newCache.iamSTSAccountsMap)
-
-	newCache.iamSTSPolicyMap.Range(func(k string, v MappedPolicy) bool {
-		cache.iamSTSPolicyMap.Store(k, v)
-		return true
-	})
-
-	cache.updatedAt = time.Now()
-
-	return nil
+	return err
 }
 
 var errTokenSigningKeyUnavailable = errors.New("token signing key unavailable")

cmd/metrics-batch.go

@@ -0,0 +1,78 @@
+package cmd
+
+import (
+	"maps"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+)
+
+type realtimeMetrics struct {
+	madmin.RealtimeMetrics
+	Aggregated realtimeMetricGroups            `json:"aggregated"`
+	ByHost     map[string]realtimeMetricGroups `json:"by_host,omitempty"`
+}
+
+type realtimeMetricGroups struct {
+	madmin.Metrics
+	BatchJobs *batchJobMetricsReport `json:"batchJobs,omitempty"`
+}
+
+type batchJobMetricsReport struct {
+	CollectedAt time.Time `json:"collected"`
+	Jobs        map[string]batchJobStatusMetric
+}
+
+func (metrics *realtimeMetrics) Merge(other *realtimeMetrics) {
+	if other == nil {
+		return
+	}
+	metrics.RealtimeMetrics.Merge(&other.RealtimeMetrics)
+	metrics.Aggregated.Merge(&other.Aggregated.Metrics)
+	if other.Aggregated.BatchJobs != nil {
+		if metrics.Aggregated.BatchJobs == nil {
+			metrics.Aggregated.BatchJobs = &batchJobMetricsReport{Jobs: make(map[string]batchJobStatusMetric)}
+		}
+		jobs := metrics.Aggregated.BatchJobs
+		if other.Aggregated.BatchJobs.CollectedAt.After(jobs.CollectedAt) {
+			jobs.CollectedAt = other.Aggregated.BatchJobs.CollectedAt
+		}
+		for id, job := range other.Aggregated.BatchJobs.Jobs {
+			if current, ok := jobs.Jobs[id]; !ok || !job.LastUpdate.Before(current.LastUpdate) {
+				jobs.Jobs[id] = job
+			}
+		}
+	}
+	if len(other.ByHost) != 0 {
+		if metrics.ByHost == nil {
+			metrics.ByHost = make(map[string]realtimeMetricGroups, len(other.ByHost))
+		}
+		maps.Copy(metrics.ByHost, other.ByHost)
+	}
+}
+
+func (metrics realtimeMetricGroups) legacy() madmin.Metrics {
+	out := metrics.Metrics
+	if metrics.BatchJobs != nil {
+		out.BatchJobs = &madmin.BatchJobMetrics{
+			CollectedAt: metrics.BatchJobs.CollectedAt,
+			Jobs:        make(map[string]madmin.JobMetric, len(metrics.BatchJobs.Jobs)),
+		}
+		for id, job := range metrics.BatchJobs.Jobs {
+			out.BatchJobs.Jobs[id] = job.JobMetric
+		}
+	}
+	return out
+}
+
+func (metrics realtimeMetrics) legacy() madmin.RealtimeMetrics {
+	out := metrics.RealtimeMetrics
+	out.Aggregated = metrics.Aggregated.legacy()
+	if metrics.ByHost != nil {
+		out.ByHost = make(map[string]madmin.Metrics, len(metrics.ByHost))
+		for host, groups := range metrics.ByHost {
+			out.ByHost[host] = groups.legacy()
+		}
+	}
+	return out
+}

cmd/metrics-batch_test.go

@@ -0,0 +1,153 @@
+package cmd
+
+import (
+	"encoding/json"
+	"fmt"
+	"net/http"
+	"net/http/httptest"
+	"reflect"
+	"testing"
+	"time"
+
+	"github.com/minio/madmin-go/v3"
+	"github.com/minio/minio/internal/grid"
+)
+
+func TestBatchMetricsStatusSnapshots(t *testing.T) {
+	fixture := newBatchAuthzFixture(t)
+	job := BatchJobRequest{
+		ID: "expire-metrics-snapshot" + getKeySeparator() + "0", Started: UTCNow(),
+		Submitter: batchRootSubmitter(t), Expire: &BatchJobExpire{},
+	}
+	for _, state := range []string{"skipped", "paused", "failed", "running"} {
+		t.Run(state, func(t *testing.T) {
+			report := &batchJobInfo{JobID: job.ID, JobType: string(job.Type()), StartTime: job.Started}
+			switch state {
+			case "skipped":
+				report.Complete = true
+				report.VersionsSkippedLocked, report.KeysRefused = 2, 1
+			case "paused":
+				report.PausedReason = "authorization plugin unavailable"
+			case "failed":
+				report.Failed = true
+				report.FailureReason = "authorization denied"
+			}
+			mustIAM(t, report.updateAfter(fixture.ctx, fixture.obj, 0, job))
+			globalBatchJobsMetrics.save(job.ID, report)
+			saved := batchJobStatusJSON(t, job.ID)
+			for _, field := range []string{"startTime", "lastUpdate"} {
+				stamp, err := time.Parse(time.RFC3339Nano, saved[field].(string))
+				mustIAM(t, err)
+				saved[field] = stamp.UTC().Format(time.RFC3339Nano)
+			}
+			target := fmt.Sprintf("http://localhost/minio/admin/v3/metrics?types=%d&by-jobID=%s&n=1&by-host=true", madmin.MetricsBatchJobs, job.ID)
+			request, err := newTestSignedRequestV4(http.MethodGet, target, 0, nil, globalActiveCred.AccessKey, globalActiveCred.SecretKey, nil)
+			mustIAM(t, err)
+			mustIAM(t, request.ParseForm())
+			response := httptest.NewRecorder()
+			request = request.WithContext(newContext(request, response, "Metrics"))
+			adminAPIHandlers{}.MetricsHandler(response, request)
+			if response.Code != http.StatusOK {
+				t.Fatalf("metrics returned %d: %s", response.Code, response.Body.String())
+			}
+			var wire struct {
+				Aggregated struct {
+					BatchJobs struct{ Jobs map[string]map[string]any }
+				}
+				ByHost map[string]struct {
+					BatchJobs struct{ Jobs map[string]map[string]any }
+				} `json:"by_host"`
+			}
+			mustIAM(t, json.Unmarshal(response.Body.Bytes(), &wire))
+			if !reflect.DeepEqual(wire.Aggregated.BatchJobs.Jobs[job.ID], saved) {
+				t.Fatalf("live snapshot differs from saved status: live=%v saved=%v", wire.Aggregated.BatchJobs.Jobs[job.ID], saved)
+			}
+			if len(wire.ByHost) != 1 {
+				t.Fatalf("missing per-host snapshot: %s", response.Body.String())
+			}
+			for _, host := range wire.ByHost {
+				if !reflect.DeepEqual(host.BatchJobs.Jobs[job.ID], saved) {
+					t.Fatal("per-host metrics lost details")
+				}
+			}
+			var legacy madmin.RealtimeMetrics
+			mustIAM(t, json.Unmarshal(response.Body.Bytes(), &legacy))
+			if legacy.Aggregated.BatchJobs.Jobs[job.ID].JobID != job.ID || legacy.Aggregated.BatchJobs.Jobs[job.ID].Complete != report.Complete {
+				t.Fatal("additive details changed legacy metrics")
+			}
+		})
+	}
+}
+
+func TestBatchMetricsPeerRoundTrip(t *testing.T) {
+	report := &batchJobInfo{JobID: "remote-job", LastUpdate: UTCNow(), PausedReason: "IAM unavailable", VersionsSkippedLocked: 3, KeysRefused: 2}
+	remote := realtimeMetrics{}
+	remote.Hosts = []string{"peer"}
+	remote.Aggregated.BatchJobs = &batchJobMetricsReport{
+		CollectedAt: UTCNow(), Jobs: map[string]batchJobStatusMetric{report.JobID: report.statusMetric()},
+	}
+	remote.ByHost = map[string]realtimeMetricGroups{"peer": remote.Aggregated}
+	encoded := madminRealtimeMetrics.NewJSONWith(&remote)
+	data, err := encoded.MarshalMsg(nil)
+	mustIAM(t, err)
+	decoded := madminRealtimeMetrics.NewJSON()
+	defer decoded.Recycle()
+	_, err = decoded.UnmarshalMsg(data)
+	mustIAM(t, err)
+	var merged realtimeMetrics
+	received := decoded.ValueOrZero()
+	merged.Merge(&received)
+	got := merged.Aggregated.BatchJobs.Jobs[report.JobID]
+	if got.PausedReason != report.PausedReason || got.KeysRefused != 2 || got.VersionsSkippedLocked == nil || *got.VersionsSkippedLocked != 3 {
+		t.Fatalf("peer aggregation lost details: %+v", got)
+	}
+	older := received
+	older.Aggregated.BatchJobs = &batchJobMetricsReport{Jobs: map[string]batchJobStatusMetric{
+		report.JobID: {JobMetric: madmin.JobMetric{JobID: report.JobID, LastUpdate: report.LastUpdate.Add(-time.Second), Complete: true}},
+	}}
+	merged.Merge(&older)
+	if got := merged.Aggregated.BatchJobs.Jobs[report.JobID]; got.Complete || got.PausedReason != report.PausedReason {
+		t.Fatal("older peer snapshot replaced current progress")
+	}
+	legacyPool := grid.NewJSONPool[madmin.RealtimeMetrics]()
+	legacy := legacyPool.NewJSON()
+	defer legacy.Recycle()
+	_, err = legacy.UnmarshalMsg(data)
+	mustIAM(t, err)
+	if got := legacy.ValueOrZero().Aggregated.BatchJobs.Jobs[report.JobID]; got.JobID != report.JobID || !got.LastUpdate.Equal(report.LastUpdate) {
+		t.Fatal("legacy peer could not read base metrics")
+	}
+	legacyData, err := legacy.MarshalMsg(nil)
+	mustIAM(t, err)
+	_, err = decoded.UnmarshalMsg(legacyData)
+	mustIAM(t, err)
+	received = decoded.ValueOrZero()
+	if received.Aggregated.BatchJobs.Jobs[report.JobID].VersionsSkippedLocked != nil {
+		t.Fatal("legacy peer falsely advertised complete status details")
+	}
+}
+
+func TestBatchMetricsPreservesOtherMetrics(t *testing.T) {
+	var legacy madmin.RealtimeMetrics
+	var extended realtimeMetrics
+	for _, host := range []string{"first", "second"} {
+		groups := madmin.Metrics{
+			Disk: &madmin.DiskMetric{CollectedAt: time.Unix(1, 0).UTC(), NDisks: 2, Offline: 1},
+			CPU:  &madmin.CPUMetrics{CPUCount: 4},
+		}
+		sample := madmin.RealtimeMetrics{
+			Hosts: []string{host}, Errors: []string{host + ": test"},
+			Aggregated: groups, ByHost: map[string]madmin.Metrics{host: groups},
+			ByDisk: map[string]madmin.DiskMetric{host: *groups.Disk},
+		}
+		data, err := json.Marshal(sample)
+		mustIAM(t, err)
+		var decoded realtimeMetrics
+		mustIAM(t, json.Unmarshal(data, &decoded))
+		extended.Merge(&decoded)
+		legacy.Merge(&sample)
+	}
+	if got := extended.legacy(); !reflect.DeepEqual(got, legacy) {
+		t.Fatalf("batch extension changed other metrics: got=%+v want=%+v", got, legacy)
+	}
+}

cmd/metrics-realtime.go

@@ -38,7 +38,7 @@ type collectMetricsOpts struct {
 	depID string
 }
 
-func collectLocalMetrics(types madmin.MetricType, opts collectMetricsOpts) (m madmin.RealtimeMetrics) {
+func collectLocalMetrics(types madmin.MetricType, opts collectMetricsOpts) (m realtimeMetrics) {
 	if types == madmin.MetricsNone {
 		return m
 	}
@@ -144,7 +144,7 @@ func collectLocalMetrics(types madmin.MetricType, opts collectMetricsOpts) (m ma
 	// Add types...
 
 	// ByHost is a shallow reference, so careful about sharing.
-	m.ByHost = map[string]madmin.Metrics{byHostName: m.Aggregated}
+	m.ByHost = map[string]realtimeMetricGroups{byHostName: m.Aggregated}
 	m.Hosts = append(m.Hosts, byHostName)
 
 	return m
@@ -221,11 +221,11 @@ func collectLocalDisksMetrics(disks map[string]struct{}) map[string]madmin.DiskM
 	return metrics
 }
 
-func collectRemoteMetrics(ctx context.Context, types madmin.MetricType, opts collectMetricsOpts) (m madmin.RealtimeMetrics) {
+func collectRemoteMetrics(ctx context.Context, types madmin.MetricType, opts collectMetricsOpts) (m realtimeMetrics) {
 	if !globalIsDistErasure {
 		return m
 	}
-	all := globalNotificationSys.GetMetrics(ctx, types, opts)
+	all := globalNotificationSys.getMetrics(ctx, types, opts)
 	for _, remote := range all {
 		m.Merge(&remote)
 	}

cmd/metrics-resource.go

@@ -246,7 +246,7 @@ func updateDriveIOStats(currentStats madmin.DiskIOStats, latestStats madmin.Disk
 	updateResourceMetrics(driveSubsystem, percUtil, float64(diffStats.TotalTicks)/(diffInSeconds*10), labels, false)
 }
 
-func collectDriveMetrics(m madmin.RealtimeMetrics) {
+func collectDriveMetrics(m realtimeMetrics) {
 	latestDriveStatsMu.Lock()
 	for d, dm := range m.ByDisk {
 		labels := map[string]string{"drive": d}

cmd/metrics-v2.go

@@ -3804,7 +3804,7 @@ func getBatchJobsMetrics(opts MetricsGroupOpts) *MetricsGroupV2 {
 	}
 
 	mg.RegisterRead(func(ctx context.Context) (metrics []MetricV2) {
-		var m madmin.RealtimeMetrics
+		var m realtimeMetrics
 		mLocal := collectLocalMetrics(madmin.MetricsBatchJobs, collectMetricsOpts{})
 		m.Merge(&mLocal)
 

cmd/notification.go

@@ -873,9 +873,9 @@ func (sys *NotificationSys) GetOSInfo(ctx context.Context) []madmin.OSInfo {
 	return reply
 }
 
-// GetMetrics - Get metrics from all peers.
-func (sys *NotificationSys) GetMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) []madmin.RealtimeMetrics {
-	reply := make([]madmin.RealtimeMetrics, len(sys.peerClients))
+// getMetrics - Get metrics from all peers.
+func (sys *NotificationSys) getMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) []realtimeMetrics {
+	reply := make([]realtimeMetrics, len(sys.peerClients))
 
 	g := errgroup.WithNErrs(len(sys.peerClients))
 	for index, client := range sys.peerClients {

cmd/peer-rest-client.go

@@ -228,7 +228,7 @@ func (client *peerRESTClient) GetMemInfo(ctx context.Context) (info madmin.MemIn
 }
 
 // GetMetrics - fetch metrics from a remote node.
-func (client *peerRESTClient) GetMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) (info madmin.RealtimeMetrics, err error) {
+func (client *peerRESTClient) GetMetrics(ctx context.Context, t madmin.MetricType, opts collectMetricsOpts) (info realtimeMetrics, err error) {
 	values := make(url.Values)
 	values.Set(peerRESTMetricsTypes, strconv.FormatUint(uint64(t), 10))
 	for disk := range opts.disks {

cmd/peer-rest-server.go

@@ -62,7 +62,7 @@ var (
 	madminOSInfo           = grid.NewJSONPool[madmin.OSInfo]()
 	madminPartitions       = grid.NewJSONPool[madmin.Partitions]()
 	madminProcInfo         = grid.NewJSONPool[madmin.ProcInfo]()
-	madminRealtimeMetrics  = grid.NewJSONPool[madmin.RealtimeMetrics]()
+	madminRealtimeMetrics  = grid.NewJSONPool[realtimeMetrics]()
 	madminServerProperties = grid.NewJSONPool[madmin.ServerProperties]()
 	madminStorageInfo      = grid.NewJSONPool[madmin.StorageInfo]()
 	madminSysConfig        = grid.NewJSONPool[madmin.SysConfig]()
@@ -84,7 +84,7 @@ var (
 	getLocksRPC                    = grid.NewSingleHandler[*grid.MSS, *localLockMap](grid.HandlerGetLocks, grid.NewMSS, func() *localLockMap { return &localLockMap{} })
 	getMemInfoRPC                  = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.MemInfo]](grid.HandlerGetMemInfo, grid.NewMSS, madminMemInfo.NewJSON)
 	getMetacacheListingRPC         = grid.NewSingleHandler[*listPathOptions, *metacache](grid.HandlerGetMetacacheListing, func() *listPathOptions { return &listPathOptions{} }, func() *metacache { return &metacache{} })
-	getMetricsRPC                  = grid.NewSingleHandler[*grid.URLValues, *grid.JSON[madmin.RealtimeMetrics]](grid.HandlerGetMetrics, grid.NewURLValues, madminRealtimeMetrics.NewJSON)
+	getMetricsRPC                  = grid.NewSingleHandler[*grid.URLValues, *grid.JSON[realtimeMetrics]](grid.HandlerGetMetrics, grid.NewURLValues, madminRealtimeMetrics.NewJSON)
 	getNetInfoRPC                  = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.NetInfo]](grid.HandlerGetNetInfo, grid.NewMSS, madminNetInfo.NewJSON)
 	getOSInfoRPC                   = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.OSInfo]](grid.HandlerGetOSInfo, grid.NewMSS, madminOSInfo.NewJSON)
 	getPartitionsRPC               = grid.NewSingleHandler[*grid.MSS, *grid.JSON[madmin.Partitions]](grid.HandlerGetPartitions, grid.NewMSS, madminPartitions.NewJSON)
@@ -423,7 +423,7 @@ func (s *peerRESTServer) GetMemInfoHandler(_ *grid.MSS) (*grid.JSON[madmin.MemIn
 }
 
 // GetMetricsHandler - returns server metrics.
-func (s *peerRESTServer) GetMetricsHandler(v *grid.URLValues) (*grid.JSON[madmin.RealtimeMetrics], *grid.RemoteErr) {
+func (s *peerRESTServer) GetMetricsHandler(v *grid.URLValues) (*grid.JSON[realtimeMetrics], *grid.RemoteErr) {
 	values := v.Values()
 	var types madmin.MetricType
 	if t, _ := strconv.ParseUint(values.Get(peerRESTMetricsTypes), 10, 64); t != 0 {