From c8561031ec450fa2b96400615f149272bb5208f2 Mon Sep 17 00:00:00 2001
From: igroene <9805809+igroene@users.noreply.github.com>
Date: Fri, 3 Jul 2026 14:35:14 -0300
Subject: [PATCH 1/9] PBM-1534 progress indicators

---
 cmd/pbm/main.go              |   1 +
 cmd/pbm/status.go            | 103 +++++++++++++++--
 pbm/backup/backup.go         |  12 ++
 pbm/backup/logical.go        |  19 +++-
 pbm/backup/physical.go       |  45 +++++++-
 pbm/backup/query.go          |  53 ++++++---
 pbm/backup/types.go          |  37 ++++---
 pbm/progress/progress.go     | 208 +++++++++++++++++++++++++++++++++++
 pbm/restore/logical.go       |  65 +++++++++--
 pbm/restore/physical.go      |  58 ++++++++--
 pbm/restore/physical_test.go |   4 +-
 pbm/restore/query.go         |  51 +++++++--
 pbm/restore/types.go         |  37 ++++---
 pbm/snapshot/dump.go         |  47 ++++++++
 14 files changed, 645 insertions(+), 95 deletions(-)
 create mode 100644 pbm/progress/progress.go

diff --git a/cmd/pbm/main.go b/cmd/pbm/main.go
index e752107b1..02b69428d 100644
--- a/cmd/pbm/main.go
+++ b/cmd/pbm/main.go
@@ -1202,6 +1202,7 @@ type snapshotStat struct {
 	Err         error            `json:"-"`
 	ErrString   string           `json:"error,omitempty"`
 	RestoreTS   int64            `json:"restoreTo"`
+	Duration    int64            `json:"duration,omitempty"`
 	PBMVersion  string           `json:"pbmVersion"`
 	Type        defs.BackupType  `json:"type"`
 	SrcBackup   string           `json:"src"`
diff --git a/cmd/pbm/status.go b/cmd/pbm/status.go
index 6756b2467..cc5156de6 100644
--- a/cmd/pbm/status.go
+++ b/cmd/pbm/status.go
@@ -19,6 +19,8 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
 	"github.com/percona/percona-backup-mongodb/pbm/oplog"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
+	"github.com/percona/percona-backup-mongodb/pbm/restore"
 	"github.com/percona/percona-backup-mongodb/pbm/slicer"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
@@ -372,11 +374,13 @@ LOOP:
 }
 
 type currOp struct {
-	Type    ctrl.Command `json:"type,omitempty"`
-	OPID    string       `json:"opID,omitempty"`
-	Name    string       `json:"name,omitempty"`
-	StartTS int64        `json:"startTS,omitempty"`
-	Status  string       `json:"status,omitempty"`
+	Type     ctrl.Command       `json:"type,omitempty"`
+	OPID     string             `json:"opID,omitempty"`
+	Name     string             `json:"name,omitempty"`
+	StartTS  int64              `json:"startTS,omitempty"`
+	Duration int64              `json:"duration,omitempty"`
+	Status   string             `json:"status,omitempty"`
+	Progress *progress.Progress `json:"progress,omitempty"`
 }
 
 func (c currOp) String() string {
@@ -388,10 +392,14 @@ func (c currOp) String() string {
 	default:
 		return fmt.Sprintf("%s [op id: %s]", c.Type, c.OPID)
 	case ctrl.CmdBackup, ctrl.CmdRestore:
-		return fmt.Sprintf("%s \"%s\", started at %s. Status: %s. [op id: %s]",
+		s := fmt.Sprintf("%s \"%s\", started at %s. Status: %s. Duration: %s. [op id: %s]",
 			c.Type, c.Name, time.Unix((c.StartTS), 0).UTC().Format("2006-01-02T15:04:05Z"),
-			c.Status, c.OPID,
+			c.Status, progress.FormatDuration(time.Duration(c.Duration)*time.Second), c.OPID,
 		)
+		if c.Progress != nil {
+			s += ". Progress: " + c.Progress.StringAt(time.Now().Unix())
+		}
+		return s
 	}
 }
 
@@ -418,6 +426,8 @@ func getCurrOps(ctx context.Context, pbm *sdk.Client) (fmt.Stringer, error) {
 
 		r.Name = bcp.Name
 		r.StartTS = bcp.StartTS
+		r.Duration = durationSeconds(bcp.StartTS, 0)
+		r.Progress = backupProgress(bcp)
 
 		switch bcp.Status {
 		case defs.StatusRunning:
@@ -435,6 +445,8 @@ func getCurrOps(ctx context.Context, pbm *sdk.Client) (fmt.Stringer, error) {
 
 		r.Name = rst.Backup
 		r.StartTS = rst.StartTS
+		r.Duration = durationSeconds(rst.StartTS, 0)
+		r.Progress = restoreProgress(rst)
 
 		switch rst.Status {
 		case defs.StatusRunning:
@@ -449,6 +461,73 @@ func getCurrOps(ctx context.Context, pbm *sdk.Client) (fmt.Stringer, error) {
 	return r, nil
 }
 
+func backupProgress(bcp *backup.BackupMeta) *progress.Progress {
+	p := combineProgress(bcp.Progress, func(yield func(*progress.Progress)) {
+		for i := range bcp.Replsets {
+			yield(bcp.Replsets[i].Progress)
+		}
+	})
+	return p
+}
+
+func restoreProgress(rst *restore.RestoreMeta) *progress.Progress {
+	p := combineProgress(rst.Progress, func(yield func(*progress.Progress)) {
+		for i := range rst.Replsets {
+			yield(rst.Replsets[i].Progress)
+		}
+	})
+	return p
+}
+
+func combineProgress(fallback *progress.Progress, each func(func(*progress.Progress))) *progress.Progress {
+	var rv *progress.Progress
+	each(func(p *progress.Progress) {
+		if p == nil {
+			return
+		}
+		if rv == nil {
+			cp := *p
+			rv = &cp
+			return
+		}
+		if p.StartedAt > 0 && (rv.StartedAt == 0 || p.StartedAt < rv.StartedAt) {
+			rv.StartedAt = p.StartedAt
+		}
+		if p.UpdatedAt > rv.UpdatedAt {
+			rv.UpdatedAt = p.UpdatedAt
+		}
+		rv.DoneBytes += p.DoneBytes
+		rv.TotalBytes += p.TotalBytes
+		rv.DoneItems += p.DoneItems
+		rv.TotalItems += p.TotalItems
+	})
+	if rv != nil {
+		return rv
+	}
+	return fallback
+}
+
+func durationSeconds(start, end int64) int64 {
+	if start <= 0 {
+		return 0
+	}
+	if end <= 0 {
+		end = time.Now().Unix()
+	}
+	if end < start {
+		return 0
+	}
+	return end - start
+}
+
+func backupDuration(bcp backup.BackupMeta, now int64) int64 {
+	end := bcp.LastTransitionTS
+	if bcp.Status.IsRunning() {
+		end = now
+	}
+	return durationSeconds(bcp.StartTS, end)
+}
+
 type storageStat struct {
 	Type     string         `json:"type"`
 	Path     string         `json:"path"`
@@ -482,9 +561,9 @@ func (s storageStat) String() string {
 		return a.RestoreTS > b.RestoreTS
 	})
 
-	ret += fmt.Sprintf("  %-24s  %-10s  %-12s  %-20s  %-5s  %-4s  %-19s  %s\n",
-		"NAME", "SIZE", "TYPE", "PROFILE", "SEL", "BASE", "RESTORE TIME", "STATUS")
-	ret += fmt.Sprintf("  %s\n", strings.Repeat("-", 24+10+12+20+5+4+19+6+(7*2)))
+	ret += fmt.Sprintf("  %-24s  %-10s  %-12s  %-20s  %-5s  %-4s  %-19s  %-10s  %s\n",
+		"NAME", "SIZE", "TYPE", "PROFILE", "SEL", "BASE", "RESTORE TIME", "DURATION", "STATUS")
+	ret += fmt.Sprintf("  %s\n", strings.Repeat("-", 24+10+12+20+5+4+19+10+6+(8*2)))
 
 	for i := range s.Snapshot {
 		ss := &s.Snapshot[i]
@@ -523,7 +602,7 @@ func (s storageStat) String() string {
 			status = strings.TrimRight(status[:maxStatusLen-3], " ") + "..."
 		}
 
-		ret += fmt.Sprintf("  %-24s  %-10s  %-12s  %-20s  %-5s  %-4s  %-19s  %s\n",
+		ret += fmt.Sprintf("  %-24s  %-10s  %-12s  %-20s  %-5s  %-4s  %-19s  %-10s  %s\n",
 			ss.Name,
 			storage.PrettySize(ss.Size),
 			bcpType,
@@ -531,6 +610,7 @@ func (s storageStat) String() string {
 			selective,
 			base,
 			fmtTS(ss.RestoreTS),
+			progress.FormatDuration(time.Duration(ss.Duration)*time.Second),
 			status)
 	}
 
@@ -625,6 +705,7 @@ func getStorageStat(
 			SrcBackup:  bcp.SrcBackup,
 			Profile:    bcp.Store.Name,
 			StoreName:  bcp.Store.Name,
+			Duration:   backupDuration(bcp, int64(now.T)),
 		}
 		if err := bcp.Error(); err != nil {
 			snpsht.Err = err
diff --git a/pbm/backup/backup.go b/pbm/backup/backup.go
index 4977331f2..a7640f608 100644
--- a/pbm/backup/backup.go
+++ b/pbm/backup/backup.go
@@ -17,6 +17,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/lock"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
 	"github.com/percona/percona-backup-mongodb/pbm/oplog"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 	"github.com/percona/percona-backup-mongodb/pbm/util"
@@ -185,6 +186,16 @@ func (b *Backup) Init(
 //
 //nolint:nonamedreturns
 func (b *Backup) Run(ctx context.Context, bcp *ctrl.BackupCmd, opid ctrl.OPID, l log.LogEvent) (err error) {
+	opStarted := time.Now()
+	defer func() {
+		elapsed := progress.FormatDuration(time.Since(opStarted))
+		if err != nil {
+			l.Info("backup failed after %s: %v", elapsed, err)
+			return
+		}
+		l.Info("backup completed after %s", elapsed)
+	}()
+
 	inf, err := topo.GetNodeInfoExt(ctx, b.nodeConn)
 	if err != nil {
 		return errors.Wrap(err, "get cluster info")
@@ -384,6 +395,7 @@ func (b *Backup) Run(ctx context.Context, bcp *ctrl.BackupCmd, opid ctrl.OPID, l
 		// PBM-1114: update file metadata with the same values as in database
 		unix := time.Now().Unix()
 		bcpm.Status = defs.StatusDone
+		bcpm.Progress = nil
 		bcpm.LastTransitionTS = unix
 		bcpm.Conditions = append(bcpm.Conditions, Condition{
 			Timestamp: unix,
diff --git a/pbm/backup/logical.go b/pbm/backup/logical.go
index e3c3b8358..640185ff4 100644
--- a/pbm/backup/logical.go
+++ b/pbm/backup/logical.go
@@ -20,6 +20,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/snapshot"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
@@ -64,6 +65,11 @@ func (b *Backup) doLogical(
 	if err != nil {
 		return errors.Wrap(err, "add shard's metadata")
 	}
+	reporter := progress.NewReporter(ctx, l, time.Minute, 0, int64(len(nssSize)),
+		func(ctx context.Context, p progress.Progress) error {
+			return SetRSProgress(ctx, b.leadConn, bcp.Name, rsMeta.Name, p)
+		})
+	defer reporter.Close("backup transfer finished")
 
 	if inf.IsLeader() {
 		err := b.reconcileStatus(ctx,
@@ -178,7 +184,7 @@ func (b *Backup) doLogical(
 		}
 	}
 
-	snapshotSize, err := snapshot.UploadDump(ctx,
+	snapshotSize, err := snapshot.UploadDumpWithProgress(ctx,
 		func(newFile archive.NewWriter) error {
 			bcp, err := archive.NewBackup(ctx, archive.BackupOptions{
 				Client:        b.nodeConn,
@@ -202,10 +208,19 @@ func (b *Backup) doLogical(
 			return stg.Save(filepath, r, storage.Size(sizeHints[ns]))
 		},
 		bcp.Compression,
-		bcp.CompressionLevel)
+		bcp.CompressionLevel,
+		func(ns string, bytes int64) {
+			reporter.AddBytes(bytes)
+			if ns != archive.MetaFileV2 {
+				reporter.AddItems(1)
+			}
+		})
 	if err != nil {
 		return errors.Wrap(err, "dump")
 	}
+	if err := reporter.Flush(); err != nil {
+		l.Warning("update progress: %v", err)
+	}
 
 	err = archive.GenerateV1FromV2(ctx, stg, bcp.Name, rsMeta.Name)
 	if err != nil {
diff --git a/pbm/backup/physical.go b/pbm/backup/physical.go
index 96d93e548..eedfed28f 100644
--- a/pbm/backup/physical.go
+++ b/pbm/backup/physical.go
@@ -22,6 +22,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 	"github.com/percona/percona-backup-mongodb/pbm/util"
@@ -328,7 +329,6 @@ func (b *Backup) doPhysical(
 	if err != nil {
 		return errors.Wrap(err, "add shard's metadata")
 	}
-
 	if inf.IsLeader() {
 		err := b.reconcileStatus(ctx,
 			bcp.Name, opid.String(), defs.StatusRunning, util.Ref(b.timeouts.StartingStatus()))
@@ -382,7 +382,14 @@ func (b *Backup) doPhysical(
 		return b.handleExternal(ctx, bcp, rsMeta, data, jrnls, bcur.Meta.DBpath, opid, inf, stg, l)
 	}
 
-	return b.uploadPhysical(ctx, bcp, rsMeta, data, jrnls, bcur.Meta.DBpath, stg, l)
+	reporter := progress.NewReporter(ctx, l, time.Minute,
+		plannedUploadSize(data, b.typ == defs.IncrementalBackup)+plannedUploadSize(jrnls, false), 0,
+		func(ctx context.Context, p progress.Progress) error {
+			return SetRSProgress(ctx, b.leadConn, bcp.Name, rsMeta.Name, p)
+		})
+	defer reporter.Close("backup transfer finished")
+
+	return b.uploadPhysical(ctx, bcp, rsMeta, data, jrnls, bcur.Meta.DBpath, stg, l, reporter)
 }
 
 func (b *Backup) handleExternal(
@@ -519,6 +526,7 @@ func (b *Backup) uploadPhysical(
 	dbpath string,
 	stg storage.Storage,
 	l log.LogEvent,
+	reporter *progress.Reporter,
 ) error {
 	numWorkers := b.getNumParallelFiles()
 	if numWorkers > 1 {
@@ -538,6 +546,7 @@ func (b *Backup) uploadPhysical(
 		bcp.CompressionLevel,
 		b.getBackupBufSize(),
 		numWorkers,
+		reporter,
 	)
 	if err != nil {
 		return errors.Wrap(err, "upload data files")
@@ -556,6 +565,7 @@ func (b *Backup) uploadPhysical(
 		bcp.CompressionLevel,
 		b.getBackupBufSize(),
 		numWorkers,
+		reporter,
 	)
 	if err != nil {
 		return errors.Wrap(err, "upload journal files")
@@ -583,6 +593,12 @@ func (b *Backup) uploadPhysical(
 		return errors.Wrapf(err, "upload filelist %q", filelistPath)
 	}
 	l.Info("uploaded: %q %s", filelistPath, storage.PrettySize(flSize))
+	if reporter != nil {
+		reporter.AddBytes(flSize)
+		if err := reporter.Flush(); err != nil {
+			l.Warning("update progress: %v", err)
+		}
+	}
 
 	totalSize := size + flSize
 	totalUncompressed := sizeUncompressed + flSize
@@ -611,6 +627,27 @@ func (b *Backup) uploadPhysical(
 	return nil
 }
 
+func plannedUploadSize(files []File, incr bool) int64 {
+	var size int64
+	for _, item := range planUploads(files, incr) {
+		if !item.upload {
+			continue
+		}
+		size += sourceFileSize(item.file)
+	}
+	return size
+}
+
+func sourceFileSize(f File) int64 {
+	if f.Len > 0 {
+		if f.Off+f.Len > f.Size {
+			return f.Size - f.Off
+		}
+		return f.Len
+	}
+	return f.Size
+}
+
 const storagebson = "storage.bson"
 
 func getStorageBSON(dbpath string) (*File, error) {
@@ -734,6 +771,7 @@ func uploadFiles(
 	comprL *int,
 	bufSize int,
 	numWorkers int,
+	reporter *progress.Reporter,
 ) ([]File, error) {
 	if len(files) == 0 {
 		return nil, nil
@@ -793,6 +831,9 @@ func uploadFiles(
 			fw.Name = fname
 
 			results[i] = *fw
+			if reporter != nil {
+				reporter.AddBytes(fw.StgSize)
+			}
 			return nil
 		})
 	}
diff --git a/pbm/backup/query.go b/pbm/backup/query.go
index 01046e0ab..77a438cf7 100644
--- a/pbm/backup/query.go
+++ b/pbm/backup/query.go
@@ -11,6 +11,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/connect"
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 )
 
@@ -97,7 +98,7 @@ func ChangeBackupStateWithUnixTime(
 	unix int64,
 	msg string,
 ) error {
-	return changeBackupState(ctx, conn, bson.D{{"name", bcpName}}, time.Now().UTC().Unix(), s, msg)
+	return changeBackupState(ctx, conn, bson.D{{"name", bcpName}}, unix, s, msg)
 }
 
 func changeBackupState(
@@ -108,15 +109,20 @@ func changeBackupState(
 	s defs.Status,
 	msg string,
 ) error {
+	update := bson.D{
+		{"$set", bson.M{"status": s}},
+		{"$set", bson.M{"last_transition_ts": ts}},
+		{"$set", bson.M{"error": msg}},
+		{"$push", bson.M{"conditions": Condition{Timestamp: ts, Status: s, Error: msg}}},
+	}
+	if !s.IsRunning() {
+		update = append(update, bson.E{"$unset", bson.M{"progress": ""}})
+	}
+
 	_, err := conn.BcpCollection().UpdateOne(
 		ctx,
 		clause,
-		bson.D{
-			{"$set", bson.M{"status": s}},
-			{"$set", bson.M{"last_transition_ts": ts}},
-			{"$set", bson.M{"error": msg}},
-			{"$push", bson.M{"conditions": Condition{Timestamp: ts, Status: s, Error: msg}}},
-		},
+		update,
 	)
 
 	return err
@@ -204,20 +210,41 @@ func AddRSMeta(ctx context.Context, conn connect.Client, bcpName string, rs Back
 
 func ChangeRSState(conn connect.Client, bcpName, rsName string, s defs.Status, msg string) error {
 	ts := time.Now().UTC().Unix()
+	update := bson.D{
+		{"$set", bson.M{"replsets.$.status": s}},
+		{"$set", bson.M{"replsets.$.last_transition_ts": ts}},
+		{"$set", bson.M{"replsets.$.error": msg}},
+		{"$push", bson.M{"replsets.$.conditions": Condition{Timestamp: ts, Status: s, Error: msg}}},
+	}
+	if !s.IsRunning() {
+		update = append(update, bson.E{"$unset", bson.M{"replsets.$.progress": ""}})
+	}
+
 	_, err := conn.BcpCollection().UpdateOne(
 		context.Background(),
 		bson.D{{"name", bcpName}, {"replsets.name", rsName}},
-		bson.D{
-			{"$set", bson.M{"replsets.$.status": s}},
-			{"$set", bson.M{"replsets.$.last_transition_ts": ts}},
-			{"$set", bson.M{"replsets.$.error": msg}},
-			{"$push", bson.M{"replsets.$.conditions": Condition{Timestamp: ts, Status: s, Error: msg}}},
-		},
+		update,
 	)
 
 	return err
 }
 
+func SetBackupProgress(ctx context.Context, conn connect.Client, bcpName string, p progress.Progress) error {
+	_, err := conn.BcpCollection().UpdateOne(ctx,
+		bson.D{{"name", bcpName}},
+		bson.D{{"$set", bson.M{"progress": p}}},
+	)
+	return err
+}
+
+func SetRSProgress(ctx context.Context, conn connect.Client, bcpName, rsName string, p progress.Progress) error {
+	_, err := conn.BcpCollection().UpdateOne(ctx,
+		bson.D{{"name", bcpName}, {"replsets.name", rsName}},
+		bson.D{{"$set", bson.M{"replsets.$.progress": p}}},
+	)
+	return err
+}
+
 // IncBackupSize increments total backup size.
 func IncBackupSize(
 	ctx context.Context,
diff --git a/pbm/backup/types.go b/pbm/backup/types.go
index 5c5bcefdf..24ca57bc0 100644
--- a/pbm/backup/types.go
+++ b/pbm/backup/types.go
@@ -13,6 +13,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/config"
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 )
 
@@ -53,6 +54,7 @@ type BackupMeta struct {
 	LastWriteTS      bson.Timestamp           `bson:"last_write_ts" json:"last_write_ts"`
 	Hb               bson.Timestamp           `bson:"hb" json:"hb"`
 	Status           defs.Status              `bson:"status" json:"status"`
+	Progress         *progress.Progress       `bson:"progress,omitempty" json:"progress,omitempty"`
 	Conditions       []Condition              `bson:"conditions" json:"conditions"`
 	Nomination       []BackupRsNomination     `bson:"n" json:"n"`
 	Err              string                   `bson:"error,omitempty" json:"error,omitempty"`
@@ -112,23 +114,24 @@ type BackupReplset struct {
 	Name string `bson:"name" json:"name"`
 
 	// Journal is not used. left for backward compatibility
-	Journal          []File           `bson:"journal,omitempty" json:"journal,omitempty"`
-	Files            []File           `bson:"files,omitempty" json:"files,omitempty"`
-	DumpName         string           `bson:"dump_name,omitempty" json:"backup_name,omitempty"`
-	OplogName        string           `bson:"oplog_name,omitempty" json:"oplog_name,omitempty"`
-	StartTS          int64            `bson:"start_ts" json:"start_ts"`
-	Status           defs.Status      `bson:"status" json:"status"`
-	Size             int64            `bson:"size" json:"size"`
-	SizeUncompressed int64            `bson:"size_uncompressed" json:"size_uncompressed"`
-	IsConfigSvr      *bool            `bson:"iscs,omitempty" json:"iscs,omitempty"`
-	IsConfigShard    *bool            `bson:"configshard,omitempty" json:"configshard,omitempty"`
-	LastTransitionTS int64            `bson:"last_transition_ts" json:"last_transition_ts"`
-	FirstWriteTS     bson.Timestamp   `bson:"first_write_ts" json:"first_write_ts"`
-	LastWriteTS      bson.Timestamp   `bson:"last_write_ts" json:"last_write_ts"`
-	Node             string           `bson:"node" json:"node"` // node that performed backup
-	Error            string           `bson:"error,omitempty" json:"error,omitempty"`
-	Conditions       []Condition      `bson:"conditions" json:"conditions"`
-	MongodOpts       *topo.MongodOpts `bson:"mongod_opts,omitempty" json:"mongod_opts,omitempty"`
+	Journal          []File             `bson:"journal,omitempty" json:"journal,omitempty"`
+	Files            []File             `bson:"files,omitempty" json:"files,omitempty"`
+	DumpName         string             `bson:"dump_name,omitempty" json:"backup_name,omitempty"`
+	OplogName        string             `bson:"oplog_name,omitempty" json:"oplog_name,omitempty"`
+	StartTS          int64              `bson:"start_ts" json:"start_ts"`
+	Status           defs.Status        `bson:"status" json:"status"`
+	Progress         *progress.Progress `bson:"progress,omitempty" json:"progress,omitempty"`
+	Size             int64              `bson:"size" json:"size"`
+	SizeUncompressed int64              `bson:"size_uncompressed" json:"size_uncompressed"`
+	IsConfigSvr      *bool              `bson:"iscs,omitempty" json:"iscs,omitempty"`
+	IsConfigShard    *bool              `bson:"configshard,omitempty" json:"configshard,omitempty"`
+	LastTransitionTS int64              `bson:"last_transition_ts" json:"last_transition_ts"`
+	FirstWriteTS     bson.Timestamp     `bson:"first_write_ts" json:"first_write_ts"`
+	LastWriteTS      bson.Timestamp     `bson:"last_write_ts" json:"last_write_ts"`
+	Node             string             `bson:"node" json:"node"` // node that performed backup
+	Error            string             `bson:"error,omitempty" json:"error,omitempty"`
+	Conditions       []Condition        `bson:"conditions" json:"conditions"`
+	MongodOpts       *topo.MongodOpts   `bson:"mongod_opts,omitempty" json:"mongod_opts,omitempty"`
 
 	// required for external backup (PBM-1252)
 	PBMVersion   string `bson:"pbm_version,omitempty" json:"pbm_version,omitempty"`
diff --git a/pbm/progress/progress.go b/pbm/progress/progress.go
new file mode 100644
index 000000000..e47873c6f
--- /dev/null
+++ b/pbm/progress/progress.go
@@ -0,0 +1,208 @@
+package progress
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/percona/percona-backup-mongodb/pbm/storage"
+)
+
+const mb = 1024 * 1024
+
+// Progress captures an operation progress snapshot. Byte counters represent
+// bytes transferred to/from backup storage.
+type Progress struct {
+	StartedAt  int64 `bson:"started_at" json:"started_at"`
+	UpdatedAt  int64 `bson:"updated_at" json:"updated_at"`
+	DoneBytes  int64 `bson:"done_bytes,omitempty" json:"done_bytes,omitempty"`
+	TotalBytes int64 `bson:"total_bytes,omitempty" json:"total_bytes,omitempty"`
+	DoneItems  int64 `bson:"done_items,omitempty" json:"done_items,omitempty"`
+	TotalItems int64 `bson:"total_items,omitempty" json:"total_items,omitempty"`
+}
+
+func New(totalBytes, totalItems int64) Progress {
+	now := time.Now().Unix()
+	return Progress{StartedAt: now, UpdatedAt: now, TotalBytes: totalBytes, TotalItems: totalItems}
+}
+
+func (p Progress) Percent() (float64, bool) {
+	switch {
+	case p.TotalBytes > 0:
+		return percent(p.DoneBytes, p.TotalBytes), true
+	case p.TotalItems > 0:
+		return percent(p.DoneItems, p.TotalItems), true
+	default:
+		return 0, false
+	}
+}
+
+func (p Progress) ThroughputMBps(now int64) float64 {
+	if p.DoneBytes <= 0 || p.StartedAt <= 0 || now <= p.StartedAt {
+		return 0
+	}
+
+	return float64(p.DoneBytes) / float64(mb) / float64(now-p.StartedAt)
+}
+
+func (p Progress) Elapsed(now int64) time.Duration {
+	if p.StartedAt <= 0 || now <= p.StartedAt {
+		return 0
+	}
+	return time.Duration(now-p.StartedAt) * time.Second
+}
+
+func (p Progress) StringAt(now int64) string {
+	parts := []string{fmt.Sprintf("elapsed=%s", FormatDuration(p.Elapsed(now)))}
+	if p.TotalItems > 0 {
+		parts = append(parts, fmt.Sprintf("items=%d/%d", p.DoneItems, p.TotalItems))
+	}
+	if p.DoneBytes > 0 || p.TotalBytes > 0 {
+		b := storage.PrettySize(p.DoneBytes)
+		if p.TotalBytes > 0 {
+			b += "/" + storage.PrettySize(p.TotalBytes)
+		}
+		parts = append(parts, "transferred="+b)
+	}
+	if pct, ok := p.Percent(); ok {
+		parts = append(parts, fmt.Sprintf("done=%.1f%%", pct))
+	}
+	if mbps := p.ThroughputMBps(now); mbps > 0 {
+		parts = append(parts, fmt.Sprintf("throughput=%.2fMB/s", mbps))
+	}
+
+	return strings.Join(parts, ", ")
+}
+
+func FormatDuration(d time.Duration) string {
+	if d < 0 {
+		return "-"
+	}
+	d = d.Truncate(time.Second)
+	h := d / time.Hour
+	d -= h * time.Hour
+	m := d / time.Minute
+	d -= m * time.Minute
+	s := d / time.Second
+	if h > 0 {
+		return fmt.Sprintf("%dh%02dm%02ds", h, m, s)
+	}
+	if m > 0 {
+		return fmt.Sprintf("%dm%02ds", m, s)
+	}
+	return fmt.Sprintf("%ds", s)
+}
+
+func percent(done, total int64) float64 {
+	if total <= 0 {
+		return 0
+	}
+	p := float64(done) * 100 / float64(total)
+	if p > 100 {
+		return 100
+	}
+	return p
+}
+
+type Logger interface {
+	Info(msg string, args ...any)
+	Warning(msg string, args ...any)
+}
+
+// Reporter periodically persists and logs operation progress.
+type Reporter struct {
+	ctx    context.Context
+	log    Logger
+	update func(context.Context, Progress) error
+
+	startedAt int64
+	totalB    atomic.Int64
+	doneB     atomic.Int64
+	totalI    atomic.Int64
+	doneI     atomic.Int64
+	stop      chan struct{}
+}
+
+func NewReporter(
+	ctx context.Context,
+	log Logger,
+	interval time.Duration,
+	totalBytes int64,
+	totalItems int64,
+	update func(context.Context, Progress) error,
+) *Reporter {
+	r := &Reporter{ctx: ctx, log: log, update: update, startedAt: time.Now().Unix(), stop: make(chan struct{})}
+	r.totalB.Store(totalBytes)
+	r.totalI.Store(totalItems)
+	_ = r.Flush()
+
+	go func() {
+		tk := time.NewTicker(interval)
+		defer tk.Stop()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-r.stop:
+				return
+			case <-tk.C:
+				p := r.Snapshot()
+				if err := r.FlushProgress(p); err != nil && log != nil {
+					log.Warning("update progress: %v", err)
+				}
+				if log != nil {
+					log.Info("progress: %s", p.StringAt(time.Now().Unix()))
+				}
+			}
+		}
+	}()
+
+	return r
+}
+
+func (r *Reporter) AddBytes(n int64) {
+	if n > 0 {
+		r.doneB.Add(n)
+	}
+}
+
+func (r *Reporter) AddItems(n int64) {
+	if n > 0 {
+		r.doneI.Add(n)
+	}
+}
+
+func (r *Reporter) SetTotalBytes(n int64) { r.totalB.Store(n) }
+func (r *Reporter) SetTotalItems(n int64) { r.totalI.Store(n) }
+
+func (r *Reporter) Snapshot() Progress {
+	return Progress{
+		StartedAt:  r.startedAt,
+		UpdatedAt:  time.Now().Unix(),
+		DoneBytes:  r.doneB.Load(),
+		TotalBytes: r.totalB.Load(),
+		DoneItems:  r.doneI.Load(),
+		TotalItems: r.totalI.Load(),
+	}
+}
+
+func (r *Reporter) Flush() error {
+	return r.FlushProgress(r.Snapshot())
+}
+
+func (r *Reporter) FlushProgress(p Progress) error {
+	if r.update == nil {
+		return nil
+	}
+	return r.update(r.ctx, p)
+}
+
+func (r *Reporter) Close(final string) {
+	close(r.stop)
+	p := r.Snapshot()
+	if r.log != nil {
+		r.log.Info("%s after %s", final, FormatDuration(p.Elapsed(time.Now().Unix())))
+	}
+}
diff --git a/pbm/restore/logical.go b/pbm/restore/logical.go
index 1f485747d..015e47018 100644
--- a/pbm/restore/logical.go
+++ b/pbm/restore/logical.go
@@ -28,6 +28,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/lock"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
 	"github.com/percona/percona-backup-mongodb/pbm/oplog"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/restore/phys"
 	"github.com/percona/percona-backup-mongodb/pbm/snapshot"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
@@ -250,7 +251,16 @@ func (r *Restore) Snapshot(
 ) (err error) {
 	l := log.LogEventFromContext(ctx)
 
-	defer func() { r.exit(log.Copy(context.Background(), ctx), err) }()
+	opStarted := time.Now()
+	defer func() {
+		elapsed := progress.FormatDuration(time.Since(opStarted))
+		if err != nil {
+			l.Info("restore failed after %s: %v", elapsed, err)
+		} else {
+			l.Info("restore completed after %s", elapsed)
+		}
+		r.exit(log.Copy(context.Background(), ctx), err)
+	}()
 
 	err = r.init(ctx, cmd.Name, opid, l)
 	if err != nil {
@@ -394,7 +404,16 @@ func (r *Restore) PITR(
 ) (err error) {
 	l := log.LogEventFromContext(ctx)
 
-	defer func() { r.exit(log.Copy(context.Background(), ctx), err) }()
+	opStarted := time.Now()
+	defer func() {
+		elapsed := progress.FormatDuration(time.Since(opStarted))
+		if err != nil {
+			l.Info("restore failed after %s: %v", elapsed, err)
+		} else {
+			l.Info("restore completed after %s", elapsed)
+		}
+		r.exit(log.Copy(context.Background(), ctx), err)
+	}()
 
 	err = r.init(ctx, cmd.Name, opid, l)
 	if err != nil {
@@ -560,7 +579,16 @@ func (r *Restore) PITR(
 
 //nolint:nonamedreturns
 func (r *Restore) ReplayOplog(ctx context.Context, cmd *ctrl.ReplayCmd, opid ctrl.OPID, l log.LogEvent) (err error) {
-	defer func() { r.exit(log.Copy(context.Background(), ctx), err) }()
+	opStarted := time.Now()
+	defer func() {
+		elapsed := progress.FormatDuration(time.Since(opStarted))
+		if err != nil {
+			l.Info("restore failed after %s: %v", elapsed, err)
+		} else {
+			l.Info("restore completed after %s", elapsed)
+		}
+		r.exit(log.Copy(context.Background(), ctx), err)
+	}()
 
 	if err = r.init(ctx, cmd.Name, opid, l); err != nil {
 		return errors.Wrap(err, "init")
@@ -1063,8 +1091,13 @@ func (r *Restore) RunSnapshot(
 	mapRS := util.MakeReverseRSMapFunc(r.rsMap)
 
 	r.log.Debug("restoring up to %d collections in parallel", r.numParallelColls)
+	reporter := progress.NewReporter(ctx, r.log, time.Minute, 0, 0,
+		func(ctx context.Context, p progress.Progress) error {
+			return SetRestoreRSProgress(ctx, r.leadConn, r.name, r.nodeInfo.SetName, p)
+		})
+	defer reporter.Close("restore transfer finished")
 
-	rdr, err := snapshot.DownloadDump(
+	rdr, err := snapshot.DownloadDumpWithProgress(
 		func(ns string) (io.ReadCloser, error) {
 			stg, err := util.StorageFromConfig(&bcp.Store.StorageConf, r.brief.Me, r.log)
 			if err != nil {
@@ -1084,10 +1117,11 @@ func (r *Restore) RunSnapshot(
 					return nil, err
 				}
 
-				err = r.loadIndexesFrom(bytes.NewReader(data), cloneNS)
+				items, err := r.loadIndexesFrom(bytes.NewReader(data), cloneNS, util.MakeSelectedPred(nss))
 				if err != nil {
 					return nil, errors.Wrap(err, "load indexes")
 				}
+				reporter.SetTotalItems(int64(items))
 
 				rdr = io.NopCloser(bytes.NewReader(data))
 			}
@@ -1096,7 +1130,13 @@ func (r *Restore) RunSnapshot(
 		},
 		bcp.Compression,
 		util.MakeSelectedPred(nss),
-		r.numParallelColls)
+		r.numParallelColls,
+		func(ns string, bytes int64) {
+			reporter.AddBytes(bytes)
+			if ns != archive.MetaFile {
+				reporter.AddItems(1)
+			}
+		})
 	if err != nil {
 		return "", err
 	}
@@ -1238,20 +1278,25 @@ func (r *Restore) restoreUsersAndRoles(ctx context.Context, nss []string) error
 	return nil
 }
 
-func (r *Restore) loadIndexesFrom(rdr io.Reader, cloneNS snapshot.CloneNS) error {
+func (r *Restore) loadIndexesFrom(rdr io.Reader, cloneNS snapshot.CloneNS, selected archive.NSFilterFn) (int, error) {
 	meta, err := archive.ReadMetadata(rdr)
 	if err != nil {
-		return errors.Wrap(err, "read metadata")
+		return 0, errors.Wrap(err, "read metadata")
 	}
 
 	fromDB, fromColl := cloneNS.SplitFromNS()
 	toDB, toColl := cloneNS.SplitToNS()
 
+	items := 0
 	for _, ns := range meta.Namespaces {
+		if selected(archive.NSify(ns.Database, ns.Collection)) {
+			items++
+		}
+
 		var md mongorestore.Metadata
 		err := bson.UnmarshalExtJSON([]byte(ns.Metadata), true, &md)
 		if err != nil {
-			return errors.Wrapf(err, "unmarshal %s.%s metadata",
+			return 0, errors.Wrapf(err, "unmarshal %s.%s metadata",
 				ns.Database, ns.Collection)
 		}
 
@@ -1283,7 +1328,7 @@ func (r *Restore) loadIndexesFrom(rdr io.Reader, cloneNS snapshot.CloneNS) error
 		}
 	}
 
-	return nil
+	return items, nil
 }
 
 func (r *Restore) restoreIndexes(ctx context.Context, nss []string) error {
diff --git a/pbm/restore/physical.go b/pbm/restore/physical.go
index 17e136712..ece19946b 100644
--- a/pbm/restore/physical.go
+++ b/pbm/restore/physical.go
@@ -38,6 +38,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
+	progresspkg "github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/restore/phys"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
@@ -1179,6 +1180,7 @@ func (r *PhysRestore) Snapshot(
 	pauseHB func(),
 ) (err error) {
 	l.Debug("port: %d", r.tmpPort)
+	opStarted := time.Now()
 
 	meta := &RestoreMeta{
 		Type:     defs.PhysicalBackup,
@@ -1195,6 +1197,13 @@ func (r *PhysRestore) Snapshot(
 
 	var progress nodeStatus
 	defer func() {
+		elapsed := progresspkg.FormatDuration(time.Since(opStarted))
+		if err != nil {
+			l.Info("restore failed after %s: %v", elapsed, err)
+		} else {
+			l.Info("restore completed after %s", elapsed)
+		}
+
 		if cmd.Exit && err == nil {
 			// nothing to cleanup in case of successful ext restore with exit
 			return
@@ -1683,6 +1692,11 @@ func (r *PhysRestore) copyFiles() (*storage.DownloadStat, error) {
 
 	setName := util.MakeReverseRSMapFunc(r.rsMap)(r.nodeInfo.SetName)
 	jobs := r.planCopyFiles(setName)
+	reporter := progresspkg.NewReporter(context.Background(), r.log, time.Minute, plannedDownloadSize(jobs), 0,
+		func(ctx context.Context, p progresspkg.Progress) error {
+			return SetRestoreRSProgress(ctx, r.leadConn, r.name, r.nodeInfo.SetName, p)
+		})
+	defer reporter.Close("restore transfer finished")
 
 	numWorkers := r.GetNumParallelFiles()
 	if numWorkers > 1 {
@@ -1727,9 +1741,11 @@ func (r *PhysRestore) copyFiles() (*storage.DownloadStat, error) {
 				if err := egCtx.Err(); err != nil {
 					return err
 				}
-				if err := r.copyFile(op.src, job.dst, op.fMeta, op.cmpr, cpBuf); err != nil {
+				n, err := r.copyFile(op.src, job.dst, op.fMeta, op.cmpr, cpBuf)
+				if err != nil {
 					return err
 				}
+				reporter.AddBytes(n)
 			}
 			return nil
 		})
@@ -1738,34 +1754,58 @@ func (r *PhysRestore) copyFiles() (*storage.DownloadStat, error) {
 	if err := eg.Wait(); err != nil {
 		return stat, err
 	}
+	if err := reporter.Flush(); err != nil {
+		r.log.Warning("update progress: %v", err)
+	}
 	return stat, nil
 }
 
+func plannedDownloadSize(jobs []copyFileJob) int64 {
+	var size int64
+	for _, job := range jobs {
+		for _, op := range job.ops {
+			if op.fMeta.StgSize > 0 {
+				size += op.fMeta.StgSize
+			} else if op.fMeta.Len > 0 {
+				size += op.fMeta.Len
+			} else {
+				size += op.fMeta.Size
+			}
+		}
+	}
+	return size
+}
+
 // copyFile copies file from the storage into local FS.
-func (r *PhysRestore) copyFile(src, dst string, fMeta backup.File, cType compress.CompressionType, cpbuf []byte) error {
+func (r *PhysRestore) copyFile(src, dst string, fMeta backup.File, cType compress.CompressionType, cpbuf []byte) (int64, error) {
 	r.log.Info("copy <%s> to <%s>", src, dst)
+	stat, err := r.bcpStg.FileStat(src)
+	if err != nil {
+		return 0, errors.Wrapf(err, "stat source <%s>", src)
+	}
+
 	sr, err := r.bcpStg.SourceReader(src)
 	if err != nil {
-		return errors.Wrapf(err, "create source reader for <%s>", src)
+		return 0, errors.Wrapf(err, "create source reader for <%s>", src)
 	}
 	defer sr.Close()
 
 	data, err := compress.Decompress(sr, cType)
 	if err != nil {
-		return errors.Wrapf(err, "decompress object %s", src)
+		return 0, errors.Wrapf(err, "decompress object %s", src)
 	}
 	defer data.Close()
 
 	fw, err := os.OpenFile(dst, os.O_WRONLY|os.O_CREATE, fMeta.Fmode)
 	if err != nil {
-		return errors.Wrapf(err, "create/open destination file <%s>", dst)
+		return 0, errors.Wrapf(err, "create/open destination file <%s>", dst)
 	}
 	defer fw.Close()
 
 	if fMeta.Off != 0 {
 		_, err := fw.Seek(fMeta.Off, io.SeekStart)
 		if err != nil {
-			return errors.Wrapf(err, "set file offset <%s>|%d", dst, fMeta.Off)
+			return 0, errors.Wrapf(err, "set file offset <%s>|%d", dst, fMeta.Off)
 		}
 	}
 
@@ -1779,16 +1819,16 @@ func (r *PhysRestore) copyFile(src, dst string, fMeta backup.File, cType compres
 		)
 	}
 	if err != nil {
-		return errors.Wrapf(err, "copy file <%s>", dst)
+		return 0, errors.Wrapf(err, "copy file <%s>", dst)
 	}
 
 	if fMeta.Size != 0 {
 		err = fw.Truncate(fMeta.Size)
 		if err != nil {
-			return errors.Wrapf(err, "truncate file <%s>|%d", dst, fMeta.Size)
+			return 0, errors.Wrapf(err, "truncate file <%s>|%d", dst, fMeta.Size)
 		}
 	}
-	return nil
+	return stat.Size, nil
 }
 
 func (r *PhysRestore) getLasOpTime() (bson.Timestamp, error) {
diff --git a/pbm/restore/physical_test.go b/pbm/restore/physical_test.go
index 940358ea2..7a2a31785 100644
--- a/pbm/restore/physical_test.go
+++ b/pbm/restore/physical_test.go
@@ -644,7 +644,7 @@ func BenchmarkCopyGenFilesFromFSStorage(b *testing.B) {
 		dst := filepath.Join(dstDir, fmt.Sprintf("%s-%d", filepath.Base(f.Name()), rand.Uint64()))
 
 		ts := time.Now()
-		err := r.copyFile(filepath.Base(f.Name()), dst, backup.File{Fmode: 0o600, Size: size}, cType, cpbuf)
+		_, err := r.copyFile(filepath.Base(f.Name()), dst, backup.File{Fmode: 0o600, Size: size}, cType, cpbuf)
 		if err != nil {
 			b.Fatalf("copyFile: %v", err)
 		}
@@ -715,7 +715,7 @@ func BenchmarkCopyFileFromFSStorage(b *testing.B) {
 		dst := filepath.Join(*localPath, fmt.Sprintf("%s-%d", filepath.Base(*storageFile), rand.Uint64()))
 
 		ts := time.Now()
-		err := r.copyFile(*storageFile, dst, backup.File{Fmode: 0o600, Size: fSize}, cType, cpbuf)
+		_, err := r.copyFile(*storageFile, dst, backup.File{Fmode: 0o600, Size: fSize}, cType, cpbuf)
 		if err != nil {
 			b.Fatalf("copyFile: %v", err)
 		}
diff --git a/pbm/restore/query.go b/pbm/restore/query.go
index e378c8093..99f83f4bc 100644
--- a/pbm/restore/query.go
+++ b/pbm/restore/query.go
@@ -14,6 +14,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/connect"
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/restore/phys"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 )
@@ -49,15 +50,20 @@ func ChangeRestoreState(ctx context.Context, m connect.Client, name string, s de
 
 func changeRestoreState(ctx context.Context, m connect.Client, clause bson.D, s defs.Status, msg string) error {
 	ts := time.Now().UTC().Unix()
+	update := bson.D{
+		{"$set", bson.M{"status": s}},
+		{"$set", bson.M{"last_transition_ts": ts}},
+		{"$set", bson.M{"error": msg}},
+		{"$push", bson.M{"conditions": Condition{Timestamp: ts, Status: s, Error: msg}}},
+	}
+	if !s.IsRunning() {
+		update = append(update, bson.E{"$unset", bson.M{"progress": ""}})
+	}
+
 	_, err := m.RestoresCollection().UpdateOne(
 		ctx,
 		clause,
-		bson.D{
-			{"$set", bson.M{"status": s}},
-			{"$set", bson.M{"last_transition_ts": ts}},
-			{"$set", bson.M{"error": msg}},
-			{"$push", bson.M{"conditions": Condition{Timestamp: ts, Status: s, Error: msg}}},
-		},
+		update,
 	)
 
 	return err
@@ -72,17 +78,38 @@ func ChangeRestoreRSState(
 	msg string,
 ) error {
 	ts := time.Now().UTC().Unix()
+	update := bson.D{
+		{"$set", bson.M{"replsets.$.status": s}},
+		{"$set", bson.M{"replsets.$.last_transition_ts": ts}},
+		{"$set", bson.M{"replsets.$.error": msg}},
+		{"$push", bson.M{"replsets.$.conditions": Condition{Timestamp: ts, Status: s, Error: msg}}},
+	}
+	if !s.IsRunning() {
+		update = append(update, bson.E{"$unset", bson.M{"replsets.$.progress": ""}})
+	}
+
 	_, err := m.RestoresCollection().UpdateOne(
 		ctx,
 		bson.D{{"name", name}, {"replsets.name", rsName}},
-		bson.D{
-			{"$set", bson.M{"replsets.$.status": s}},
-			{"$set", bson.M{"replsets.$.last_transition_ts": ts}},
-			{"$set", bson.M{"replsets.$.error": msg}},
-			{"$push", bson.M{"replsets.$.conditions": Condition{Timestamp: ts, Status: s, Error: msg}}},
-		},
+		update,
+	)
+
+	return err
+}
+
+func SetRestoreProgress(ctx context.Context, m connect.Client, name string, p progress.Progress) error {
+	_, err := m.RestoresCollection().UpdateOne(ctx,
+		bson.D{{"name", name}},
+		bson.D{{"$set", bson.M{"progress": p}}},
 	)
+	return err
+}
 
+func SetRestoreRSProgress(ctx context.Context, m connect.Client, name, rsName string, p progress.Progress) error {
+	_, err := m.RestoresCollection().UpdateOne(ctx,
+		bson.D{{"name", name}, {"replsets.name", rsName}},
+		bson.D{{"$set", bson.M{"replsets.$.progress": p}}},
+	)
 	return err
 }
 
diff --git a/pbm/restore/types.go b/pbm/restore/types.go
index 65740141c..421c1cbb2 100644
--- a/pbm/restore/types.go
+++ b/pbm/restore/types.go
@@ -7,33 +7,36 @@ import (
 	"go.mongodb.org/mongo-driver/v2/bson"
 
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/restore/phys"
 )
 
 type RestoreMeta struct {
-	Status           defs.Status       `bson:"status" json:"status"`
-	Error            string            `bson:"error,omitempty" json:"error,omitempty"`
-	Name             string            `bson:"name" json:"name"`
-	OPID             string            `bson:"opid" json:"opid"`
-	Backup           string            `bson:"backup" json:"backup"`
-	BcpChain         []string          `bson:"bcp_chain" json:"bcp_chain"` // for incremental
-	Namespaces       []string          `bson:"nss,omitempty" json:"nss,omitempty"`
-	StartPITR        int64             `bson:"start_pitr" json:"start_pitr"`
-	PITR             int64             `bson:"pitr" json:"pitr"`
-	Replsets         []RestoreReplset  `bson:"replsets" json:"replsets"`
-	Hb               bson.Timestamp    `bson:"hb" json:"hb"`
-	StartTS          int64             `bson:"start_ts" json:"start_ts"`
-	LastTransitionTS int64             `bson:"last_transition_ts" json:"last_transition_ts"`
-	Conditions       Conditions        `bson:"conditions" json:"conditions"`
-	Type             defs.BackupType   `bson:"type" json:"type"`
-	Leader           string            `bson:"l,omitempty" json:"l,omitempty"`
-	Stat             *phys.RestoreStat `bson:"stat,omitempty" json:"stat,omitempty"`
+	Status           defs.Status        `bson:"status" json:"status"`
+	Progress         *progress.Progress `bson:"progress,omitempty" json:"progress,omitempty"`
+	Error            string             `bson:"error,omitempty" json:"error,omitempty"`
+	Name             string             `bson:"name" json:"name"`
+	OPID             string             `bson:"opid" json:"opid"`
+	Backup           string             `bson:"backup" json:"backup"`
+	BcpChain         []string           `bson:"bcp_chain" json:"bcp_chain"` // for incremental
+	Namespaces       []string           `bson:"nss,omitempty" json:"nss,omitempty"`
+	StartPITR        int64              `bson:"start_pitr" json:"start_pitr"`
+	PITR             int64              `bson:"pitr" json:"pitr"`
+	Replsets         []RestoreReplset   `bson:"replsets" json:"replsets"`
+	Hb               bson.Timestamp     `bson:"hb" json:"hb"`
+	StartTS          int64              `bson:"start_ts" json:"start_ts"`
+	LastTransitionTS int64              `bson:"last_transition_ts" json:"last_transition_ts"`
+	Conditions       Conditions         `bson:"conditions" json:"conditions"`
+	Type             defs.BackupType    `bson:"type" json:"type"`
+	Leader           string             `bson:"l,omitempty" json:"l,omitempty"`
+	Stat             *phys.RestoreStat  `bson:"stat,omitempty" json:"stat,omitempty"`
 }
 
 type RestoreReplset struct {
 	Name             string                `bson:"name" json:"name"`
 	StartTS          int64                 `bson:"start_ts" json:"start_ts"`
 	Status           defs.Status           `bson:"status" json:"status"`
+	Progress         *progress.Progress    `bson:"progress,omitempty" json:"progress,omitempty"`
 	CommittedTxn     []phys.RestoreTxn     `bson:"committed_txn" json:"committed_txn"`
 	CommittedTxnSet  bool                  `bson:"txn_set" json:"txn_set"`
 	PartialTxn       []db.Oplog            `bson:"partial_txn" json:"partial_txn"`
diff --git a/pbm/snapshot/dump.go b/pbm/snapshot/dump.go
index 246457fb1..70eb35a98 100644
--- a/pbm/snapshot/dump.go
+++ b/pbm/snapshot/dump.go
@@ -11,6 +11,7 @@ import (
 )
 
 type UploadFunc func(ns, ext string, r io.Reader) error
+type ProgressFunc func(ns string, bytes int64)
 
 func UploadDump(
 	ctx context.Context,
@@ -18,6 +19,17 @@ func UploadDump(
 	upload UploadFunc,
 	compression compress.CompressionType,
 	compressionLevel *int,
+) (int64, error) {
+	return UploadDumpWithProgress(ctx, dump, upload, compression, compressionLevel, nil)
+}
+
+func UploadDumpWithProgress(
+	ctx context.Context,
+	dump func(archive.NewWriter) error,
+	upload UploadFunc,
+	compression compress.CompressionType,
+	compressionLevel *int,
+	progress ProgressFunc,
 ) (int64, error) {
 	uploadSize := int64(0)
 
@@ -42,6 +54,9 @@ func UploadDump(
 			}
 
 			atomic.AddInt64(&uploadSize, rc.n)
+			if progress != nil {
+				progress(ns, rc.n)
+			}
 		}()
 
 		w, err := compress.Compress(pw, compression, compressionLevel)
@@ -65,6 +80,16 @@ func DownloadDump(
 	compression compress.CompressionType,
 	match archive.NSFilterFn,
 	numParallelColls int,
+) (io.ReadCloser, error) {
+	return DownloadDumpWithProgress(download, compression, match, numParallelColls, nil)
+}
+
+func DownloadDumpWithProgress(
+	download DownloadFunc,
+	compression compress.CompressionType,
+	match archive.NSFilterFn,
+	numParallelColls int,
+	progress ProgressFunc,
 ) (io.ReadCloser, error) {
 	pr, pw := io.Pipe()
 
@@ -78,6 +103,9 @@ func DownloadDump(
 			if err != nil {
 				return nil, errors.Wrapf(err, "download: %q", ns)
 			}
+			if progress != nil {
+				r = &readCounterCloser{ReadCloser: r, ns: ns, progress: progress}
+			}
 
 			if ns == archive.MetaFile {
 				return r, nil
@@ -94,6 +122,25 @@ func DownloadDump(
 	return pr, nil
 }
 
+type readCounterCloser struct {
+	io.ReadCloser
+	ns       string
+	n        int64
+	progress ProgressFunc
+}
+
+func (c *readCounterCloser) Read(p []byte) (int, error) {
+	n, err := c.ReadCloser.Read(p)
+	c.n += int64(n)
+	return n, err
+}
+
+func (c *readCounterCloser) Close() error {
+	err := c.ReadCloser.Close()
+	c.progress(c.ns, c.n)
+	return err
+}
+
 type readCounter struct {
 	r io.Reader
 	n int64

From b73a7ca1cd711ac302b2adf5f3b222f82045367e Mon Sep 17 00:00:00 2001
From: igroene <9805809+igroene@users.noreply.github.com>
Date: Mon, 6 Jul 2026 08:21:54 -0300
Subject: [PATCH 2/9] add eta and improve throughput

---
 cmd/pbm/backup.go        | 13 ++++++++++
 cmd/pbm/main.go          | 18 +++++++++++++
 cmd/pbm/restore.go       |  7 ++++-
 pbm/progress/progress.go | 55 ++++++++++++++++++++++++++++++++--------
 4 files changed, 82 insertions(+), 11 deletions(-)

diff --git a/cmd/pbm/backup.go b/cmd/pbm/backup.go
index 6c83bd431..0e6051fe4 100644
--- a/cmd/pbm/backup.go
+++ b/cmd/pbm/backup.go
@@ -20,6 +20,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
 	"github.com/percona/percona-backup-mongodb/pbm/util"
@@ -388,6 +389,11 @@ type bcpDesc struct {
 	StorageType        storage.Type    `json:"storage_type,omitempty" yaml:"storage_type,omitempty"`
 	OPID               string          `json:"opid" yaml:"opid"`
 	Type               defs.BackupType `json:"type" yaml:"type"`
+	StartTS            int64           `json:"start_ts" yaml:"-"`
+	StartTime          string          `json:"start" yaml:"start"`
+	FinishTime         *string         `json:"finish,omitempty" yaml:"finish,omitempty"`
+	Duration           int64           `json:"duration" yaml:"-"`
+	DurationH          string          `json:"duration_h" yaml:"duration"`
 	LastWriteTS        int64           `json:"last_write_ts" yaml:"-"`
 	LastTransitionTS   int64           `json:"last_transition_ts" yaml:"-"`
 	LastWriteTime      string          `json:"last_write_time" yaml:"last_write_time"`
@@ -484,6 +490,8 @@ func describeBackup(
 		StorageType:        bcp.Store.Type,
 		OPID:               bcp.OPID,
 		Type:               bcp.Type,
+		StartTS:            bcp.StartTS,
+		StartTime:          time.Unix(bcp.StartTS, 0).UTC().Format(time.RFC3339),
 		Namespaces:         bcp.Namespaces,
 		SelUserAndRoles:    bcp.SelUsersAndRoles,
 		MongoVersion:       bcp.MongoVersion,
@@ -499,6 +507,11 @@ func describeBackup(
 		SizeUncompressed:   bcp.SizeUncompressed,
 		HSizeUncompressed:  byteCountIEC(bcp.SizeUncompressed),
 	}
+	rv.Duration = operationDurationSeconds(bcp.Status, bcp.StartTS, bcp.LastTransitionTS)
+	rv.DurationH = progress.FormatDuration(time.Duration(rv.Duration) * time.Second)
+	if isTerminalStatus(bcp.Status) {
+		rv.FinishTime = util.Ref(time.Unix(bcp.LastTransitionTS, 0).UTC().Format(time.RFC3339))
+	}
 	if bcp.SizeUncompressed > 0 {
 		rv.HSizeUncompressed = byteCountIEC(bcp.SizeUncompressed)
 	}
diff --git a/cmd/pbm/main.go b/cmd/pbm/main.go
index 02b69428d..2f44c5f05 100644
--- a/cmd/pbm/main.go
+++ b/cmd/pbm/main.go
@@ -1225,6 +1225,24 @@ func fmtTS(ts int64) string {
 	return strings.TrimSuffix(t, "Z")
 }
 
+func isTerminalStatus(status defs.Status) bool {
+	return !status.IsRunning() || status == defs.StatusPartlyDone
+}
+
+func operationDurationSeconds(status defs.Status, start, transition int64) int64 {
+	if start <= 0 {
+		return 0
+	}
+	end := transition
+	if !isTerminalStatus(status) {
+		end = time.Now().Unix()
+	}
+	if end < start {
+		return 0
+	}
+	return end - start
+}
+
 type outMsg struct {
 	Msg string `json:"msg"`
 }
diff --git a/cmd/pbm/restore.go b/cmd/pbm/restore.go
index 079a7bcaf..ccfc7694d 100644
--- a/cmd/pbm/restore.go
+++ b/cmd/pbm/restore.go
@@ -21,6 +21,7 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/defs"
 	"github.com/percona/percona-backup-mongodb/pbm/errors"
 	"github.com/percona/percona-backup-mongodb/pbm/log"
+	"github.com/percona/percona-backup-mongodb/pbm/progress"
 	"github.com/percona/percona-backup-mongodb/pbm/restore"
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 	"github.com/percona/percona-backup-mongodb/pbm/topo"
@@ -728,6 +729,8 @@ type describeRestoreResult struct {
 	StartTS            *int64           `json:"start_ts,omitempty" yaml:"-"`
 	StartTime          *string          `json:"start,omitempty" yaml:"start,omitempty"`
 	FinishTime         *string          `json:"finish,omitempty" yaml:"finish,omitempty"`
+	Duration           int64            `json:"duration,omitempty" yaml:"-"`
+	DurationH          string           `json:"duration_h,omitempty" yaml:"duration,omitempty"`
 	PITR               *int64           `json:"ts_to_restore,omitempty" yaml:"-"`
 	PITRTime           *string          `json:"time_to_restore,omitempty" yaml:"time_to_restore,omitempty"`
 	LastTransitionTS   int64            `json:"last_transition_ts" yaml:"-"`
@@ -804,7 +807,9 @@ func describeRestore(
 	res.LastTransitionTS = meta.LastTransitionTS
 	res.LastTransitionTime = time.Unix(res.LastTransitionTS, 0).UTC().Format(time.RFC3339)
 	res.StartTime = util.Ref(time.Unix(meta.StartTS, 0).UTC().Format(time.RFC3339))
-	if meta.Status == defs.StatusDone {
+	res.Duration = operationDurationSeconds(meta.Status, meta.StartTS, meta.LastTransitionTS)
+	res.DurationH = progress.FormatDuration(time.Duration(res.Duration) * time.Second)
+	if isTerminalStatus(meta.Status) {
 		res.FinishTime = util.Ref(time.Unix(meta.LastTransitionTS, 0).UTC().Format(time.RFC3339))
 	}
 	if meta.Status == defs.StatusError {
diff --git a/pbm/progress/progress.go b/pbm/progress/progress.go
index e47873c6f..ed50d6499 100644
--- a/pbm/progress/progress.go
+++ b/pbm/progress/progress.go
@@ -21,6 +21,8 @@ type Progress struct {
 	TotalBytes int64 `bson:"total_bytes,omitempty" json:"total_bytes,omitempty"`
 	DoneItems  int64 `bson:"done_items,omitempty" json:"done_items,omitempty"`
 	TotalItems int64 `bson:"total_items,omitempty" json:"total_items,omitempty"`
+	// ThroughputBytesPerSecond is the last known completed transfer rate.
+	ThroughputBytesPerSecond int64 `bson:"throughput_bps,omitempty" json:"throughput_bps,omitempty"`
 }
 
 func New(totalBytes, totalItems int64) Progress {
@@ -40,11 +42,20 @@ func (p Progress) Percent() (float64, bool) {
 }
 
 func (p Progress) ThroughputMBps(now int64) float64 {
-	if p.DoneBytes <= 0 || p.StartedAt <= 0 || now <= p.StartedAt {
-		return 0
+	_ = now
+	if p.ThroughputBytesPerSecond > 0 {
+		return float64(p.ThroughputBytesPerSecond) / float64(mb)
 	}
 
-	return float64(p.DoneBytes) / float64(mb) / float64(now-p.StartedAt)
+	return 0
+}
+
+func (p Progress) ETA() (time.Duration, bool) {
+	if p.TotalBytes <= 0 || p.DoneBytes <= 0 || p.DoneBytes >= p.TotalBytes || p.ThroughputBytesPerSecond <= 0 {
+		return 0, false
+	}
+
+	return time.Duration((p.TotalBytes-p.DoneBytes)/p.ThroughputBytesPerSecond) * time.Second, true
 }
 
 func (p Progress) Elapsed(now int64) time.Duration {
@@ -72,6 +83,9 @@ func (p Progress) StringAt(now int64) string {
 	if mbps := p.ThroughputMBps(now); mbps > 0 {
 		parts = append(parts, fmt.Sprintf("throughput=%.2fMB/s", mbps))
 	}
+	if eta, ok := p.ETA(); ok {
+		parts = append(parts, "eta="+FormatDuration(eta))
+	}
 
 	return strings.Join(parts, ", ")
 }
@@ -122,6 +136,9 @@ type Reporter struct {
 	doneB     atomic.Int64
 	totalI    atomic.Int64
 	doneI     atomic.Int64
+	lastTick  atomic.Int64
+	lastBytes atomic.Int64
+	lastBPS   atomic.Int64
 	stop      chan struct{}
 }
 
@@ -136,6 +153,7 @@ func NewReporter(
 	r := &Reporter{ctx: ctx, log: log, update: update, startedAt: time.Now().Unix(), stop: make(chan struct{})}
 	r.totalB.Store(totalBytes)
 	r.totalI.Store(totalItems)
+	r.lastTick.Store(r.startedAt)
 	_ = r.Flush()
 
 	go func() {
@@ -148,7 +166,7 @@ func NewReporter(
 			case <-r.stop:
 				return
 			case <-tk.C:
-				p := r.Snapshot()
+				p := r.TickSnapshot()
 				if err := r.FlushProgress(p); err != nil && log != nil {
 					log.Warning("update progress: %v", err)
 				}
@@ -179,15 +197,32 @@ func (r *Reporter) SetTotalItems(n int64) { r.totalI.Store(n) }
 
 func (r *Reporter) Snapshot() Progress {
 	return Progress{
-		StartedAt:  r.startedAt,
-		UpdatedAt:  time.Now().Unix(),
-		DoneBytes:  r.doneB.Load(),
-		TotalBytes: r.totalB.Load(),
-		DoneItems:  r.doneI.Load(),
-		TotalItems: r.totalI.Load(),
+		StartedAt:                r.startedAt,
+		UpdatedAt:                time.Now().Unix(),
+		DoneBytes:                r.doneB.Load(),
+		TotalBytes:               r.totalB.Load(),
+		DoneItems:                r.doneI.Load(),
+		TotalItems:               r.totalI.Load(),
+		ThroughputBytesPerSecond: r.lastBPS.Load(),
 	}
 }
 
+func (r *Reporter) TickSnapshot() Progress {
+	now := time.Now().Unix()
+	done := r.doneB.Load()
+	lastTick := r.lastTick.Load()
+	lastBytes := r.lastBytes.Load()
+	if deltaSeconds := now - lastTick; deltaSeconds > 0 && done > lastBytes {
+		r.lastBPS.Store((done - lastBytes) / deltaSeconds)
+		r.lastTick.Store(now)
+		r.lastBytes.Store(done)
+	}
+
+	p := r.Snapshot()
+	p.UpdatedAt = now
+	return p
+}
+
 func (r *Reporter) Flush() error {
 	return r.FlushProgress(r.Snapshot())
 }

From 42a22de54550047d3c4f7ccd96bad2f77e499bde Mon Sep 17 00:00:00 2001
From: igroene <9805809+igroene@users.noreply.github.com>
Date: Mon, 6 Jul 2026 08:45:24 -0300
Subject: [PATCH 3/9] fix eta for logical bkp

---
 pbm/backup/logical.go    | 4 +++-
 pbm/progress/progress.go | 4 ++--
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/pbm/backup/logical.go b/pbm/backup/logical.go
index 640185ff4..4720b6e08 100644
--- a/pbm/backup/logical.go
+++ b/pbm/backup/logical.go
@@ -48,6 +48,7 @@ func (b *Backup) doLogical(
 	}
 
 	sizeHints := make(map[string]int64, len(nssSize))
+	totalSizeHint := int64(0)
 	for ns, cs := range nssSize {
 		if bcp.Compression == compress.CompressionTypeNone {
 			// Uncompressed dump: the output size matches the logical BSON size.
@@ -56,6 +57,7 @@ func (b *Backup) doLogical(
 			// Compressed: WiredTiger on-disk size approximates compressed output.
 			sizeHints[ns] = cs.StorageSize
 		}
+		totalSizeHint += sizeHints[ns]
 	}
 
 	rsMeta.Status = defs.StatusRunning
@@ -65,7 +67,7 @@ func (b *Backup) doLogical(
 	if err != nil {
 		return errors.Wrap(err, "add shard's metadata")
 	}
-	reporter := progress.NewReporter(ctx, l, time.Minute, 0, int64(len(nssSize)),
+	reporter := progress.NewReporter(ctx, l, time.Minute, totalSizeHint, int64(len(nssSize)),
 		func(ctx context.Context, p progress.Progress) error {
 			return SetRSProgress(ctx, b.leadConn, bcp.Name, rsMeta.Name, p)
 		})
diff --git a/pbm/progress/progress.go b/pbm/progress/progress.go
index ed50d6499..b84401616 100644
--- a/pbm/progress/progress.go
+++ b/pbm/progress/progress.go
@@ -32,10 +32,10 @@ func New(totalBytes, totalItems int64) Progress {
 
 func (p Progress) Percent() (float64, bool) {
 	switch {
-	case p.TotalBytes > 0:
-		return percent(p.DoneBytes, p.TotalBytes), true
 	case p.TotalItems > 0:
 		return percent(p.DoneItems, p.TotalItems), true
+	case p.TotalBytes > 0:
+		return percent(p.DoneBytes, p.TotalBytes), true
 	default:
 		return 0, false
 	}

From 70184056379c3d4b6b8afeab58f929bb257045d6 Mon Sep 17 00:00:00 2001
From: igroene <9805809+igroene@users.noreply.github.com>
Date: Mon, 6 Jul 2026 09:22:22 -0300
Subject: [PATCH 4/9] fix eta logical

---
 pbm/backup/logical.go    |  4 +--
 pbm/progress/progress.go |  4 +--
 pbm/restore/logical.go   |  4 +--
 pbm/snapshot/dump.go     | 62 +++++++++++++++++++++++++++++++++-------
 4 files changed, 58 insertions(+), 16 deletions(-)

diff --git a/pbm/backup/logical.go b/pbm/backup/logical.go
index 4720b6e08..fa7bdb477 100644
--- a/pbm/backup/logical.go
+++ b/pbm/backup/logical.go
@@ -211,9 +211,9 @@ func (b *Backup) doLogical(
 		},
 		bcp.Compression,
 		bcp.CompressionLevel,
-		func(ns string, bytes int64) {
+		func(ns string, bytes int64, done bool) {
 			reporter.AddBytes(bytes)
-			if ns != archive.MetaFileV2 {
+			if done && ns != archive.MetaFileV2 {
 				reporter.AddItems(1)
 			}
 		})
diff --git a/pbm/progress/progress.go b/pbm/progress/progress.go
index b84401616..ed50d6499 100644
--- a/pbm/progress/progress.go
+++ b/pbm/progress/progress.go
@@ -32,10 +32,10 @@ func New(totalBytes, totalItems int64) Progress {
 
 func (p Progress) Percent() (float64, bool) {
 	switch {
-	case p.TotalItems > 0:
-		return percent(p.DoneItems, p.TotalItems), true
 	case p.TotalBytes > 0:
 		return percent(p.DoneBytes, p.TotalBytes), true
+	case p.TotalItems > 0:
+		return percent(p.DoneItems, p.TotalItems), true
 	default:
 		return 0, false
 	}
diff --git a/pbm/restore/logical.go b/pbm/restore/logical.go
index 015e47018..b452a5ee5 100644
--- a/pbm/restore/logical.go
+++ b/pbm/restore/logical.go
@@ -1131,9 +1131,9 @@ func (r *Restore) RunSnapshot(
 		bcp.Compression,
 		util.MakeSelectedPred(nss),
 		r.numParallelColls,
-		func(ns string, bytes int64) {
+		func(ns string, bytes int64, done bool) {
 			reporter.AddBytes(bytes)
-			if ns != archive.MetaFile {
+			if done && ns != archive.MetaFile {
 				reporter.AddItems(1)
 			}
 		})
diff --git a/pbm/snapshot/dump.go b/pbm/snapshot/dump.go
index 70eb35a98..544c37e22 100644
--- a/pbm/snapshot/dump.go
+++ b/pbm/snapshot/dump.go
@@ -11,7 +11,9 @@ import (
 )
 
 type UploadFunc func(ns, ext string, r io.Reader) error
-type ProgressFunc func(ns string, bytes int64)
+type ProgressFunc func(ns string, bytes int64, done bool)
+
+const progressThresholdBytes = 16 << 20
 
 func UploadDump(
 	ctx context.Context,
@@ -45,7 +47,7 @@ func UploadDumpWithProgress(
 		go func() {
 			defer close(done)
 
-			rc := &readCounter{r: pr}
+			rc := &readCounter{r: pr, ns: ns, progress: progress}
 			err := upload(ns, compression.Suffix(), rc)
 			if err != nil {
 				err = errors.Wrapf(err, "upload: %q", ns)
@@ -53,10 +55,8 @@ func UploadDumpWithProgress(
 				done <- err
 			}
 
+			rc.finish()
 			atomic.AddInt64(&uploadSize, rc.n)
-			if progress != nil {
-				progress(ns, rc.n)
-			}
 		}()
 
 		w, err := compress.Compress(pw, compression, compressionLevel)
@@ -126,32 +126,74 @@ type readCounterCloser struct {
 	io.ReadCloser
 	ns       string
 	n        int64
+	pending  int64
 	progress ProgressFunc
 }
 
 func (c *readCounterCloser) Read(p []byte) (int, error) {
 	n, err := c.ReadCloser.Read(p)
-	c.n += int64(n)
+	c.add(int64(n))
 	return n, err
 }
 
 func (c *readCounterCloser) Close() error {
 	err := c.ReadCloser.Close()
-	c.progress(c.ns, c.n)
+	c.finish()
 	return err
 }
 
+func (c *readCounterCloser) add(n int64) {
+	if n <= 0 {
+		return
+	}
+	c.n += n
+	c.pending += n
+	if c.progress != nil && c.pending >= progressThresholdBytes {
+		c.progress(c.ns, c.pending, false)
+		c.pending = 0
+	}
+}
+
+func (c *readCounterCloser) finish() {
+	if c.progress != nil {
+		c.progress(c.ns, c.pending, true)
+	}
+	c.pending = 0
+}
+
 type readCounter struct {
-	r io.Reader
-	n int64
+	r        io.Reader
+	n        int64
+	ns       string
+	pending  int64
+	progress ProgressFunc
 }
 
 func (c *readCounter) Read(p []byte) (int, error) {
 	n, err := c.r.Read(p)
-	c.n += int64(n)
+	c.add(int64(n))
 	return n, err
 }
 
+func (c *readCounter) add(n int64) {
+	if n <= 0 {
+		return
+	}
+	c.n += n
+	c.pending += n
+	if c.progress != nil && c.pending >= progressThresholdBytes {
+		c.progress(c.ns, c.pending, false)
+		c.pending = 0
+	}
+}
+
+func (c *readCounter) finish() {
+	if c.progress != nil {
+		c.progress(c.ns, c.pending, true)
+	}
+	c.pending = 0
+}
+
 type funcCloser func() error
 
 func (f funcCloser) Close() error {

From 881c961c281ab9ab454ac3fe0ef70c79724c2b83 Mon Sep 17 00:00:00 2001
From: igroene <9805809+igroene@users.noreply.github.com>
Date: Mon, 6 Jul 2026 10:19:05 -0300
Subject: [PATCH 5/9] fix logical restore eta

---
 pbm/restore/logical.go | 14 +++++++++++++-
 1 file changed, 13 insertions(+), 1 deletion(-)

diff --git a/pbm/restore/logical.go b/pbm/restore/logical.go
index b452a5ee5..2e7cd3cf7 100644
--- a/pbm/restore/logical.go
+++ b/pbm/restore/logical.go
@@ -1089,9 +1089,10 @@ func (r *Restore) RunSnapshot(
 	}
 
 	mapRS := util.MakeReverseRSMapFunc(r.rsMap)
+	restoreTotalBytes := logicalRestoreTotalBytes(bcp, mapRS(r.brief.SetName))
 
 	r.log.Debug("restoring up to %d collections in parallel", r.numParallelColls)
-	reporter := progress.NewReporter(ctx, r.log, time.Minute, 0, 0,
+	reporter := progress.NewReporter(ctx, r.log, time.Minute, restoreTotalBytes, 0,
 		func(ctx context.Context, p progress.Progress) error {
 			return SetRestoreRSProgress(ctx, r.leadConn, r.name, r.nodeInfo.SetName, p)
 		})
@@ -1178,6 +1179,17 @@ func (r *Restore) RunSnapshot(
 	return sysSessionsUUID, nil
 }
 
+func logicalRestoreTotalBytes(bcp *backup.BackupMeta, rsName string) int64 {
+	for i := range bcp.Replsets {
+		rs := &bcp.Replsets[i]
+		if rs.Name == rsName && rs.Size > 0 {
+			return rs.Size
+		}
+	}
+
+	return bcp.Size
+}
+
 func (r *Restore) restoreLegacyArchive(
 	ctx context.Context,
 	dump string,

From b80775d5cdc7b103b50e153bdba4481ad3532fc4 Mon Sep 17 00:00:00 2001
From: igroene <9805809+igroene@users.noreply.github.com>
Date: Mon, 6 Jul 2026 11:14:28 -0300
Subject: [PATCH 6/9] fix physical bkp/restore eta

---
 pbm/progress/progress.go     | 53 +++++++++++++++++++++++++++++++++++-
 pbm/restore/physical.go      | 13 +++++++--
 pbm/restore/physical_test.go |  4 +--
 3 files changed, 64 insertions(+), 6 deletions(-)

diff --git a/pbm/progress/progress.go b/pbm/progress/progress.go
index ed50d6499..c5dfab11d 100644
--- a/pbm/progress/progress.go
+++ b/pbm/progress/progress.go
@@ -3,6 +3,7 @@ package progress
 import (
 	"context"
 	"fmt"
+	"io"
 	"strings"
 	"sync/atomic"
 	"time"
@@ -10,7 +11,10 @@ import (
 	"github.com/percona/percona-backup-mongodb/pbm/storage"
 )
 
-const mb = 1024 * 1024
+const (
+	mb                            = 1024 * 1024
+	DefaultProgressThresholdBytes = 16 << 20
+)
 
 // Progress captures an operation progress snapshot. Byte counters represent
 // bytes transferred to/from backup storage.
@@ -241,3 +245,50 @@ func (r *Reporter) Close(final string) {
 		r.log.Info("%s after %s", final, FormatDuration(p.Elapsed(time.Now().Unix())))
 	}
 }
+
+type CountingReadCloser struct {
+	io.ReadCloser
+	reporter  *Reporter
+	threshold int64
+	pending   int64
+}
+
+func NewCountingReadCloser(r io.ReadCloser, reporter *Reporter, threshold int64) io.ReadCloser {
+	if reporter == nil {
+		return r
+	}
+	if threshold <= 0 {
+		threshold = DefaultProgressThresholdBytes
+	}
+	return &CountingReadCloser{ReadCloser: r, reporter: reporter, threshold: threshold}
+}
+
+func (r *CountingReadCloser) Read(p []byte) (int, error) {
+	n, err := r.ReadCloser.Read(p)
+	r.add(int64(n))
+	return n, err
+}
+
+func (r *CountingReadCloser) Close() error {
+	err := r.ReadCloser.Close()
+	r.flush()
+	return err
+}
+
+func (r *CountingReadCloser) add(n int64) {
+	if n <= 0 {
+		return
+	}
+	r.pending += n
+	if r.pending >= r.threshold {
+		r.flush()
+	}
+}
+
+func (r *CountingReadCloser) flush() {
+	if r.pending <= 0 {
+		return
+	}
+	r.reporter.AddBytes(r.pending)
+	r.pending = 0
+}
diff --git a/pbm/restore/physical.go b/pbm/restore/physical.go
index ece19946b..5eee38870 100644
--- a/pbm/restore/physical.go
+++ b/pbm/restore/physical.go
@@ -1741,11 +1741,10 @@ func (r *PhysRestore) copyFiles() (*storage.DownloadStat, error) {
 				if err := egCtx.Err(); err != nil {
 					return err
 				}
-				n, err := r.copyFile(op.src, job.dst, op.fMeta, op.cmpr, cpBuf)
+				_, err := r.copyFile(op.src, job.dst, op.fMeta, op.cmpr, cpBuf, reporter)
 				if err != nil {
 					return err
 				}
-				reporter.AddBytes(n)
 			}
 			return nil
 		})
@@ -1777,7 +1776,14 @@ func plannedDownloadSize(jobs []copyFileJob) int64 {
 }
 
 // copyFile copies file from the storage into local FS.
-func (r *PhysRestore) copyFile(src, dst string, fMeta backup.File, cType compress.CompressionType, cpbuf []byte) (int64, error) {
+func (r *PhysRestore) copyFile(
+	src,
+	dst string,
+	fMeta backup.File,
+	cType compress.CompressionType,
+	cpbuf []byte,
+	reporter *progresspkg.Reporter,
+) (int64, error) {
 	r.log.Info("copy <%s> to <%s>", src, dst)
 	stat, err := r.bcpStg.FileStat(src)
 	if err != nil {
@@ -1788,6 +1794,7 @@ func (r *PhysRestore) copyFile(src, dst string, fMeta backup.File, cType compres
 	if err != nil {
 		return 0, errors.Wrapf(err, "create source reader for <%s>", src)
 	}
+	sr = progresspkg.NewCountingReadCloser(sr, reporter, progresspkg.DefaultProgressThresholdBytes)
 	defer sr.Close()
 
 	data, err := compress.Decompress(sr, cType)
diff --git a/pbm/restore/physical_test.go b/pbm/restore/physical_test.go
index 7a2a31785..b9a0356c2 100644
--- a/pbm/restore/physical_test.go
+++ b/pbm/restore/physical_test.go
@@ -644,7 +644,7 @@ func BenchmarkCopyGenFilesFromFSStorage(b *testing.B) {
 		dst := filepath.Join(dstDir, fmt.Sprintf("%s-%d", filepath.Base(f.Name()), rand.Uint64()))
 
 		ts := time.Now()
-		_, err := r.copyFile(filepath.Base(f.Name()), dst, backup.File{Fmode: 0o600, Size: size}, cType, cpbuf)
+		_, err := r.copyFile(filepath.Base(f.Name()), dst, backup.File{Fmode: 0o600, Size: size}, cType, cpbuf, nil)
 		if err != nil {
 			b.Fatalf("copyFile: %v", err)
 		}
@@ -715,7 +715,7 @@ func BenchmarkCopyFileFromFSStorage(b *testing.B) {
 		dst := filepath.Join(*localPath, fmt.Sprintf("%s-%d", filepath.Base(*storageFile), rand.Uint64()))
 
 		ts := time.Now()
-		_, err := r.copyFile(*storageFile, dst, backup.File{Fmode: 0o600, Size: fSize}, cType, cpbuf)
+		_, err := r.copyFile(*storageFile, dst, backup.File{Fmode: 0o600, Size: fSize}, cType, cpbuf, nil)
 		if err != nil {
 			b.Fatalf("copyFile: %v", err)
 		}

From bde247a0245ec77a7a110114bb7f71e2588b10ec Mon Sep 17 00:00:00 2001
From: igroene <9805809+igroene@users.noreply.github.com>
Date: Mon, 6 Jul 2026 11:35:18 -0300
Subject: [PATCH 7/9] fix logical bkp/restore eta

---
 pbm/backup/physical.go      | 142 +++++++++++++++++++++++++++++++++---
 pbm/backup/physical_test.go |   4 +-
 pbm/progress/progress.go    |  15 +++-
 3 files changed, 148 insertions(+), 13 deletions(-)

diff --git a/pbm/backup/physical.go b/pbm/backup/physical.go
index eedfed28f..3c83bdac2 100644
--- a/pbm/backup/physical.go
+++ b/pbm/backup/physical.go
@@ -5,10 +5,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"io"
 	"os"
 	"path"
 	"path/filepath"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/google/uuid"
@@ -382,14 +384,15 @@ func (b *Backup) doPhysical(
 		return b.handleExternal(ctx, bcp, rsMeta, data, jrnls, bcur.Meta.DBpath, opid, inf, stg, l)
 	}
 
-	reporter := progress.NewReporter(ctx, l, time.Minute,
-		plannedUploadSize(data, b.typ == defs.IncrementalBackup)+plannedUploadSize(jrnls, false), 0,
+	totalSourceSize := plannedUploadSize(data, b.typ == defs.IncrementalBackup) + plannedUploadSize(jrnls, false)
+	reporter := progress.NewReporter(ctx, l, time.Minute, totalSourceSize, 0,
 		func(ctx context.Context, p progress.Progress) error {
 			return SetRSProgress(ctx, b.leadConn, bcp.Name, rsMeta.Name, p)
 		})
+	progressTracker := newPhysicalUploadProgress(reporter, totalSourceSize)
 	defer reporter.Close("backup transfer finished")
 
-	return b.uploadPhysical(ctx, bcp, rsMeta, data, jrnls, bcur.Meta.DBpath, stg, l, reporter)
+	return b.uploadPhysical(ctx, bcp, rsMeta, data, jrnls, bcur.Meta.DBpath, stg, l, reporter, progressTracker)
 }
 
 func (b *Backup) handleExternal(
@@ -527,6 +530,7 @@ func (b *Backup) uploadPhysical(
 	stg storage.Storage,
 	l log.LogEvent,
 	reporter *progress.Reporter,
+	progressTracker *physicalUploadProgress,
 ) error {
 	numWorkers := b.getNumParallelFiles()
 	if numWorkers > 1 {
@@ -546,7 +550,7 @@ func (b *Backup) uploadPhysical(
 		bcp.CompressionLevel,
 		b.getBackupBufSize(),
 		numWorkers,
-		reporter,
+		progressTracker,
 	)
 	if err != nil {
 		return errors.Wrap(err, "upload data files")
@@ -565,7 +569,7 @@ func (b *Backup) uploadPhysical(
 		bcp.CompressionLevel,
 		b.getBackupBufSize(),
 		numWorkers,
-		reporter,
+		progressTracker,
 	)
 	if err != nil {
 		return errors.Wrap(err, "upload journal files")
@@ -771,7 +775,7 @@ func uploadFiles(
 	comprL *int,
 	bufSize int,
 	numWorkers int,
-	reporter *progress.Reporter,
+	progressTracker *physicalUploadProgress,
 ) ([]File, error) {
 	if len(files) == 0 {
 		return nil, nil
@@ -824,6 +828,7 @@ func uploadFiles(
 				bufs.cp,
 				bufs.save,
 				bufs.fsSave,
+				progressTracker,
 			)
 			if err != nil {
 				return errors.Wrapf(err, "upload file `%s`", s.file.Name)
@@ -831,9 +836,6 @@ func uploadFiles(
 			fw.Name = fname
 
 			results[i] = *fw
-			if reporter != nil {
-				reporter.AddBytes(fw.StgSize)
-			}
 			return nil
 		})
 	}
@@ -855,6 +857,7 @@ func writeFile(
 	cpBuf []byte,
 	saveBuf []byte,
 	fsSaveBuf []byte,
+	progressTracker *physicalUploadProgress,
 ) (*File, error) {
 	fstat, err := os.Stat(file.Name)
 	if err != nil {
@@ -877,6 +880,11 @@ func writeFile(
 	if len(cpBuf) > 0 {
 		src = NewFileReader(*file, cpBuf)
 	}
+	var progressState *physicalFileProgress
+	if progressTracker != nil {
+		progressState = progressTracker.newFile(sz)
+		src = &physicalProgressSource{src: src, tracker: progressTracker, state: progressState}
+	}
 	_, err = storage.UploadWithOpts(ctx, src, stg, compression, compressLevel, dst,
 		sz, saveBuf, fsSaveBuf)
 	if err != nil {
@@ -887,6 +895,9 @@ func writeFile(
 	if err != nil {
 		return nil, errors.Wrapf(err, "get storage file stat %s", dst)
 	}
+	if progressTracker != nil {
+		progressTracker.completeFile(progressState, sz, finf.Size)
+	}
 
 	return &File{
 		Name:                file.Name,
@@ -898,3 +909,116 @@ func writeFile(
 		Len:                 file.Len,
 	}, nil
 }
+
+type physicalUploadProgress struct {
+	reporter    *progress.Reporter
+	totalSource int64
+
+	mu              sync.Mutex
+	completedSource int64
+	completedStg    int64
+}
+
+type physicalFileProgress struct {
+	estimatedStg int64
+}
+
+func newPhysicalUploadProgress(reporter *progress.Reporter, totalSource int64) *physicalUploadProgress {
+	if reporter == nil || totalSource <= 0 {
+		return nil
+	}
+	return &physicalUploadProgress{reporter: reporter, totalSource: totalSource}
+}
+
+func (p *physicalUploadProgress) newFile(_ int64) *physicalFileProgress {
+	return &physicalFileProgress{}
+}
+
+func (p *physicalUploadProgress) addSource(f *physicalFileProgress, n int64) {
+	if p == nil || f == nil || n <= 0 {
+		return
+	}
+
+	p.mu.Lock()
+	estimated := int64(float64(n) * p.ratioLocked())
+	f.estimatedStg += estimated
+	p.mu.Unlock()
+
+	p.reporter.AddBytes(estimated)
+}
+
+func (p *physicalUploadProgress) completeFile(f *physicalFileProgress, sourceSize, stgSize int64) {
+	if p == nil || f == nil {
+		return
+	}
+
+	p.mu.Lock()
+	correction := stgSize - f.estimatedStg
+	p.completedSource += sourceSize
+	p.completedStg += stgSize
+	totalStg := int64(float64(p.totalSource) * p.ratioLocked())
+	if totalStg < p.completedStg {
+		totalStg = p.completedStg
+	}
+	p.mu.Unlock()
+
+	p.reporter.AddBytes(correction)
+	p.reporter.SetTotalBytes(totalStg)
+}
+
+func (p *physicalUploadProgress) ratioLocked() float64 {
+	if p.completedSource <= 0 {
+		return 1
+	}
+	return float64(p.completedStg) / float64(p.completedSource)
+}
+
+type physicalProgressSource struct {
+	src     storage.Source
+	tracker *physicalUploadProgress
+	state   *physicalFileProgress
+}
+
+func (s *physicalProgressSource) WriteTo(w io.Writer) (int64, error) {
+	cw := &physicalProgressWriter{
+		w:         w,
+		tracker:   s.tracker,
+		state:     s.state,
+		threshold: progress.DefaultProgressThresholdBytes,
+	}
+	n, err := s.src.WriteTo(cw)
+	cw.flush()
+	return n, err
+}
+
+type physicalProgressWriter struct {
+	w         io.Writer
+	tracker   *physicalUploadProgress
+	state     *physicalFileProgress
+	threshold int64
+	pending   int64
+}
+
+func (w *physicalProgressWriter) Write(p []byte) (int, error) {
+	n, err := w.w.Write(p)
+	w.add(int64(n))
+	return n, err
+}
+
+func (w *physicalProgressWriter) add(n int64) {
+	if n <= 0 {
+		return
+	}
+	w.pending += n
+	if w.pending >= w.threshold {
+		w.flush()
+	}
+}
+
+func (w *physicalProgressWriter) flush() {
+	if w.pending <= 0 {
+		return
+	}
+	w.tracker.addSource(w.state, w.pending)
+	w.pending = 0
+}
diff --git a/pbm/backup/physical_test.go b/pbm/backup/physical_test.go
index 8179929bb..694a3bf7e 100644
--- a/pbm/backup/physical_test.go
+++ b/pbm/backup/physical_test.go
@@ -294,7 +294,7 @@ func BenchmarkCopyGenFilesToFSStorage(b *testing.B) {
 
 		ts := time.Now()
 		f, err := writeFile(context.Background(), &File{Name: tmpFile.Name()},
-			dst, stg, cType, cLevel, cpBuf, saveBuf, fsSaveBuf)
+			dst, stg, cType, cLevel, cpBuf, saveBuf, fsSaveBuf, nil)
 		if err != nil {
 			b.Fatalf("writeFile: %v", err)
 		}
@@ -358,7 +358,7 @@ func BenchmarkCopyFileToFSStorage(b *testing.B) {
 
 		ts := time.Now()
 		f, err := writeFile(context.Background(), &File{Name: *localFile},
-			dst, stg, cType, cLevel, cpBuf, saveBuf, fsSaveBuf)
+			dst, stg, cType, cLevel, cpBuf, saveBuf, fsSaveBuf, nil)
 		if err != nil {
 			b.Fatalf("writeFile: %v", err)
 		}
diff --git a/pbm/progress/progress.go b/pbm/progress/progress.go
index c5dfab11d..fd469d296 100644
--- a/pbm/progress/progress.go
+++ b/pbm/progress/progress.go
@@ -185,8 +185,19 @@ func NewReporter(
 }
 
 func (r *Reporter) AddBytes(n int64) {
-	if n > 0 {
-		r.doneB.Add(n)
+	if n == 0 {
+		return
+	}
+
+	for {
+		old := r.doneB.Load()
+		next := old + n
+		if next < 0 {
+			next = 0
+		}
+		if r.doneB.CompareAndSwap(old, next) {
+			return
+		}
 	}
 }
 

From b94d9c6ae13b1c80580830d08d2beccbcf82ee9e Mon Sep 17 00:00:00 2001
From: igroene <9805809+igroene@users.noreply.github.com>
Date: Mon, 6 Jul 2026 12:29:16 -0300
Subject: [PATCH 8/9] fix physical restore eta

---
 pbm/progress/progress.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/pbm/progress/progress.go b/pbm/progress/progress.go
index fd469d296..e8409de7e 100644
--- a/pbm/progress/progress.go
+++ b/pbm/progress/progress.go
@@ -253,6 +253,7 @@ func (r *Reporter) Close(final string) {
 	close(r.stop)
 	p := r.Snapshot()
 	if r.log != nil {
+		r.log.Info("progress: %s", p.StringAt(time.Now().Unix()))
 		r.log.Info("%s after %s", final, FormatDuration(p.Elapsed(time.Now().Unix())))
 	}
 }

From 5a5ba3118ace268c9f40100eec223ff242a91e2b Mon Sep 17 00:00:00 2001
From: igroene <9805809+igroene@users.noreply.github.com>
Date: Mon, 6 Jul 2026 12:49:26 -0300
Subject: [PATCH 9/9] fix physical restore eta

---
 pbm/restore/physical.go | 5 +----
 1 file changed, 1 insertion(+), 4 deletions(-)

diff --git a/pbm/restore/physical.go b/pbm/restore/physical.go
index 5eee38870..fe0d2cedb 100644
--- a/pbm/restore/physical.go
+++ b/pbm/restore/physical.go
@@ -1692,10 +1692,7 @@ func (r *PhysRestore) copyFiles() (*storage.DownloadStat, error) {
 
 	setName := util.MakeReverseRSMapFunc(r.rsMap)(r.nodeInfo.SetName)
 	jobs := r.planCopyFiles(setName)
-	reporter := progresspkg.NewReporter(context.Background(), r.log, time.Minute, plannedDownloadSize(jobs), 0,
-		func(ctx context.Context, p progresspkg.Progress) error {
-			return SetRestoreRSProgress(ctx, r.leadConn, r.name, r.nodeInfo.SetName, p)
-		})
+	reporter := progresspkg.NewReporter(context.Background(), r.log, time.Minute, plannedDownloadSize(jobs), 0, nil)
 	defer reporter.Close("restore transfer finished")
 
 	numWorkers := r.GetNumParallelFiles()