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 e752107b1..2f44c5f05 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"`
@@ -1224,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/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..fa7bdb477 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"
@@ -47,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.
@@ -55,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
@@ -64,6 +67,11 @@ func (b *Backup) doLogical(
 	if err != nil {
 		return errors.Wrap(err, "add shard's metadata")
 	}
+	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)
+		})
+	defer reporter.Close("backup transfer finished")
 
 	if inf.IsLeader() {
 		err := b.reconcileStatus(ctx,
@@ -178,7 +186,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 +210,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, done bool) {
+			reporter.AddBytes(bytes)
+			if done && 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..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"
@@ -22,6 +24,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 +331,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 +384,15 @@ 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)
+	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, progressTracker)
 }
 
 func (b *Backup) handleExternal(
@@ -519,6 +529,8 @@ func (b *Backup) uploadPhysical(
 	dbpath string,
 	stg storage.Storage,
 	l log.LogEvent,
+	reporter *progress.Reporter,
+	progressTracker *physicalUploadProgress,
 ) error {
 	numWorkers := b.getNumParallelFiles()
 	if numWorkers > 1 {
@@ -538,6 +550,7 @@ func (b *Backup) uploadPhysical(
 		bcp.CompressionLevel,
 		b.getBackupBufSize(),
 		numWorkers,
+		progressTracker,
 	)
 	if err != nil {
 		return errors.Wrap(err, "upload data files")
@@ -556,6 +569,7 @@ func (b *Backup) uploadPhysical(
 		bcp.CompressionLevel,
 		b.getBackupBufSize(),
 		numWorkers,
+		progressTracker,
 	)
 	if err != nil {
 		return errors.Wrap(err, "upload journal files")
@@ -583,6 +597,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 +631,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 +775,7 @@ func uploadFiles(
 	comprL *int,
 	bufSize int,
 	numWorkers int,
+	progressTracker *physicalUploadProgress,
 ) ([]File, error) {
 	if len(files) == 0 {
 		return nil, nil
@@ -786,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)
@@ -814,6 +857,7 @@ func writeFile(
 	cpBuf []byte,
 	saveBuf []byte,
 	fsSaveBuf []byte,
+	progressTracker *physicalUploadProgress,
 ) (*File, error) {
 	fstat, err := os.Stat(file.Name)
 	if err != nil {
@@ -836,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 {
@@ -846,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,
@@ -857,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/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..e8409de7e
--- /dev/null
+++ b/pbm/progress/progress.go
@@ -0,0 +1,306 @@
+package progress
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"strings"
+	"sync/atomic"
+	"time"
+
+	"github.com/percona/percona-backup-mongodb/pbm/storage"
+)
+
+const (
+	mb                            = 1024 * 1024
+	DefaultProgressThresholdBytes = 16 << 20
+)
+
+// 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"`
+	// ThroughputBytesPerSecond is the last known completed transfer rate.
+	ThroughputBytesPerSecond int64 `bson:"throughput_bps,omitempty" json:"throughput_bps,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 {
+	_ = now
+	if p.ThroughputBytesPerSecond > 0 {
+		return float64(p.ThroughputBytesPerSecond) / float64(mb)
+	}
+
+	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 {
+	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))
+	}
+	if eta, ok := p.ETA(); ok {
+		parts = append(parts, "eta="+FormatDuration(eta))
+	}
+
+	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
+	lastTick  atomic.Int64
+	lastBytes atomic.Int64
+	lastBPS   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.lastTick.Store(r.startedAt)
+	_ = 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.TickSnapshot()
+				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 {
+		return
+	}
+
+	for {
+		old := r.doneB.Load()
+		next := old + n
+		if next < 0 {
+			next = 0
+		}
+		if r.doneB.CompareAndSwap(old, next) {
+			return
+		}
+	}
+}
+
+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(),
+		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())
+}
+
+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("progress: %s", p.StringAt(time.Now().Unix()))
+		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/logical.go b/pbm/restore/logical.go
index 1f485747d..2e7cd3cf7 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")
@@ -1061,10 +1089,16 @@ 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, restoreTotalBytes, 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 +1118,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 +1131,13 @@ func (r *Restore) RunSnapshot(
 		},
 		bcp.Compression,
 		util.MakeSelectedPred(nss),
-		r.numParallelColls)
+		r.numParallelColls,
+		func(ns string, bytes int64, done bool) {
+			reporter.AddBytes(bytes)
+			if done && ns != archive.MetaFile {
+				reporter.AddItems(1)
+			}
+		})
 	if err != nil {
 		return "", err
 	}
@@ -1138,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,
@@ -1238,20 +1290,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 +1340,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..fe0d2cedb 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,8 @@ 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, nil)
+	defer reporter.Close("restore transfer finished")
 
 	numWorkers := r.GetNumParallelFiles()
 	if numWorkers > 1 {
@@ -1727,7 +1738,8 @@ 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 {
+				_, err := r.copyFile(op.src, job.dst, op.fMeta, op.cmpr, cpBuf, reporter)
+				if err != nil {
 					return err
 				}
 			}
@@ -1738,34 +1750,66 @@ 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,
+	reporter *progresspkg.Reporter,
+) (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)
 	}
+	sr = progresspkg.NewCountingReadCloser(sr, reporter, progresspkg.DefaultProgressThresholdBytes)
 	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 +1823,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..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)
 		}
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..544c37e22 100644
--- a/pbm/snapshot/dump.go
+++ b/pbm/snapshot/dump.go
@@ -11,6 +11,9 @@ import (
 )
 
 type UploadFunc func(ns, ext string, r io.Reader) error
+type ProgressFunc func(ns string, bytes int64, done bool)
+
+const progressThresholdBytes = 16 << 20
 
 func UploadDump(
 	ctx context.Context,
@@ -18,6 +21,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)
 
@@ -33,7 +47,7 @@ func UploadDump(
 		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)
@@ -41,6 +55,7 @@ func UploadDump(
 				done <- err
 			}
 
+			rc.finish()
 			atomic.AddInt64(&uploadSize, rc.n)
 		}()
 
@@ -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,17 +122,78 @@ func DownloadDump(
 	return pr, nil
 }
 
+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.add(int64(n))
+	return n, err
+}
+
+func (c *readCounterCloser) Close() error {
+	err := c.ReadCloser.Close()
+	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 {