diff --git a/go/base/context.go b/go/base/context.go
index 5a06d0d4d..4d0c3b09d 100644
--- a/go/base/context.go
+++ b/go/base/context.go
@@ -238,7 +238,7 @@ type MigrationContext struct {
 	AbortError error
 	abortMutex *sync.Mutex
 
-	Metrics metrics.MemStatsGaugeEmitter
+	Metrics metrics.Emitter
 
 	OriginalTableColumnsOnApplier    *sql.ColumnList
 	OriginalTableColumns             *sql.ColumnList
diff --git a/go/logic/migrator.go b/go/logic/migrator.go
index 5ed2e2d9a..e21700ab0 100644
--- a/go/logic/migrator.go
+++ b/go/logic/migrator.go
@@ -130,7 +130,7 @@ func NewMigrator(context *base.MigrationContext, appVersion string) *Migrator {
 
 // sleepWhileTrue sleeps indefinitely until the given function returns 'false'
 // (or fails with error)
-func (mgtr *Migrator) sleepWhileTrue(operation func() (bool, error)) error {
+func (mgtr *Migrator) sleepWhileTrue(stage string, operation func() (bool, error)) error {
 	for {
 		// Check for abort before continuing
 		if err := mgtr.checkAbort(); err != nil {
@@ -143,6 +143,7 @@ func (mgtr *Migrator) sleepWhileTrue(operation func() (bool, error)) error {
 		if !shouldSleep {
 			return nil
 		}
+		metrics.RecordSleep(mgtr.migrationContext.Metrics, stage, time.Second)
 		time.Sleep(time.Second)
 	}
 }
@@ -166,7 +167,9 @@ func (mgtr *Migrator) retryOperation(operation func() error, notFatalHint ...boo
 	for i := 0; i < maxRetries; i++ {
 		if i != 0 {
 			// sleep after previous iteration
-			RetrySleepFn(1 * time.Second)
+			sleepDuration := 1 * time.Second
+			metrics.RecordSleep(mgtr.migrationContext.Metrics, "retry_backoff", sleepDuration)
+			RetrySleepFn(sleepDuration)
 		}
 		// Check for abort/context cancellation before each retry
 		if abortErr := mgtr.checkAbort(); abortErr != nil {
@@ -207,7 +210,9 @@ func (mgtr *Migrator) retryOperationWithExponentialBackoff(operation func() erro
 		)
 
 		if i != 0 {
-			RetrySleepFn(time.Duration(interval) * time.Second)
+			sleepDuration := time.Duration(interval) * time.Second
+			metrics.RecordSleep(mgtr.migrationContext.Metrics, "retry_backoff", sleepDuration)
+			RetrySleepFn(sleepDuration)
 		}
 		// Check for abort/context cancellation before each retry
 		if abortErr := mgtr.checkAbort(); abortErr != nil {
@@ -842,6 +847,7 @@ func (mgtr *Migrator) cutOver() (err error) {
 	mgtr.migrationContext.MarkPointOfInterest()
 	mgtr.migrationContext.Log.Debugf("checking for cut-over postpone")
 	if err := mgtr.sleepWhileTrue(
+		"cut_over_postpone",
 		func() (bool, error) {
 			heartbeatLag := mgtr.migrationContext.TimeSinceLastHeartbeatOnChangelog()
 			maxLagMillisecondsThrottle := time.Duration(atomic.LoadInt64(&mgtr.migrationContext.MaxLagMillisecondsThrottleThreshold)) * time.Millisecond
@@ -1168,7 +1174,7 @@ func (mgtr *Migrator) initiateInspector() (err error) {
 	return nil
 }
 
-// emitProgressMetrics emits StatsD gauges from a progress snapshot.
+// emitProgressMetrics emits StatsD gauges
 func (mgtr *Migrator) emitProgressMetrics(snap migrationProgressSnapshot) {
 	metrics.EmitProgressGauges(
 		mgtr.migrationContext.Metrics,
@@ -1181,6 +1187,13 @@ func (mgtr *Migrator) emitProgressMetrics(snap migrationProgressSnapshot) {
 		snap.applyEventsBacklog,
 		snap.applyEventsCapacity,
 	)
+	isThrottled, _, _ := mgtr.migrationContext.IsThrottled()
+	metrics.EmitLagGauges(
+		mgtr.migrationContext.Metrics,
+		snap.replicationLagSeconds,
+		snap.heartbeatLagSeconds,
+		isThrottled,
+	)
 }
 
 // reportStatus samples progress, emits metrics, and optionally prints status output.
@@ -1754,7 +1767,9 @@ func (mgtr *Migrator) Checkpoint(ctx context.Context) (*Checkpoint, error) {
 			return chk, err
 		}
 		mgtr.applier.CurrentCoordinatesMutex.Unlock()
-		time.Sleep(500 * time.Millisecond)
+		sleepDuration := 500 * time.Millisecond
+		metrics.RecordSleep(mgtr.migrationContext.Metrics, "replica_wait", sleepDuration)
+		time.Sleep(sleepDuration)
 	}
 }
 
@@ -1859,7 +1874,10 @@ func (mgtr *Migrator) executeWriteFuncs() error {
 							copyRowsDuration := time.Since(copyRowsStartTime)
 							sleepTimeNanosecondFloat64 := niceRatio * float64(copyRowsDuration.Nanoseconds())
 							sleepTime := time.Duration(int64(sleepTimeNanosecondFloat64)) * time.Nanosecond
-							time.Sleep(sleepTime)
+							if sleepTime > 0 {
+								metrics.RecordSleep(mgtr.migrationContext.Metrics, "chunk_throttle", sleepTime)
+								time.Sleep(sleepTime)
+							}
 						}
 					}
 				default:
diff --git a/go/logic/migrator_test.go b/go/logic/migrator_test.go
index 31e161b5e..d939ae97d 100644
--- a/go/logic/migrator_test.go
+++ b/go/logic/migrator_test.go
@@ -410,11 +410,19 @@ func TestMigratorGetMigrationStateAndETA(t *testing.T) {
 type progressGaugeSpy struct {
 	names  []string
 	values []float64
+	tags   [][]string
 }
 
-func (s *progressGaugeSpy) Gauge(name string, value float64, _ ...string) {
+func (s *progressGaugeSpy) Gauge(name string, value float64, tags ...string) {
 	s.names = append(s.names, name)
 	s.values = append(s.values, value)
+	s.tags = append(s.tags, append([]string(nil), tags...))
+}
+
+func (s *progressGaugeSpy) Count(name string, value int64, tags ...string) {
+}
+
+func (s *progressGaugeSpy) Histogram(name string, value float64, tags ...string) {
 }
 
 func TestReportStatusEmitsProgressGaugesEveryTick(t *testing.T) {
@@ -427,7 +435,7 @@ func TestReportStatusEmitsProgressGaugesEveryTick(t *testing.T) {
 
 	migrator := NewMigrator(ctx, "test")
 	migrator.reportStatus(NoPrintStatusRule, io.Discard)
-	require.Len(t, spy.names, 6)
+	require.Len(t, spy.names, 8)
 	assert.Equal(t, []string{
 		"row_copy.rows_copied",
 		"row_copy.rows_estimate",
@@ -435,16 +443,18 @@ func TestReportStatusEmitsProgressGaugesEveryTick(t *testing.T) {
 		"binlog.backlog_size",
 		"binlog.backlog_capacity",
 		"binlog.backlog_utilization",
+		"lag.replication_seconds",
+		"lag.heartbeat_seconds",
 	}, spy.names)
-	assert.Equal(t, []float64{1000, 5000, 42, 0, float64(cap(migrator.applyEventsQueue)), 0}, spy.values)
+	assert.Equal(t, []float64{1000, 5000, 42, 0, float64(cap(migrator.applyEventsQueue)), 0}, spy.values[:6])
 	assert.InDelta(t, 20.0, ctx.GetProgressPct(), 0.01)
 
 	spy.names = nil
 	spy.values = nil
 	atomic.StoreInt64(&ctx.TotalDMLEventsApplied, 100)
 	migrator.reportStatus(NoPrintStatusRule, io.Discard)
-	require.Len(t, spy.names, 6)
-	assert.Equal(t, []float64{1000, 5000, 100, 0, float64(cap(migrator.applyEventsQueue)), 0}, spy.values)
+	require.Len(t, spy.names, 8)
+	assert.Equal(t, []float64{1000, 5000, 100, 0, float64(cap(migrator.applyEventsQueue)), 0}, spy.values[:6])
 }
 
 func TestReportStatusEmitsBinlogBacklogGauges(t *testing.T) {
@@ -463,7 +473,7 @@ func TestReportStatusEmitsBinlogBacklogGauges(t *testing.T) {
 	migrator.reportStatus(NoPrintStatusRule, io.Discard)
 
 	capacity := float64(cap(migrator.applyEventsQueue))
-	require.Len(t, spy.names, 6)
+	require.Len(t, spy.names, 8)
 	assert.Equal(t, float64(2), spy.values[3])
 	assert.Equal(t, capacity, spy.values[4])
 	assert.InDelta(t, 2/capacity, spy.values[5], 1e-9)
@@ -480,7 +490,7 @@ func TestReportStatusEmitsGaugesWhenRowCopyComplete(t *testing.T) {
 	atomic.StoreInt64(&migrator.rowCopyCompleteFlag, 1)
 	migrator.reportStatus(NoPrintStatusRule, io.Discard)
 
-	require.Len(t, spy.names, 6)
+	require.Len(t, spy.names, 8)
 	assert.Equal(t, float64(5000), spy.values[0])
 	assert.Equal(t, float64(5000), spy.values[1], "rows_estimate tracks rows_copied when row copy is complete")
 }
@@ -500,8 +510,27 @@ func TestReportStatusEmitsGaugesWhenPrintSuppressed(t *testing.T) {
 	require.False(t, migrator.shouldPrintStatus(HeuristicPrintStatusRule, snap.elapsedSeconds, etaDuration))
 
 	migrator.reportStatus(HeuristicPrintStatusRule, io.Discard)
-	require.Len(t, spy.names, 6)
-	assert.Equal(t, []float64{1000, 5000, 0, 0, float64(cap(migrator.applyEventsQueue)), 0}, spy.values)
+	require.Len(t, spy.names, 8)
+	assert.Equal(t, []float64{1000, 5000, 0, 0, float64(cap(migrator.applyEventsQueue)), 0}, spy.values[:6])
+}
+
+func TestReportStatusEmitsLagGaugesWhenThrottled(t *testing.T) {
+	spy := &progressGaugeSpy{}
+	ctx := base.NewMigrationContext()
+	ctx.Metrics = spy
+	ctx.SetThrottled(true, "max-lag-millis", base.NoThrottleReasonHint)
+	atomic.StoreInt64(&ctx.CurrentLag, int64(5*time.Second))
+	ctx.SetLastHeartbeatOnChangelogTime(time.Now().Add(-4 * time.Second))
+
+	migrator := NewMigrator(ctx, "test")
+	migrator.reportStatus(NoPrintStatusRule, io.Discard)
+
+	require.GreaterOrEqual(t, len(spy.names), 8)
+	assert.Equal(t, "lag.replication_seconds", spy.names[6])
+	assert.Equal(t, "lag.heartbeat_seconds", spy.names[7])
+	require.Len(t, spy.tags[6], 1)
+	assert.Equal(t, "throttled:true", spy.tags[6][0])
+	assert.Equal(t, "throttled:true", spy.tags[7][0])
 }
 
 func TestMigratorShouldPrintStatus(t *testing.T) {
diff --git a/go/metrics/binlog_backlog.go b/go/metrics/binlog_backlog.go
deleted file mode 100644
index 2d6ea5a4b..000000000
--- a/go/metrics/binlog_backlog.go
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
-   Copyright 2026 GitHub Inc.
-	 See https://github.com/github/gh-ost/blob/master/LICENSE
-*/
-
-package metrics
-
-// EmitBinlogBacklogGauges emits apply-events queue depth gauges (namespace is applied by the client):
-// gh_ost.binlog.backlog_size, gh_ost.binlog.backlog_capacity, gh_ost.binlog.backlog_utilization.
-func EmitBinlogBacklogGauges(emit MemStatsGaugeEmitter, backlogSize, backlogCapacity int) {
-	if emit == nil {
-		return
-	}
-	emit.Gauge("binlog.backlog_size", float64(backlogSize))
-	emit.Gauge("binlog.backlog_capacity", float64(backlogCapacity))
-	emit.Gauge("binlog.backlog_utilization", binlogBacklogUtilization(backlogSize, backlogCapacity))
-}
-
-func binlogBacklogUtilization(backlogSize, backlogCapacity int) float64 {
-	if backlogCapacity <= 0 {
-		return 0
-	}
-	utilization := float64(backlogSize) / float64(backlogCapacity)
-	if utilization > 1 {
-		return 1
-	}
-	if utilization < 0 {
-		return 0
-	}
-	return utilization
-}
diff --git a/go/metrics/binlog_backlog_test.go b/go/metrics/binlog_backlog_test.go
deleted file mode 100644
index 4f99dc1b1..000000000
--- a/go/metrics/binlog_backlog_test.go
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
-   Copyright 2026 GitHub Inc.
-	 See https://github.com/github/gh-ost/blob/master/LICENSE
-*/
-
-package metrics
-
-import "testing"
-
-func TestEmitBinlogBacklogGauges(t *testing.T) {
-	spy := &gaugeSpy{}
-	EmitBinlogBacklogGauges(spy, 250, 1000)
-
-	wantNames := []string{
-		"binlog.backlog_size",
-		"binlog.backlog_capacity",
-		"binlog.backlog_utilization",
-	}
-	wantVals := []float64{250, 1000, 0.25}
-
-	if len(spy.names) != len(wantNames) {
-		t.Fatalf("got %d gauges, want %d", len(spy.names), len(wantNames))
-	}
-	for i := range wantNames {
-		if spy.names[i] != wantNames[i] || spy.values[i] != wantVals[i] {
-			t.Fatalf("[%d] got %s=%v want %s=%v", i, spy.names[i], spy.values[i], wantNames[i], wantVals[i])
-		}
-	}
-}
-
-func TestEmitBinlogBacklogGauges_nilSafe(t *testing.T) {
-	EmitBinlogBacklogGauges(nil, 1, 2)
-}
-
-func TestBinlogBacklogUtilization(t *testing.T) {
-	tests := []struct {
-		size, capacity int
-		want           float64
-	}{
-		{0, 1000, 0},
-		{250, 1000, 0.25},
-		{1000, 1000, 1},
-		{1500, 1000, 1},
-		{-1, 1000, 0},
-		{10, 0, 0},
-	}
-	for _, tt := range tests {
-		got := binlogBacklogUtilization(tt.size, tt.capacity)
-		if got != tt.want {
-			t.Fatalf("utilization(%d, %d) = %v, want %v", tt.size, tt.capacity, got, tt.want)
-		}
-	}
-}
diff --git a/go/metrics/catalog.md b/go/metrics/catalog.md
new file mode 100644
index 000000000..5fcb8c847
--- /dev/null
+++ b/go/metrics/catalog.md
@@ -0,0 +1,15 @@
+# gh-ost metrics catalog
+
+Metrics are emitted with the `gh_ost.` namespace.
+
+## Sleep and wait metrics
+
+- `gh_ost.sleep.duration_milliseconds` — histogram of one sleep/wait interval in milliseconds, tagged with `stage`.
+- `gh_ost.sleep.total_milliseconds` — count of milliseconds spent sleeping/waiting, tagged with `stage`.
+
+### Sleep stages
+
+- `cut_over_postpone` — waiting for cut-over postponement or heartbeat lag preconditions to clear before the final table swap.
+- `chunk_throttle` — `--nice-ratio` sleep after a row-copy chunk.
+- `retry_backoff` — fixed or exponential retry backoff between failed operations.
+- `replica_wait` — waiting for streamed coordinates to catch up before checkpointing.
diff --git a/go/metrics/client.go b/go/metrics/client.go
index ed6acc096..1d78df216 100644
--- a/go/metrics/client.go
+++ b/go/metrics/client.go
@@ -21,6 +21,13 @@ type Client struct {
 	sd *statsd.Client
 }
 
+// Emitter is implemented by *Client; used for tests without UDP.
+type Emitter interface {
+	Gauge(name string, value float64, tags ...string)
+	Count(name string, value int64, tags ...string)
+	Histogram(name string, value float64, tags ...string)
+}
+
 // NewClient connects to addr for StatsD. If addr is empty, returns Noop and nil error.
 // namespace is typically "gh_ost." (metrics are named namespace + short name, e.g. gh_ost.startup).
 // tags are global tags applied to every metric (repeatable --statsd-tags).
@@ -60,6 +67,13 @@ func (c *Client) Count(name string, value int64, tags ...string) {
 	_ = c.sd.Count(name, value, tags, 1.0)
 }
 
+func (c *Client) Histogram(name string, value float64, tags ...string) {
+	if c.sd == nil {
+		return
+	}
+	_ = c.sd.Histogram(name, value, tags, 1.0)
+}
+
 // Close flushes buffered metrics; safe for Noop.
 func (c *Client) Close() error {
 	if c.sd == nil {
diff --git a/go/metrics/emit.go b/go/metrics/emit.go
new file mode 100644
index 000000000..d5337a5e3
--- /dev/null
+++ b/go/metrics/emit.go
@@ -0,0 +1,132 @@
+/*
+   Copyright 2026 GitHub Inc.
+	 See https://github.com/github/gh-ost/blob/master/LICENSE
+*/
+
+package metrics
+
+import (
+	"context"
+	"fmt"
+	"runtime"
+	"time"
+)
+
+// EmitProgressGauges emits row-copy and DML progress gauges (namespace is applied by the client):
+// gh_ost.row_copy.rows_copied, gh_ost.row_copy.rows_estimate, gh_ost.dml.events_applied.
+func EmitProgressGauges(emit Emitter, rowsCopied, rowsEstimate, dmlEventsApplied int64) {
+	if emit == nil {
+		return
+	}
+	emit.Gauge("row_copy.rows_copied", float64(rowsCopied))
+	emit.Gauge("row_copy.rows_estimate", float64(rowsEstimate))
+	emit.Gauge("dml.events_applied", float64(dmlEventsApplied))
+}
+
+// EmitBinlogBacklogGauges emits apply-events queue depth gauges (namespace is applied by the client):
+// gh_ost.binlog.backlog_size, gh_ost.binlog.backlog_capacity, gh_ost.binlog.backlog_utilization.
+func EmitBinlogBacklogGauges(emit Emitter, backlogSize, backlogCapacity int) {
+	if emit == nil {
+		return
+	}
+	emit.Gauge("binlog.backlog_size", float64(backlogSize))
+	emit.Gauge("binlog.backlog_capacity", float64(backlogCapacity))
+	emit.Gauge("binlog.backlog_utilization", binlogBacklogUtilization(backlogSize, backlogCapacity))
+}
+
+func binlogBacklogUtilization(backlogSize, backlogCapacity int) float64 {
+	if backlogCapacity <= 0 {
+		return 0
+	}
+	utilization := float64(backlogSize) / float64(backlogCapacity)
+	if utilization > 1 {
+		return 1
+	}
+	if utilization < 0 {
+		return 0
+	}
+	return utilization
+}
+
+// EmitLagGauges emits replication and heartbeat lag gauges (namespace is applied by the client):
+// gh_ost.lag.replication_seconds, gh_ost.lag.heartbeat_seconds, each tagged throttled:true|false.
+//
+// These are point-in-time readings each status tick (not a distribution), so gauges are used
+// rather than histograms; DogStatsD histogram aggregation exposes count/max series that do not
+// match the log line lag values in Prometheus/Grafana.
+func EmitLagGauges(emit Emitter, replicationLagSeconds, heartbeatLagSeconds float64, throttled bool) {
+	if emit == nil {
+		return
+	}
+	tags := []string{fmt.Sprintf("throttled:%t", throttled)}
+	emit.Gauge("lag.replication_seconds", replicationLagSeconds, tags...)
+	emit.Gauge("lag.heartbeat_seconds", heartbeatLagSeconds, tags...)
+}
+
+// EmitGoRuntimeGauges emits gh_ost.go_runtime.* gauges (namespace is applied by the client).
+// m and numGoroutine are typically from runtime.ReadMemStats and runtime.NumGoroutine.
+func EmitGoRuntimeGauges(emit Emitter, m *runtime.MemStats, numGoroutine int) {
+	if emit == nil || m == nil {
+		return
+	}
+	emit.Gauge("go_runtime.alloc_bytes", float64(m.Alloc))
+	emit.Gauge("go_runtime.sys_bytes", float64(m.Sys))
+	emit.Gauge("go_runtime.heap_inuse_bytes", float64(m.HeapInuse))
+	emit.Gauge("go_runtime.num_gc", float64(m.NumGC))
+	emit.Gauge("go_runtime.gc_pause_total_ns", float64(m.PauseTotalNs))
+	emit.Gauge("go_runtime.goroutines", float64(numGoroutine))
+}
+
+// StartGoRuntimeReporter periodically samples runtime memory and goroutines and emits gauges
+// until ctx is cancelled. It is a no-op when interval <= 0, client is nil, or StatsD is disabled
+// (noop client).
+func StartGoRuntimeReporter(ctx context.Context, client *Client, interval time.Duration) {
+	if ctx == nil || client == nil || interval <= 0 || client.sd == nil {
+		return
+	}
+
+	emit := func() {
+		var m runtime.MemStats
+		runtime.ReadMemStats(&m)
+		EmitGoRuntimeGauges(client, &m, runtime.NumGoroutine())
+	}
+
+	go func() {
+		ticker := time.NewTicker(interval)
+		defer ticker.Stop()
+
+		emit()
+		for {
+			select {
+			case <-ctx.Done():
+				return
+			case <-ticker.C:
+				emit()
+			}
+		}
+	}()
+}
+
+type sleepHistogramEmitter interface {
+	Histogram(name string, value float64, tags ...string)
+}
+
+type sleepCountEmitter interface {
+	Count(name string, value int64, tags ...string)
+}
+
+// RecordSleep emits per-stage sleep/wait metrics (namespace is applied by the client):
+// gh_ost.sleep.duration_milliseconds and gh_ost.sleep.total_milliseconds, both tagged by stage.
+func RecordSleep(emit any, stage string, d time.Duration) {
+	if emit == nil || stage == "" || d < 0 {
+		return
+	}
+	tags := []string{"stage:" + stage}
+	milliseconds := d.Milliseconds()
+	if histogramEmitter, ok := emit.(sleepHistogramEmitter); ok {
+		histogramEmitter.Histogram("sleep.duration_milliseconds", float64(milliseconds), tags...)
+	}
+	if countEmitter, ok := emit.(sleepCountEmitter); ok {
+		countEmitter.Count("sleep.total_milliseconds", milliseconds, tags...)
+	}
+}
diff --git a/go/metrics/emit_test.go b/go/metrics/emit_test.go
new file mode 100644
index 000000000..4cb12d13e
--- /dev/null
+++ b/go/metrics/emit_test.go
@@ -0,0 +1,250 @@
+/*
+   Copyright 2026 GitHub Inc.
+	 See https://github.com/github/gh-ost/blob/master/LICENSE
+*/
+
+package metrics
+
+import (
+	"context"
+	"runtime"
+	"slices"
+	"testing"
+	"time"
+)
+
+type gaugeSpy struct {
+	names  []string
+	values []float64
+	tags   [][]string
+}
+
+func (g *gaugeSpy) Gauge(name string, value float64, tags ...string) {
+	g.names = append(g.names, name)
+	g.values = append(g.values, value)
+	g.tags = append(g.tags, append([]string(nil), tags...))
+}
+
+func (g *gaugeSpy) Count(name string, value int64, tags ...string) {
+}
+
+func (g *gaugeSpy) Histogram(name string, value float64, tags ...string) {
+}
+
+func TestEmitProgressGauges(t *testing.T) {
+	spy := &gaugeSpy{}
+	EmitProgressGauges(spy, 1000, 5000, 42)
+
+	wantNames := []string{
+		"row_copy.rows_copied",
+		"row_copy.rows_estimate",
+		"dml.events_applied",
+	}
+	wantVals := []float64{1000, 5000, 42}
+
+	if len(spy.names) != len(wantNames) {
+		t.Fatalf("got %d gauges, want %d", len(spy.names), len(wantNames))
+	}
+	for i := range wantNames {
+		if spy.names[i] != wantNames[i] || spy.values[i] != wantVals[i] {
+			t.Fatalf("[%d] got %s=%v want %s=%v", i, spy.names[i], spy.values[i], wantNames[i], wantVals[i])
+		}
+	}
+}
+
+func TestEmitProgressGauges_nilSafe(t *testing.T) {
+	EmitProgressGauges(nil, 1, 2, 3)
+}
+
+func TestEmitBinlogBacklogGauges(t *testing.T) {
+	spy := &gaugeSpy{}
+	EmitBinlogBacklogGauges(spy, 250, 1000)
+
+	wantNames := []string{
+		"binlog.backlog_size",
+		"binlog.backlog_capacity",
+		"binlog.backlog_utilization",
+	}
+	wantVals := []float64{250, 1000, 0.25}
+
+	if len(spy.names) != len(wantNames) {
+		t.Fatalf("got %d gauges, want %d", len(spy.names), len(wantNames))
+	}
+	for i := range wantNames {
+		if spy.names[i] != wantNames[i] || spy.values[i] != wantVals[i] {
+			t.Fatalf("[%d] got %s=%v want %s=%v", i, spy.names[i], spy.values[i], wantNames[i], wantVals[i])
+		}
+	}
+}
+
+func TestEmitBinlogBacklogGauges_nilSafe(t *testing.T) {
+	EmitBinlogBacklogGauges(nil, 1, 2)
+}
+
+func TestBinlogBacklogUtilization(t *testing.T) {
+	tests := []struct {
+		size, capacity int
+		want           float64
+	}{
+		{0, 1000, 0},
+		{250, 1000, 0.25},
+		{1000, 1000, 1},
+		{1500, 1000, 1},
+		{-1, 1000, 0},
+		{10, 0, 0},
+	}
+	for _, tt := range tests {
+		got := binlogBacklogUtilization(tt.size, tt.capacity)
+		if got != tt.want {
+			t.Fatalf("utilization(%d, %d) = %v, want %v", tt.size, tt.capacity, got, tt.want)
+		}
+	}
+}
+
+func TestEmitLagGauges_notThrottled(t *testing.T) {
+	spy := &gaugeSpy{}
+	EmitLagGauges(spy, 2.5, 1.25, false)
+
+	wantNames := []string{"lag.replication_seconds", "lag.heartbeat_seconds"}
+	wantVals := []float64{2.5, 1.25}
+	wantTags := []string{"throttled:false"}
+
+	if len(spy.names) != len(wantNames) {
+		t.Fatalf("got %d gauges, want %d", len(spy.names), len(wantNames))
+	}
+	for i := range wantNames {
+		if spy.names[i] != wantNames[i] || spy.values[i] != wantVals[i] {
+			t.Fatalf("[%d] got %s=%v want %s=%v", i, spy.names[i], spy.values[i], wantNames[i], wantVals[i])
+		}
+		if len(spy.tags[i]) != 1 || spy.tags[i][0] != wantTags[0] {
+			t.Fatalf("[%d] got tags %v want [%s]", i, spy.tags[i], wantTags[0])
+		}
+	}
+}
+
+func TestEmitLagGauges_throttled(t *testing.T) {
+	spy := &gaugeSpy{}
+	EmitLagGauges(spy, 4.0, 3.0, true)
+
+	if len(spy.names) != 2 {
+		t.Fatalf("got %d gauges, want 2", len(spy.names))
+	}
+	for i := range spy.names {
+		if len(spy.tags[i]) != 1 || spy.tags[i][0] != "throttled:true" {
+			t.Fatalf("[%d] got tags %v want [throttled:true]", i, spy.tags[i])
+		}
+	}
+}
+
+func TestEmitLagGauges_nilSafe(t *testing.T) {
+	EmitLagGauges(nil, 1, 2, false)
+}
+
+func TestEmitGoRuntimeGauges(t *testing.T) {
+	spy := &gaugeSpy{}
+	m := &runtime.MemStats{
+		Alloc:        100,
+		Sys:          200,
+		HeapInuse:    300,
+		NumGC:        7,
+		PauseTotalNs: 42,
+	}
+	EmitGoRuntimeGauges(spy, m, 123)
+
+	wantNames := []string{
+		"go_runtime.alloc_bytes",
+		"go_runtime.sys_bytes",
+		"go_runtime.heap_inuse_bytes",
+		"go_runtime.num_gc",
+		"go_runtime.gc_pause_total_ns",
+		"go_runtime.goroutines",
+	}
+	wantVals := []float64{100, 200, 300, 7, 42, 123}
+
+	if len(spy.names) != len(wantNames) {
+		t.Fatalf("got %d gauges, want %d", len(spy.names), len(wantNames))
+	}
+	for i := range wantNames {
+		if spy.names[i] != wantNames[i] || spy.values[i] != wantVals[i] {
+			t.Fatalf("[%d] got %s=%v want %s=%v", i, spy.names[i], spy.values[i], wantNames[i], wantVals[i])
+		}
+	}
+}
+
+func TestEmitGoRuntimeGauges_nilSafe(t *testing.T) {
+	EmitGoRuntimeGauges(nil, &runtime.MemStats{}, 1)
+	EmitGoRuntimeGauges(&gaugeSpy{}, nil, 1)
+}
+
+func TestStartGoRuntimeReporter_stopsOnCancel(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	c := &Client{} // sd nil, so the reporter should not start.
+	StartGoRuntimeReporter(ctx, c, time.Millisecond)
+	cancel()
+	time.Sleep(20 * time.Millisecond)
+}
+
+type sleepSpy struct {
+	histogramNames  []string
+	histogramValues []float64
+	histogramTags   [][]string
+	countNames      []string
+	countValues     []int64
+	countTags       [][]string
+}
+
+func (s *sleepSpy) Histogram(name string, value float64, tags ...string) {
+	s.histogramNames = append(s.histogramNames, name)
+	s.histogramValues = append(s.histogramValues, value)
+	s.histogramTags = append(s.histogramTags, tags)
+}
+
+func (s *sleepSpy) Count(name string, value int64, tags ...string) {
+	s.countNames = append(s.countNames, name)
+	s.countValues = append(s.countValues, value)
+	s.countTags = append(s.countTags, tags)
+}
+
+func TestRecordSleep(t *testing.T) {
+	spy := &sleepSpy{}
+
+	RecordSleep(spy, "retry_backoff", 2*time.Second)
+
+	if len(spy.histogramNames) != 1 {
+		t.Fatalf("got %d histograms, want 1", len(spy.histogramNames))
+	}
+	if spy.histogramNames[0] != "sleep.duration_milliseconds" || spy.histogramValues[0] != 2000 {
+		t.Fatalf("got histogram %s=%v, want sleep.duration_milliseconds=2000", spy.histogramNames[0], spy.histogramValues[0])
+	}
+	if !slices.Equal(spy.histogramTags[0], []string{"stage:retry_backoff"}) {
+		t.Fatalf("got histogram tags %#v", spy.histogramTags[0])
+	}
+	if len(spy.countNames) != 1 {
+		t.Fatalf("got %d counts, want 1", len(spy.countNames))
+	}
+	if spy.countNames[0] != "sleep.total_milliseconds" || spy.countValues[0] != 2000 {
+		t.Fatalf("got count %s=%v, want sleep.total_milliseconds=2000", spy.countNames[0], spy.countValues[0])
+	}
+	if !slices.Equal(spy.countTags[0], []string{"stage:retry_backoff"}) {
+		t.Fatalf("got count tags %#v", spy.countTags[0])
+	}
+}
+
+func TestRecordSleepSubSecond(t *testing.T) {
+	spy := &sleepSpy{}
+
+	RecordSleep(spy, "replica_wait", 500*time.Millisecond)
+
+	if spy.histogramNames[0] != "sleep.duration_milliseconds" || spy.histogramValues[0] != 500 {
+		t.Fatalf("got histogram %s=%v, want sleep.duration_milliseconds=500", spy.histogramNames[0], spy.histogramValues[0])
+	}
+	if spy.countNames[0] != "sleep.total_milliseconds" || spy.countValues[0] != 500 {
+		t.Fatalf("got count %s=%v, want sleep.total_milliseconds=500", spy.countNames[0], spy.countValues[0])
+	}
+}
+
+func TestRecordSleepNilSafe(t *testing.T) {
+	RecordSleep(nil, "retry_backoff", time.Second)
+	RecordSleep(&sleepSpy{}, "", time.Second)
+	RecordSleep(&sleepSpy{}, "retry_backoff", -time.Second)
+}
diff --git a/go/metrics/go_runtime.go b/go/metrics/go_runtime.go
deleted file mode 100644
index 24ae2c6b5..000000000
--- a/go/metrics/go_runtime.go
+++ /dev/null
@@ -1,61 +0,0 @@
-/*
-   Copyright 2022 GitHub Inc.
-	 See https://github.com/github/gh-ost/blob/master/LICENSE
-*/
-
-package metrics
-
-import (
-	"context"
-	"runtime"
-	"time"
-)
-
-// MemStatsGaugeEmitter is implemented by *Client; used for tests without UDP.
-type MemStatsGaugeEmitter interface {
-	Gauge(name string, value float64, tags ...string)
-}
-
-// EmitGoRuntimeGauges emits gh_ost.go_runtime.* gauges (namespace is applied by the client).
-// m and numGoroutine are typically from runtime.ReadMemStats and runtime.NumGoroutine.
-func EmitGoRuntimeGauges(emit MemStatsGaugeEmitter, m *runtime.MemStats, numGoroutine int) {
-	if emit == nil || m == nil {
-		return
-	}
-	emit.Gauge("go_runtime.alloc_bytes", float64(m.Alloc))
-	emit.Gauge("go_runtime.sys_bytes", float64(m.Sys))
-	emit.Gauge("go_runtime.heap_inuse_bytes", float64(m.HeapInuse))
-	emit.Gauge("go_runtime.num_gc", float64(m.NumGC))
-	emit.Gauge("go_runtime.gc_pause_total_ns", float64(m.PauseTotalNs))
-	emit.Gauge("go_runtime.goroutines", float64(numGoroutine))
-}
-
-// StartGoRuntimeReporter periodically samples runtime memory and goroutines and emits gauges
-// until ctx is cancelled. It is a no-op when interval <= 0, client is nil, or StatsD is disabled
-// (noop client).
-func StartGoRuntimeReporter(ctx context.Context, client *Client, interval time.Duration) {
-	if ctx == nil || client == nil || interval <= 0 || client.sd == nil {
-		return
-	}
-
-	emit := func() {
-		var m runtime.MemStats
-		runtime.ReadMemStats(&m)
-		EmitGoRuntimeGauges(client, &m, runtime.NumGoroutine())
-	}
-
-	go func() {
-		ticker := time.NewTicker(interval)
-		defer ticker.Stop()
-
-		emit()
-		for {
-			select {
-			case <-ctx.Done():
-				return
-			case <-ticker.C:
-				emit()
-			}
-		}
-	}()
-}
diff --git a/go/metrics/go_runtime_test.go b/go/metrics/go_runtime_test.go
deleted file mode 100644
index 24811206b..000000000
--- a/go/metrics/go_runtime_test.go
+++ /dev/null
@@ -1,67 +0,0 @@
-/*
-   Copyright 2022 GitHub Inc.
-	 See https://github.com/github/gh-ost/blob/master/LICENSE
-*/
-
-package metrics
-
-import (
-	"context"
-	"runtime"
-	"testing"
-	"time"
-)
-
-type gaugeSpy struct {
-	names  []string
-	values []float64
-}
-
-func (g *gaugeSpy) Gauge(name string, value float64, _ ...string) {
-	g.names = append(g.names, name)
-	g.values = append(g.values, value)
-}
-
-func TestEmitGoRuntimeGauges(t *testing.T) {
-	spy := &gaugeSpy{}
-	m := &runtime.MemStats{
-		Alloc:        100,
-		Sys:          200,
-		HeapInuse:    300,
-		NumGC:        7,
-		PauseTotalNs: 42,
-	}
-	EmitGoRuntimeGauges(spy, m, 123)
-
-	wantNames := []string{
-		"go_runtime.alloc_bytes",
-		"go_runtime.sys_bytes",
-		"go_runtime.heap_inuse_bytes",
-		"go_runtime.num_gc",
-		"go_runtime.gc_pause_total_ns",
-		"go_runtime.goroutines",
-	}
-	wantVals := []float64{100, 200, 300, 7, 42, 123}
-
-	if len(spy.names) != len(wantNames) {
-		t.Fatalf("got %d gauges, want %d", len(spy.names), len(wantNames))
-	}
-	for i := range wantNames {
-		if spy.names[i] != wantNames[i] || spy.values[i] != wantVals[i] {
-			t.Fatalf("[%d] got %s=%v want %s=%v", i, spy.names[i], spy.values[i], wantNames[i], wantVals[i])
-		}
-	}
-}
-
-func TestEmitGoRuntimeGauges_nilSafe(t *testing.T) {
-	EmitGoRuntimeGauges(nil, &runtime.MemStats{}, 1)
-	EmitGoRuntimeGauges(&gaugeSpy{}, nil, 1)
-}
-
-func TestStartGoRuntimeReporter_stopsOnCancel(t *testing.T) {
-	ctx, cancel := context.WithCancel(context.Background())
-	c := &Client{} // sd nil — should not start
-	StartGoRuntimeReporter(ctx, c, time.Millisecond)
-	cancel()
-	time.Sleep(20 * time.Millisecond)
-}
diff --git a/go/metrics/progress.go b/go/metrics/progress.go
deleted file mode 100644
index b6b3361b6..000000000
--- a/go/metrics/progress.go
+++ /dev/null
@@ -1,17 +0,0 @@
-/*
-   Copyright 2022 GitHub Inc.
-	 See https://github.com/github/gh-ost/blob/master/LICENSE
-*/
-
-package metrics
-
-// EmitProgressGauges emits row-copy and DML progress gauges (namespace is applied by the client):
-// gh_ost.row_copy.rows_copied, gh_ost.row_copy.rows_estimate, gh_ost.dml.events_applied.
-func EmitProgressGauges(emit MemStatsGaugeEmitter, rowsCopied, rowsEstimate, dmlEventsApplied int64) {
-	if emit == nil {
-		return
-	}
-	emit.Gauge("row_copy.rows_copied", float64(rowsCopied))
-	emit.Gauge("row_copy.rows_estimate", float64(rowsEstimate))
-	emit.Gauge("dml.events_applied", float64(dmlEventsApplied))
-}
diff --git a/go/metrics/progress_test.go b/go/metrics/progress_test.go
deleted file mode 100644
index 4a6ce5d3b..000000000
--- a/go/metrics/progress_test.go
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
-   Copyright 2022 GitHub Inc.
-	 See https://github.com/github/gh-ost/blob/master/LICENSE
-*/
-
-package metrics
-
-import "testing"
-
-func TestEmitProgressGauges(t *testing.T) {
-	spy := &gaugeSpy{}
-	EmitProgressGauges(spy, 1000, 5000, 42)
-
-	wantNames := []string{
-		"row_copy.rows_copied",
-		"row_copy.rows_estimate",
-		"dml.events_applied",
-	}
-	wantVals := []float64{1000, 5000, 42}
-
-	if len(spy.names) != len(wantNames) {
-		t.Fatalf("got %d gauges, want %d", len(spy.names), len(wantNames))
-	}
-	for i := range wantNames {
-		if spy.names[i] != wantNames[i] || spy.values[i] != wantVals[i] {
-			t.Fatalf("[%d] got %s=%v want %s=%v", i, spy.names[i], spy.values[i], wantNames[i], wantVals[i])
-		}
-	}
-}
-
-func TestEmitProgressGauges_nilSafe(t *testing.T) {
-	EmitProgressGauges(nil, 1, 2, 3)
-}