diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md
index 12c3735..7ccc1d9 100644
--- a/docs/content/about/release-notes.md
+++ b/docs/content/about/release-notes.md
@@ -47,6 +47,7 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde
 * Changed the loader to send the `monitor-stopped` and `monitor-stop-timeout` events from its stop goroutine, so an event receiver that panics on either event crashes Blip. Previously, when the API stopped a monitor, `net/http` recovered the panic. Event receivers must not panic.
 * **Behavior change:** `monitor.Loader.Stop` now keeps the monitor stopped across `Load` and `StartMonitors` until `Start`, unless its config changes or it's removed and loaded again. Go callers that relied on the next reload to restart a stopped monitor must call `Start`. `POST /monitors/stop` keeps a monitor stopped across reloads, as before.
 * **Behavior change:** `POST /monitors/start` now calls the `StartMonitor` plugin, so it returns 409 for a monitor that the plugin declines. Previously the API started the monitor directly and skipped the plugin, so it could start a monitor that the plugin filtered out. Integrations that start filtered monitors through the API must allow them in the plugin.
+* A monitor stop, or a restart after a subsystem fails, no longer waits for the next collection tick, plan change check, or heartbeat. It cancels a collection, read-only check, or heartbeat write in progress, and drops metrics not yet sent to sinks.
 * **Behavior change:** Applied `timeout-connect` (default `10s`) to MySQL connections. Previously it was ignored, so connecting to an unreachable host waited for the operating system TCP connect timeout. Set it to `0` to retain the old behavior. An invalid `timeout-connect` value, which used to be ignored, now fails config validation, so Blip doesn't start, or a monitor in a reload doesn't load. For example, a bare number such as `timeout-connect: 5` has no unit and is invalid; use `5s`.
 * Added per-monitor fault counters, such as collector panics and timeouts, engine max runtime (EMR) timeouts, and dropped metrics, reported by the new [`GET /status/monitors/counters`]({{< ref "/api/status#get-statusmonitorscounters" >}}) API endpoint. Integrations can read them with the new `status.ReportCounters`, and register their own with `status.Counter`; the counter names are `status` constants like `status.COLLECTOR_PANICS`.
 * Added the optional `sink.CounterSink` interface for sinks that report monitor status counters; the monitor calls `RegisterCounters` each time it starts. A custom sink that wraps `sink.Retry` must forward `RegisterCounters` to it, else the retry buffer's `sink-retry-dropped` counter isn't reported.
diff --git a/docs/content/config/config-file.md b/docs/content/config/config-file.md
index d97e157..c96376b 100644
--- a/docs/content/config/config-file.md
+++ b/docs/content/config/config-file.md
@@ -194,7 +194,17 @@ A monitor that hasn't stopped by then keeps stopping in the background, and Blip
 Blip reports it again every 10 times the stop timeout until it stops.
 If the monitor is reloaded with the same ID, the new monitor doesn't start until the old one has stopped.
 
-A healthy monitor usually stops in about 1 second, but a slow collection or a heartbeat on a read-only server can take longer than the default stop timeout.
+A stop doesn't wait for the next collection, plan change check, or heartbeat:
+
+* The level collector stops right away. It cancels a collection in progress and drops its metrics, and it drops metrics that haven't been sent to the sinks yet (a send in progress isn't canceled). A canceled collection isn't counted as an engine max runtime timeout. The same happens when a monitor restarts because one of its subsystems failed.
+* The plan changer and the heartbeat writer stop right away while they wait. The plan changer waits between checks, and backs off while MySQL is offline (up to about 90 seconds: 60 seconds plus jitter). The heartbeat writer waits between heartbeats, before it retries a first heartbeat that failed, and while MySQL is read-only. A stop cancels the plan changer's read-only check and a heartbeat write in progress, so it doesn't wait for a hung or unreachable MySQL.
+
+A running monitor usually stops in milliseconds, even when MySQL is hung or unreachable.
+A stop can take longer in these cases:
+
+* A collector, or its cleanup function, doesn't return promptly when its collection is canceled. A stop waits for every collector to return, then runs their cleanup functions.
+* A plan change is in progress, and a collector's `Prepare` doesn't return promptly when it's canceled. A stop cancels the plan change and waits for it.
+* The monitor is starting. A stop waits for it to make its connections, which calls its credentials function (for example, for an IAM auth token, up to 500 milliseconds each time) or a custom `DbProviderFactory`, and to load its plans, which queries MySQL only if they're in a table ([`plans.table`](#table-1)): up to 5 seconds, connecting included, when MySQL is unreachable or hung.
 
 ---
 
diff --git a/heartbeat/writer.go b/heartbeat/writer.go
index 1c8f01d..c9965b4 100644
--- a/heartbeat/writer.go
+++ b/heartbeat/writer.go
@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/cashapp/blip/v2"
+	"github.com/cashapp/blip/v2/internal/stopchan"
 	"github.com/cashapp/blip/v2/sqlutil"
 	"github.com/cashapp/blip/v2/status"
 )
@@ -79,6 +80,11 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error {
 		cancel context.CancelFunc
 	)
 
+	// stopCtx is canceled on stop, so a stop doesn't wait for a write to a
+	// hung or unreachable MySQL (up to WriteTimeout), or for a sleep
+	stopCtx, stopCancel := stopchan.Context(stopChan)
+	defer stopCancel()
+
 	// First INSERT: either creates row if it doesn't exist for this monitor ID,
 	// or it updates an existing row with the current timestamp and frequency.
 	// This must be done else the simpler UPDATE statements below, which is the
@@ -95,9 +101,12 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error {
 	blip.Debug("%s: first heartbeat: %s", w.monitorId, ping)
 	for {
 		status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "first insert")
-		ctx, cancel = context.WithTimeout(context.Background(), WriteTimeout)
+		ctx, cancel = context.WithTimeout(stopCtx, WriteTimeout)
 		_, err = w.db.ExecContext(ctx, ping, params...)
 		cancel()
+		if stopCtx.Err() != nil {
+			return nil // stopped; a canceled write isn't an error
+		}
 		if err == nil { // success
 			status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "sleep")
 			break
@@ -105,19 +114,15 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error {
 
 		// Error --
 		blip.Debug("%s: first heartbeat failed: %s", w.monitorId, err)
+		wait := InitErrorWait
 		if sqlutil.ReadOnly(err) {
-			status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "init: MySQL is read-only, sleeping %s", ReadOnlyWait)
-			time.Sleep(ReadOnlyWait)
+			wait = ReadOnlyWait
+			status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "init: MySQL is read-only, sleeping %s", wait)
 		} else {
-			status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "init: error: %s (sleeping %s)", err, InitErrorWait)
-			time.Sleep(InitErrorWait)
+			status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "init: error: %s (sleeping %s)", err, wait)
 		}
-
-		// Was Stop called?
-		select {
-		case <-stopChan: // yes, return immediately
-			return nil
-		default: // no
+		if !stopchan.Sleep(stopCtx.Done(), wait) {
+			return nil // stopped
 		}
 	}
 
@@ -131,17 +136,24 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error {
 	ping = fmt.Sprintf("UPDATE %s SET ts=NOW(3) WHERE src_id=?", w.table)
 	blip.Debug("%s: heartbeat: %s", w.monitorId, ping)
 	for {
-		time.Sleep(w.freq)
+		if !stopchan.Sleep(stopCtx.Done(), w.freq) {
+			return nil // stopped
+		}
 
 		status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "write")
-		ctx, cancel = context.WithTimeout(context.Background(), WriteTimeout)
+		ctx, cancel = context.WithTimeout(stopCtx, WriteTimeout)
 		_, err = w.db.ExecContext(ctx, ping, w.srcId)
 		cancel()
+		if stopCtx.Err() != nil {
+			return nil // stopped; a canceled write isn't an error
+		}
 		if err != nil {
 			blip.Debug("%s: %s", w.monitorId, err.Error())
 			if sqlutil.ReadOnly(err) {
 				status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "MySQL is read-only, sleeping %s", ReadOnlyWait)
-				time.Sleep(ReadOnlyWait)
+				if !stopchan.Sleep(stopCtx.Done(), ReadOnlyWait) {
+					return nil // stopped
+				}
 			} else {
 				status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "write error: %s", err)
 				// No special sleep on random errors; keep trying to write at freq
@@ -151,12 +163,5 @@ func (w *Writer) Write(stopChan, doneChan chan struct{}) error {
 			// doesn't overwrite status set on Exec error; "sleep" = "write OK"
 			status.Monitor(w.monitorId, status.HEARTBEAT_WRITER, "sleep")
 		}
-
-		// Was Stop called?
-		select {
-		case <-stopChan: // yes, return immediately
-			return nil
-		default: // no
-		}
 	}
 }
diff --git a/heartbeat/writer_stop_test.go b/heartbeat/writer_stop_test.go
new file mode 100644
index 0000000..aea6778
--- /dev/null
+++ b/heartbeat/writer_stop_test.go
@@ -0,0 +1,168 @@
+// Copyright 2026 Block, Inc.
+
+package heartbeat_test
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"errors"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/go-sql-driver/mysql"
+
+	"github.com/cashapp/blip/v2"
+	"github.com/cashapp/blip/v2/heartbeat"
+	"github.com/cashapp/blip/v2/status"
+)
+
+// writeTestDriver is a fake SQL driver for the heartbeat writer: every write
+// succeeds, or fails with a MySQL read-only error if writeTestReadOnly is set,
+// or blocks until its context is canceled from write number writeTestHangAt on
+// (counted by writeTestWrites), if it's set.
+const writeTestDriver = "blip-heartbeat-write-test"
+
+var (
+	writeTestReadOnly atomic.Bool
+	writeTestHangAt   atomic.Int64
+	writeTestWrites   atomic.Int64
+)
+
+func init() {
+	sql.Register(writeTestDriver, writeTestSQLDriver{})
+}
+
+type writeTestSQLDriver struct{}
+
+func (writeTestSQLDriver) Open(string) (driver.Conn, error) { return writeTestConn{}, nil }
+
+type writeTestConn struct{}
+
+func (writeTestConn) Prepare(string) (driver.Stmt, error) {
+	return nil, errors.New("no prepared statements")
+}
+func (writeTestConn) Close() error              { return nil }
+func (writeTestConn) Begin() (driver.Tx, error) { return nil, errors.New("no transactions") }
+func (writeTestConn) ExecContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Result, error) {
+	n := writeTestWrites.Add(1)
+	if at := writeTestHangAt.Load(); at > 0 && n >= at {
+		<-ctx.Done()
+		return nil, ctx.Err()
+	}
+	if writeTestReadOnly.Load() {
+		return nil, &mysql.MySQLError{Number: 1290, Message: "read-only"}
+	}
+	return driver.RowsAffected(1), nil
+}
+
+// stopWriter starts w, waits until its status contains want, then stops it.
+// It returns how long Write took to return after stop.
+func stopWriter(t *testing.T, w *heartbeat.Writer, monitorId, want string) time.Duration {
+	t.Helper()
+	stopChan := make(chan struct{})
+	doneChan := make(chan struct{})
+	go w.Write(stopChan, doneChan)
+
+	deadline := time.Now().Add(5 * time.Second)
+	for !strings.Contains(status.ReportMonitors(monitorId)[monitorId][status.HEARTBEAT_WRITER], want) {
+		if time.Now().After(deadline) {
+			close(stopChan)
+			t.Fatalf("timeout waiting for heartbeat writer status %q", want)
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	// A stop isn't an error. The writer sets its final status, "stopped", when
+	// it returns, so check the status until then. This catches an error status
+	// only some of the time, so callers stop the writer several times.
+	t0 := time.Now()
+	close(stopChan)
+	timeout := time.After(time.Minute)
+	errStatus := ""
+	for {
+		if s := status.ReportMonitors(monitorId)[monitorId][status.HEARTBEAT_WRITER]; strings.Contains(s, "error") {
+			errStatus = s
+		}
+		select {
+		case <-doneChan:
+			if errStatus != "" {
+				t.Errorf("heartbeat writer status %q on stop, expected no error", errStatus)
+			}
+			return time.Since(t0)
+		case <-timeout:
+			t.Fatal("heartbeat writer did not stop")
+		default:
+		}
+	}
+}
+
+func TestWriterStopsDuringWaits(t *testing.T) {
+	// Stopping the writer must not wait for its sleep: the heartbeat
+	// frequency, or the read-only wait (20s by default).
+	const monitorId = "hb-write-stop"
+	defer status.RemoveMonitor(monitorId)
+	db, err := sql.Open(writeTestDriver, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+
+	const wait = 30 * time.Second
+	defer func(d time.Duration) { heartbeat.ReadOnlyWait = d }(heartbeat.ReadOnlyWait)
+	heartbeat.ReadOnlyWait = wait
+
+	// Read-only: the first write fails, and the writer sleeps ReadOnlyWait
+	writeTestReadOnly.Store(true)
+	defer writeTestReadOnly.Store(false)
+	w := heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "1s", Table: "heartbeat"})
+	if d := stopWriter(t, w, monitorId, "read-only"); d > wait/10 {
+		t.Errorf("stop during read-only wait took %s, expected well under %s", d, wait)
+	}
+
+	// Writable: the writer sleeps the heartbeat frequency between writes
+	writeTestReadOnly.Store(false)
+	w = heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: wait.String(), Table: "heartbeat"})
+	if d := stopWriter(t, w, monitorId, "sleep"); d > wait/10 {
+		t.Errorf("stop during heartbeat sleep took %s, expected well under %s", d, wait)
+	}
+}
+
+func TestWriterStopsDuringWrite(t *testing.T) {
+	// A write can wait up to WriteTimeout on a hung or partitioned MySQL.
+	// Stopping the writer cancels it instead of waiting.
+	const monitorId = "hb-write-stop-hung"
+	defer status.RemoveMonitor(monitorId)
+	db, err := sql.Open(writeTestDriver, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+
+	const timeout = 30 * time.Second
+	defer func(d time.Duration) { heartbeat.WriteTimeout = d }(heartbeat.WriteTimeout)
+	heartbeat.WriteTimeout = timeout
+
+	defer writeTestHangAt.Store(0)
+
+	for i := 0; i < 20; i++ {
+		// First write (INSERT) hangs. The status is set before the write, so the
+		// stop can come just before the write starts, which tests stopping it too.
+		writeTestWrites.Store(0)
+		writeTestHangAt.Store(1)
+		w := heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "1s", Table: "heartbeat"})
+		if d := stopWriter(t, w, monitorId, "first insert"); d > timeout/10 {
+			t.Errorf("stop during first write took %s, expected well under %s", d, timeout)
+		}
+
+		// The first write succeeds, and the first heartbeat (UPDATE) hangs
+		writeTestWrites.Store(0)
+		writeTestHangAt.Store(2)
+		w = heartbeat.NewWriter(monitorId, db, blip.ConfigHeartbeat{Freq: "10ms", Table: "heartbeat"})
+		if d := stopWriter(t, w, monitorId, "write"); d > timeout/10 {
+			t.Errorf("stop during heartbeat write took %s, expected well under %s", d, timeout)
+		}
+	}
+}
diff --git a/internal/stopchan/stopchan.go b/internal/stopchan/stopchan.go
new file mode 100644
index 0000000..c028bfa
--- /dev/null
+++ b/internal/stopchan/stopchan.go
@@ -0,0 +1,37 @@
+// Copyright 2026 Block, Inc.
+
+// Package stopchan has helpers for goroutines that stop when a stop channel is
+// closed, like the monitor subsystems. It's internal so that both the monitor
+// and heartbeat packages can use it without adding to Blip's public API.
+package stopchan
+
+import (
+	"context"
+	"time"
+)
+
+// Sleep waits for d or until stopChan is closed. It returns false if stopChan
+// was closed. To sleep until a context is done, pass ctx.Done() as stopChan.
+func Sleep(stopChan <-chan struct{}, d time.Duration) bool {
+	select {
+	case <-stopChan:
+		return false
+	case <-time.After(d): // an unfired timer is collected (Go 1.23+)
+		return true
+	}
+}
+
+// Context returns a context that's canceled when stopChan is closed, so a
+// query or other call that takes the context returns on stop. Call cancel when
+// done with the context to release its resources.
+func Context(stopChan <-chan struct{}) (ctx context.Context, cancel context.CancelFunc) {
+	ctx, cancel = context.WithCancel(context.Background())
+	go func() {
+		select {
+		case <-stopChan:
+			cancel()
+		case <-ctx.Done():
+		}
+	}()
+	return ctx, cancel
+}
diff --git a/internal/stopchan/stopchan_test.go b/internal/stopchan/stopchan_test.go
new file mode 100644
index 0000000..300e407
--- /dev/null
+++ b/internal/stopchan/stopchan_test.go
@@ -0,0 +1,47 @@
+// Copyright 2026 Block, Inc.
+
+package stopchan_test
+
+import (
+	"testing"
+	"time"
+
+	"github.com/cashapp/blip/v2/internal/stopchan"
+)
+
+func TestSleep(t *testing.T) {
+	stopChan := make(chan struct{})
+	if !stopchan.Sleep(stopChan, time.Millisecond) {
+		t.Error("Sleep returned false before stop")
+	}
+	close(stopChan)
+	t0 := time.Now()
+	if stopchan.Sleep(stopChan, time.Minute) {
+		t.Error("Sleep returned true after stop")
+	}
+	if d := time.Since(t0); d > time.Second {
+		t.Errorf("Sleep after stop took %s", d)
+	}
+}
+
+func TestContext(t *testing.T) {
+	stopChan := make(chan struct{})
+	ctx, cancel := stopchan.Context(stopChan)
+	defer cancel()
+	if ctx.Err() != nil {
+		t.Fatalf("context done before stop: %s", ctx.Err())
+	}
+	close(stopChan)
+	select {
+	case <-ctx.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("context not canceled on stop")
+	}
+
+	// cancel without stop
+	ctx, cancel = stopchan.Context(make(chan struct{}))
+	cancel()
+	if ctx.Err() == nil {
+		t.Error("context not canceled by cancel")
+	}
+}
diff --git a/monitor/engine.go b/monitor/engine.go
index cde473a..2b0c309 100644
--- a/monitor/engine.go
+++ b/monitor/engine.go
@@ -319,7 +319,8 @@ func (e *Engine) Prepare(ctx context.Context, plan blip.Plan, before, after func
 //   - []metrics is sorted ascending by Interval
 //
 // Collect returns when all collectors it starts return, or when emrCtx
-// (engine max runtime) expires. The former is the normal case.
+// (engine max runtime) expires or is canceled. The former is the normal case.
+// A canceled emrCtx isn't counted as an EMR timeout.
 //
 // Both metrics and an error can be returned in the case of partially success:
 // some collectors work but others fail. Caller should check returned metrics
@@ -444,7 +445,7 @@ SWEEP:
 	// are missing from this interval. Not an error event because, when the
 	// host is CPU-starved, this can happen every interval for every monitor.
 	notReturned := len(running)
-	if emrTimeout {
+	if emrTimeout && emrCtx.Err() == context.DeadlineExceeded { // not canceled (stopping)
 		e.counters.emrTimeouts.Add(1)
 		e.counters.domainsNotStarted.Add(uint64(notStarted))
 		e.counters.domainsNotReturned.Add(uint64(notReturned))
diff --git a/monitor/export_test.go b/monitor/export_test.go
new file mode 100644
index 0000000..0424536
--- /dev/null
+++ b/monitor/export_test.go
@@ -0,0 +1,25 @@
+// Copyright 2026 Block, Inc.
+
+package monitor
+
+import "sync/atomic"
+
+// StopAfterTicks makes every LCO Run return at tick n+1, before it collects at
+// that tick, by closing stopChan. Use it for a test that runs one LCO, and call
+// the returned func to remove it. Ticks are counted across all LCOs, and
+// stopChan is closed only once.
+func StopAfterTicks(n uint, stopChan chan struct{}) func() {
+	var ticks atomic.Uint64
+	tickerMux.Lock()
+	tickHook = func() {
+		if ticks.Add(1) == uint64(n)+1 {
+			close(stopChan)
+		}
+	}
+	tickerMux.Unlock()
+	return func() {
+		tickerMux.Lock()
+		tickHook = nil
+		tickerMux.Unlock()
+	}
+}
diff --git a/monitor/fault_counters_test.go b/monitor/fault_counters_test.go
index 00cb3b4..0416c49 100644
--- a/monitor/fault_counters_test.go
+++ b/monitor/fault_counters_test.go
@@ -219,12 +219,9 @@ func TestLevelCollectorCountsDroppedTicks(t *testing.T) {
 	c.emr = 2 * time.Second // longer than the slow collection, so the EMR doesn't expire
 	c.paused = false
 
-	// Run returns on the 7th tick: each tick reads one value, then stopChan is closed
-	stopChan := make(chan struct{}, 6)
-	for i := 0; i < 6; i++ {
-		stopChan <- struct{}{}
-	}
-	close(stopChan)
+	// Run returns on the 7th tick, after 6 ticks
+	stopChan := make(chan struct{})
+	defer StopAfterTicks(6, stopChan)()
 	doneChan := make(chan struct{})
 	go c.Run(stopChan, doneChan)
 	select {
@@ -260,7 +257,7 @@ func TestLevelCollectorCountsDroppedMetrics(t *testing.T) {
 	for i := 0; i < cap(c.metricsChan); i++ {
 		c.metricsChan <- nil
 	}
-	c.collect(1, level, time.Now())
+	c.collect(context.Background(), 1, level, time.Now())
 
 	if v := counters(t, monitorId)[status.LEVEL_METRICS_DROPPED]; v != 1 {
 		t.Errorf("%s = %d, expected 1", status.LEVEL_METRICS_DROPPED, v)
@@ -440,9 +437,10 @@ func TestLevelCollectorCountsTransformDrops(t *testing.T) {
 		Config:           blip.ConfigMonitor{MonitorId: monitorId},
 		TransformMetrics: func([]*blip.Metrics) error { return errors.New("transform failed") },
 	}, nil)
-	stopSinksChan := make(chan struct{})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
 	doneChan := make(chan struct{})
-	go c.recvMetrics(stopSinksChan, doneChan)
+	go c.recvMetrics(ctx, doneChan)
 	c.metricsChan <- []*blip.Metrics{{MonitorId: monitorId}}
 	c.metricsChan <- []*blip.Metrics{{MonitorId: monitorId}}
 
@@ -450,7 +448,7 @@ func TestLevelCollectorCountsTransformDrops(t *testing.T) {
 	for counters(t, monitorId)[status.LEVEL_TRANSFORM_DROPPED] < 2 && time.Now().Before(deadline) {
 		time.Sleep(5 * time.Millisecond)
 	}
-	close(stopSinksChan)
+	cancel()
 	<-doneChan
 	if v := counters(t, monitorId)[status.LEVEL_TRANSFORM_DROPPED]; v != 2 {
 		t.Errorf("%s = %d, expected 2", status.LEVEL_TRANSFORM_DROPPED, v)
diff --git a/monitor/level_collector.go b/monitor/level_collector.go
index 76d3aef..debb9c9 100644
--- a/monitor/level_collector.go
+++ b/monitor/level_collector.go
@@ -15,6 +15,7 @@ import (
 
 	"github.com/cashapp/blip/v2"
 	"github.com/cashapp/blip/v2/event"
+	"github.com/cashapp/blip/v2/internal/stopchan"
 	"github.com/cashapp/blip/v2/plan"
 	"github.com/cashapp/blip/v2/status"
 )
@@ -140,13 +141,23 @@ var (
 	tickerMux      = &sync.Mutex{}   // make go test -race happy
 	tickerDuration = 1 * time.Second // used for testing
 	timeElapsed    = 1 * time.Second // used for testing
+
+	// tickHook, if set, is called by Run after each tick. Tests use it to stop
+	// Run by closing stopChan, which Run sees before it collects at that tick.
+	tickHook func() // used for testing
 )
 
 // recvMetrics receives metrics on metricsChan and send them to all the sinks.
 // This is a goroutine run by keepRecvMetrics and restarted by keepRecvMetrics
-// if a sink (or the transformMetrics) panics. It stops when stopSinksChan is
-// closed in Run, and it closes doneChan when stopped.
-func (c *lco) recvMetrics(stopSinksChan, doneChan chan struct{}) {
+// if a sink (or the transformMetrics) panics. It stops when ctx is canceled
+// in Run, and it closes doneChan when stopped.
+//
+// When ctx is canceled, metrics not yet sent are dropped, so a stopped or
+// removed monitor doesn't send metrics. recvMetrics checks after it receives
+// metrics and before each sink Send. A Send in progress isn't canceled, but
+// after it returns, recvMetrics doesn't set a status, which would add back the
+// status of a monitor that was removed.
+func (c *lco) recvMetrics(ctx context.Context, doneChan chan struct{}) {
 	defer func() {
 		close(doneChan)
 		if r := recover(); r != nil {
@@ -155,14 +166,22 @@ func (c *lco) recvMetrics(stopSinksChan, doneChan chan struct{}) {
 			perr := fmt.Errorf("PANIC: sinks: %s: %v\n%s", c.monitorId, r, string(b[0:n]))
 			c.event.Error(event.LCO_RECEIVER_PANIC, perr.Error())
 		}
+		if ctx.Err() != nil {
+			// Clear "TransformMetrics" or "sending to"; don't set a status,
+			// which would add back the status of a monitor that was removed
+			status.RemoveComponent(c.monitorId, status.LEVEL_SINKS)
+		}
 	}()
 RECV:
-	for {
+	for ctx.Err() == nil {
 		status.Monitor(c.monitorId, status.LEVEL_SINKS, "idle")
 		select {
-		case <-stopSinksChan:
+		case <-ctx.Done():
 			return
 		case metrics := <-c.metricsChan:
+			if ctx.Err() != nil { // select picks randomly when both are ready
+				return
+			}
 			if c.transformMetrics != nil {
 				blip.Debug("%s: transform metrics", c.monitorId)
 				status.Monitor(c.monitorId, status.LEVEL_SINKS, "TransformMetrics")
@@ -175,11 +194,19 @@ RECV:
 			for _, m := range metrics {
 				coId := fmt.Sprintf("%s/%s/%d", m.Plan, m.Level, m.Interval)
 				for _, sink := range c.sinks {
+					if ctx.Err() != nil {
+						return
+					}
 					sinkName := sink.Name()
 					status.Monitor(c.monitorId, status.LEVEL_SINKS, "%s", coId+": sending to "+sinkName)
 					err := sink.Send(context.Background(), m) // @todo ctx with timeout
 					if err != nil {
 						c.event.Errorf(event.SINK_SEND_ERROR, "%s :%s", sinkName, err) // log by default
+					}
+					if ctx.Err() != nil { // stopped during Send
+						return
+					}
+					if err != nil {
 						status.Monitor(c.monitorId, "error:"+sinkName, "%s", err.Error())
 					} else {
 						status.RemoveComponent(c.monitorId, "error:"+sinkName)
@@ -192,72 +219,72 @@ RECV:
 
 // keepRecvMetrics keeps a recvMetrics goroutine running. If a sink or the
 // transformMetrics plugin panic, it must be restarted to keep metrics flowing.
-func (c *lco) keepRecvMetrics(stopSinksChan chan struct{}) {
-	for {
+func (c *lco) keepRecvMetrics(ctx context.Context) {
+	for ctx.Err() == nil {
 		doneChan := make(chan struct{})
-		go c.recvMetrics(stopSinksChan, doneChan)
-		select {
-		case <-stopSinksChan:
-			return
-		case <-doneChan:
-			// recvMetrics goroutine stopped but why? Did it return because
-			// stopSinksChan, or because of a panic?
-			select {
-			case <-stopSinksChan:
-				return // stopSinksChan closed
-			default:
-				// Probably a sink panic
-			}
-		}
+		go c.recvMetrics(ctx, doneChan)
+		<-doneChan // stopped, or restart after a sink panic
 	}
 }
 
 func (c *lco) Run(stopChan, doneChan chan struct{}) error {
 	defer close(doneChan)
 
+	// ctx is canceled on stop, so a stop doesn't wait for a collection in
+	// progress, and metrics not sent to sinks yet are dropped
+	ctx, cancel := stopchan.Context(stopChan)
+	defer cancel()
+
 	// Keep receiving metrics for as long as the LCO is running
-	stopSinksChan := make(chan struct{})
-	go c.keepRecvMetrics(stopSinksChan)
-	defer close(stopSinksChan)
+	go c.keepRecvMetrics(ctx)
 
 	// -----------------------------------------------------------------------
 	// LCO main loop: collect metrics every configured minimum level interval
 
 	status.Monitor(c.monitorId, status.LEVEL_COLLECTOR, "started at %s (paused until plan change)", blip.FormatTime(time.Now()))
 	tickerMux.Lock() // make go test -race happy
-	td, te := tickerDuration, timeElapsed
+	td, te, onTick := tickerDuration, timeElapsed, tickHook
 	tickerMux.Unlock()
 	s := -1 * te // -1 so first tick=0 and all levels collected
 	interval := uint(0)
 	ticker := time.NewTicker(td)
 	defer ticker.Stop()
-	var lastTick time.Time
-	for startTime := range ticker.C {
-		s = s + te
 
-		// Count ticks dropped while the previous iteration was blocked, usually
-		// by a slow collection. Run doesn't catch up on dropped ticks, so each one
-		// delays every level by one tick.
-		if n := droppedTicks(lastTick, startTime, td); n > 0 {
-			c.ticksDropped.Add(n)
+	var lastTick time.Time
+	for {
+		// Wait for the next tick or stop, so stopping doesn't wait for the tick
+		var startTime time.Time
+		select {
+		case <-stopChan:
+		case startTime = <-ticker.C:
+			if onTick != nil {
+				onTick()
+			}
 		}
-		lastTick = startTime
 
-		// Was monitor stopped?
+		// Was monitor stopped? Check after a tick, too, because select picks
+		// randomly when both the tick and stop are ready.
 		select {
-		case _, ok := <-stopChan: // yes, return immediately
+		case <-stopChan: // yes, return immediately
+			blip.Debug("stopChan closed s=%d interval=%d", s, interval)
+			cancel() // drop metrics not sent to sinks yet
 			// Stop changePlan goroutine (if any) and prevent new ones in the
 			// pathological case that the LCH calls ChangePlan while the LCO
 			// is terminating
-			if ok {
-				break
-			}
-			blip.Debug("stopChan closed at %s s=%d interval=%d", startTime, s, interval)
 			c.stopChangePlan()
 			c.engine.Stop() // stop all collectors and run their cleanup func
 			return nil
 		default: // no
 		}
+		s = s + te
+
+		// Count ticks dropped while the previous iteration was blocked, usually
+		// by a slow collection. Run doesn't catch up on dropped ticks, so each one
+		// delays every level by one tick.
+		if n := droppedTicks(lastTick, startTime, td); n > 0 {
+			c.ticksDropped.Add(n)
+		}
+		lastTick = startTime
 
 		// Paused because no plan is set or it's being changed?
 		c.stateMux.Lock() // -- LOCK --
@@ -282,11 +309,10 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error {
 
 		// Collect metrics at this level
 		interval += 1
-		c.collect(interval, c.levels[level].Name, startTime)
+		c.collect(ctx, interval, c.levels[level].Name, startTime)
 
 		c.stateMux.Unlock() // -- UNLOCK --
 	}
-	return nil
 }
 
 // stopChangePlan prevents new plan changes, cancels the current plan change,
@@ -306,14 +332,23 @@ func (c *lco) stopChangePlan() {
 	c.changePlanDoneChan = nil
 }
 
-func (c *lco) collect(interval uint, levelName string, startTime time.Time) {
+// collect collects metrics at one level and sends them to the sinks. If ctx
+// is canceled (the LCO is stopping), the collection is canceled, and its
+// metrics are dropped. Metrics that collect has queued for the sinks but
+// recvMetrics hasn't sent yet are dropped too.
+func (c *lco) collect(ctx context.Context, interval uint, levelName string, startTime time.Time) {
 	status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s: collecting", c.plan.Name, levelName)
+	queued := false // for the sinks; recvMetrics drops them if ctx is canceled
 	defer func() {
 		if err := recover(); err != nil { // catch panic in engine and TransformMetrics
 			b := make([]byte, 4096)
 			n := runtime.Stack(b, false)
 			c.event.Errorf(event.LCO_COLLECT_PANIC, "PANIC: %s: %s\n%s", c.monitorId, err, string(b[0:n]))
 		}
+		if !queued && ctx.Err() != nil {
+			status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s/%d: stopped; metrics dropped", c.plan.Name, levelName, interval)
+			return
+		}
 		status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s/%d: collected in %s", c.plan.Name, levelName, interval, time.Now().Sub(startTime))
 	}()
 
@@ -322,10 +357,13 @@ func (c *lco) collect(interval uint, levelName string, startTime time.Time) {
 	//
 	// Collect all metrics at this level. This is where metrics
 	// collection begins. Then Engine.Collect does the real work.
-	emrCtx, emrCancel := context.WithDeadline(context.Background(), startTime.Add(c.emr))
+	emrCtx, emrCancel := context.WithDeadline(ctx, startTime.Add(c.emr))
 	defer emrCancel()
 	metrics, err := c.engine.Collect(emrCtx, interval, levelName, startTime)
 	blip.Debug("%s: level %s: done in %s", c.monitorId, levelName, metrics[0].End.Sub(metrics[0].Begin))
+	if ctx.Err() != nil {
+		return // stopped: drop the metrics
+	}
 
 	if err != nil {
 		status.Monitor(c.monitorId, "error:collect", "%s", err.Error())
@@ -337,6 +375,7 @@ func (c *lco) collect(interval uint, levelName string, startTime time.Time) {
 	status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s: sending", c.plan.Name, levelName)
 	select {
 	case c.metricsChan <- metrics:
+		queued = true
 	default:
 		c.metricsDropped.Add(1)
 		c.event.Errorf(event.LCO_METRICS_FAULT, "metrics channel blocked (check for sink errors), dropping metrics: %s", metrics)
@@ -437,7 +476,7 @@ func (c *lco) changePlan(ctx context.Context, doneChan chan struct{}, newState,
 		errMsg := fmt.Sprintf("%s: error loading new plan %s: %s (retrying)", change, newPlanName, err)
 		status.Monitor(c.monitorId, status.LEVEL_CHANGE_PLAN, "%s", errMsg)
 		c.event.Sendf(event.CHANGE_PLAN_ERROR, "%s", errMsg)
-		if !waitForChangePlanRetry(ctx, 2*time.Second) {
+		if !stopchan.Sleep(ctx.Done(), 2*time.Second) {
 			blip.Debug("changePlan canceled while loading plan")
 			return
 		}
@@ -515,7 +554,7 @@ func (c *lco) changePlan(ctx context.Context, doneChan chan struct{}, newState,
 			return // changePlan goroutine has been cancelled
 		}
 		status.Monitor(c.monitorId, status.LEVEL_CHANGE_PLAN, "%s: error preparing new plan %s: %s (retrying)", change, newPlan.Name, err)
-		if !waitForChangePlanRetry(ctx, retry.NextBackOff()) {
+		if !stopchan.Sleep(ctx.Done(), retry.NextBackOff()) {
 			blip.Debug("changePlan canceled while waiting to retry plan preparation")
 			return
 		}
@@ -525,18 +564,6 @@ func (c *lco) changePlan(ctx context.Context, doneChan chan struct{}, newState,
 	c.event.Sendf(event.CHANGE_PLAN_SUCCESS, "%s", change)
 }
 
-func waitForChangePlanRetry(ctx context.Context, delay time.Duration) bool {
-	timer := time.NewTimer(delay)
-	defer timer.Stop()
-
-	select {
-	case <-ctx.Done():
-		return false
-	case <-timer.C:
-		return true
-	}
-}
-
 // Pause pauses metrics collection until ChangePlan is called. Run still runs,
 // but it doesn't collect when paused. The only way to resume after pausing is
 // to call ChangePlan again.
diff --git a/monitor/level_collector_stop_test.go b/monitor/level_collector_stop_test.go
new file mode 100644
index 0000000..f92bf25
--- /dev/null
+++ b/monitor/level_collector_stop_test.go
@@ -0,0 +1,280 @@
+// Copyright 2026 Block, Inc.
+
+package monitor
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/cashapp/blip/v2"
+	"github.com/cashapp/blip/v2/plan"
+	"github.com/cashapp/blip/v2/status"
+	"github.com/cashapp/blip/v2/test/mock"
+)
+
+// runStopLCO runs an LCO with one collector, calls ready to wait until it's
+// running, then stops it. It returns how long Run took to return after stop.
+func runStopLCO(t *testing.T, c *lco, ready func()) time.Duration {
+	t.Helper()
+	stopChan := make(chan struct{})
+	doneChan := make(chan struct{})
+	go c.Run(stopChan, doneChan)
+	ready()
+	t0 := time.Now()
+	close(stopChan)
+	select {
+	case <-doneChan:
+	case <-time.After(2 * time.Minute):
+		t.Fatal("LCO did not stop")
+	}
+	return time.Since(t0)
+}
+
+func newStopTestLCO(monitorId, level string, collector blip.Collector) *lco {
+	c := newLevelCollectorWithDBProvider(LevelCollectorArgs{Config: blip.ConfigMonitor{MonitorId: monitorId}}, nil)
+	c.engine = faultTestEngine(monitorId, level, collector)
+	c.plan = c.engine.plan
+	c.levels = []plan.SortedLevel{{Freq: time.Second, Name: level}}
+	c.emr = time.Second
+	c.paused = false
+	return c
+}
+
+func TestLevelCollectorStopDoesNotWaitForTick(t *testing.T) {
+	// Run waits for the next tick or stop, so a stop doesn't wait for the
+	// tick. Use a long tick: before, a stop right after Run started waited
+	// for the whole tick.
+	const monitorId = "lco-stop-tick"
+	defer status.RemoveMonitor(monitorId)
+	const tick = 10 * time.Second
+	TickerDuration(tick, time.Second)
+	defer TickerDuration(time.Second, time.Second)
+
+	c := newStopTestLCO(monitorId, "every", mock.MetricsCollector{DomainFunc: func() string { return "test.lco-stop-tick" }})
+	d := runStopLCO(t, c, func() {})
+	if d > tick/10 {
+		t.Errorf("stop took %s, expected well under the %s tick", d, tick)
+	}
+}
+
+func TestLevelCollectorStopCancelsCollection(t *testing.T) {
+	// A stop cancels a collection in progress instead of waiting for it to
+	// finish, which can take up to the engine max runtime (EMR). The canceled
+	// collection isn't counted as an EMR timeout.
+	const monitorId = "lco-stop-collect"
+	defer status.RemoveMonitor(monitorId)
+	TickerDuration(10*time.Millisecond, time.Second)
+	defer TickerDuration(time.Second, time.Second)
+
+	started := make(chan struct{}, 1)
+	slow := mock.MetricsCollector{
+		DomainFunc: func() string { return "test.lco-stop-collect" },
+		CollectFunc: func(ctx context.Context, _ string) ([]blip.MetricValue, error) {
+			select {
+			case started <- struct{}{}:
+			default:
+			}
+			<-ctx.Done() // until canceled or the collector max runtime
+			return nil, ctx.Err()
+		},
+	}
+	const maxRuntime = time.Minute
+	c := newStopTestLCO(monitorId, "every", slow)
+	c.emr = maxRuntime
+	for _, cl := range c.engine.collectors {
+		cl.cmr = maxRuntime
+	}
+
+	d := runStopLCO(t, c, func() {
+		select {
+		case <-started:
+		case <-time.After(5 * time.Second):
+			t.Fatal("timeout waiting for collection to start")
+		}
+	})
+	if d > maxRuntime/10 {
+		t.Errorf("stop took %s, expected well under the %s max runtime", d, maxRuntime)
+	}
+	if v := counters(t, monitorId)[status.ENGINE_EMR_TIMEOUTS]; v != 0 {
+		t.Errorf("%s = %d, expected 0", status.ENGINE_EMR_TIMEOUTS, v)
+	}
+	// The canceled collection's metrics are dropped, not reported collected
+	if s := status.ReportMonitors(monitorId)[monitorId][status.LEVEL_COLLECT]; !strings.HasSuffix(s, ": stopped; metrics dropped") {
+		t.Errorf("%s status = %q, expected stopped; metrics dropped", status.LEVEL_COLLECT, s)
+	}
+}
+
+func TestLevelCollectorStopDropsQueuedMetrics(t *testing.T) {
+	// Metrics from a complete collection that haven't been sent to the sinks
+	// when the LCO stops are dropped, not sent after the monitor stopped.
+	// Block TransformMetrics so that metrics are received and queued, then
+	// stop the LCO, then let TransformMetrics return.
+	const monitorId = "lco-stop-queued"
+	defer status.RemoveMonitor(monitorId)
+	TickerDuration(10*time.Millisecond, time.Second)
+	defer TickerDuration(time.Second, time.Second)
+
+	entered := make(chan struct{})
+	release := make(chan struct{})
+	var enteredOnce, releaseOnce sync.Once
+	defer releaseOnce.Do(func() { close(release) })
+	var sent atomic.Int32
+	c := newStopTestLCO(monitorId, "every", mock.MetricsCollector{DomainFunc: func() string { return "test.lco-stop-queued" }})
+	c.transformMetrics = func([]*blip.Metrics) error {
+		enteredOnce.Do(func() { close(entered) })
+		<-release
+		return nil
+	}
+	c.sinks = []blip.Sink{mock.Sink{SendFunc: func(context.Context, *blip.Metrics) error {
+		sent.Add(1)
+		return nil
+	}}}
+
+	runStopLCO(t, c, func() {
+		select {
+		case <-entered:
+		case <-time.After(5 * time.Second):
+			t.Fatal("timeout waiting for TransformMetrics")
+		}
+		time.Sleep(50 * time.Millisecond) // queue more metrics
+	})
+	releaseOnce.Do(func() { close(release) })
+	time.Sleep(100 * time.Millisecond)
+	if n := sent.Load(); n != 0 {
+		t.Errorf("sinks got %d sends after stop, expected 0", n)
+	}
+	// The sinks status isn't left at "TransformMetrics"
+	if s, ok := status.ReportMonitors(monitorId)[monitorId][status.LEVEL_SINKS]; ok {
+		t.Errorf("%s status = %q, expected none after the metrics were dropped", status.LEVEL_SINKS, s)
+	}
+}
+
+func TestLevelCollectorStopDropsMetricsAfterSend(t *testing.T) {
+	// After a sink Send returns, recvMetrics can receive metrics that were
+	// queued during the Send even though the LCO has stopped, because select
+	// picks randomly when both are ready. It must drop them before it calls
+	// TransformMetrics. Block the first Send so that metrics queue, stop the
+	// LCO, then let the Send return. One run catches a missing check half the
+	// time, so run several.
+	const monitorId = "lco-stop-after-send"
+	defer status.RemoveMonitor(monitorId)
+	TickerDuration(10*time.Millisecond, time.Second)
+	defer TickerDuration(time.Second, time.Second)
+
+	for i := 0; i < 20; i++ {
+		entered := make(chan struct{})
+		release := make(chan struct{})
+		var enteredOnce, releaseOnce sync.Once
+		var transforms atomic.Int32
+		c := newStopTestLCO(monitorId, "every", mock.MetricsCollector{DomainFunc: func() string { return "test.lco-stop-after-send" }})
+		c.transformMetrics = func([]*blip.Metrics) error {
+			transforms.Add(1)
+			return nil
+		}
+		c.sinks = []blip.Sink{mock.Sink{SendFunc: func(context.Context, *blip.Metrics) error {
+			enteredOnce.Do(func() {
+				close(entered)
+				<-release
+			})
+			return nil
+		}}}
+
+		runStopLCO(t, c, func() {
+			select {
+			case <-entered:
+			case <-time.After(5 * time.Second):
+				t.Fatal("timeout waiting for sink Send")
+			}
+			time.Sleep(30 * time.Millisecond) // queue more metrics
+		})
+		before := transforms.Load()
+		releaseOnce.Do(func() { close(release) })
+		time.Sleep(30 * time.Millisecond)
+		if n := transforms.Load() - before; n != 0 {
+			t.Fatalf("run %d: TransformMetrics called %d times after stop, expected 0", i, n)
+		}
+	}
+}
+
+func TestLevelCollectorStopDuringSendDoesNotAddStatus(t *testing.T) {
+	// A sink Send in progress when the LCO stops can return after the Loader
+	// has removed the monitor's status. recvMetrics must not set a status
+	// then, which would add back the status of a removed monitor. Block the
+	// first Send, stop the LCO, remove the status, then let the Send return.
+	for _, sendErr := range []error{errors.New("send failed"), nil} {
+		t.Run(fmt.Sprintf("err=%v", sendErr), func(t *testing.T) {
+			const monitorId = "lco-stop-during-send"
+			defer status.RemoveMonitor(monitorId)
+			TickerDuration(10*time.Millisecond, time.Second)
+			defer TickerDuration(time.Second, time.Second)
+
+			entered := make(chan struct{})
+			release := make(chan struct{})
+			returned := make(chan struct{})
+			var once sync.Once
+			c := newStopTestLCO(monitorId, "every", mock.MetricsCollector{DomainFunc: func() string { return "test.lco-stop-during-send" }})
+			c.sinks = []blip.Sink{mock.Sink{SendFunc: func(context.Context, *blip.Metrics) error {
+				err := errors.New("not first send")
+				once.Do(func() {
+					close(entered)
+					<-release
+					defer close(returned)
+					err = sendErr
+				})
+				return err
+			}}}
+
+			runStopLCO(t, c, func() {
+				select {
+				case <-entered:
+				case <-time.After(5 * time.Second):
+					t.Fatal("timeout waiting for sink Send")
+				}
+			})
+			status.RemoveMonitor(monitorId)
+			close(release)
+			<-returned
+			time.Sleep(50 * time.Millisecond) // let recvMetrics return
+			if s, ok := status.ReportMonitors(monitorId)[monitorId]; ok {
+				t.Errorf("status of removed monitor added back: %v", s)
+			}
+		})
+	}
+}
+
+// cancelAfterFirstCheck is a context that reports canceled after the first
+// Err call. In collect, that's the check right after Engine.Collect, so it's
+// as if the LCO stopped after collect queued the metrics.
+type cancelAfterFirstCheck struct {
+	context.Context
+	checks atomic.Int32
+}
+
+func (c *cancelAfterFirstCheck) Err() error {
+	if c.checks.Add(1) > 1 {
+		return context.Canceled
+	}
+	return nil
+}
+
+func TestLevelCollectorStopAfterQueueReportsCollected(t *testing.T) {
+	// If the LCO stops after collect queued the metrics, collect reports
+	// them collected, not dropped: recvMetrics drops them, if it hasn't sent
+	// them yet, and clears the sinks status.
+	const monitorId = "lco-stop-after-queue"
+	defer status.RemoveMonitor(monitorId)
+	c := newStopTestLCO(monitorId, "every", mock.MetricsCollector{DomainFunc: func() string { return "test.lco-stop-after-queue" }})
+	c.collect(&cancelAfterFirstCheck{Context: context.Background()}, 1, "every", time.Now())
+	if n := len(c.metricsChan); n != 1 {
+		t.Fatalf("%d metrics queued, expected 1", n)
+	}
+	if s := status.ReportMonitors(monitorId)[monitorId][status.LEVEL_COLLECT]; !strings.Contains(s, ": collected in ") {
+		t.Errorf("%s status = %q, expected collected in", status.LEVEL_COLLECT, s)
+	}
+}
diff --git a/monitor/level_collector_test.go b/monitor/level_collector_test.go
index 8bdcbf7..11ae565 100644
--- a/monitor/level_collector_test.go
+++ b/monitor/level_collector_test.go
@@ -567,19 +567,16 @@ func TestLevelCollector_RGB_DomainPrioirty(t *testing.T) {
 	monitor.TickerDuration(100*time.Millisecond, 100*time.Millisecond)
 	defer monitor.TickerDuration(time.Second, time.Second)
 
-	// Create LCO and prime its stopChan for 5 iterations
+	// Create LCO and stop it after 5 ticks
 	lco := monitor.NewLevelCollector(monitor.LevelCollectorArgs{
 		Config:     moncfg,
 		DB:         db,
 		PlanLoader: pl,
 		Sinks:      []blip.Sink{},
 	})
-	stopChan := make(chan struct{}, 5)
+	stopChan := make(chan struct{})
 	doneChan := make(chan struct{})
-	for i := 0; i < 5; i++ {
-		stopChan <- struct{}{}
-	}
-	close(stopChan)
+	defer monitor.StopAfterTicks(5, stopChan)()
 
 	// readyChan is closed by a fake event receiver when lco.ChangePlan sends
 	// event.CHANGE_PLAN_SUCCESS, which means the plan is loaded and ready.
@@ -692,12 +689,9 @@ func TestLevelCollector_RGB_SlowBlue(t *testing.T) {
 		Sinks:            []blip.Sink{},
 		TransformMetrics: xf,
 	})
-	stopChan := make(chan struct{}, 4)
+	stopChan := make(chan struct{})
 	doneChan := make(chan struct{})
-	for i := 0; i < 4; i++ {
-		stopChan <- struct{}{}
-	}
-	close(stopChan)
+	defer monitor.StopAfterTicks(4, stopChan)()
 
 	readyChan := planSet(nil)
 	defer event.RemoveSubscribers()
@@ -854,12 +848,9 @@ func TestLevelCollector_RGB_ProgressiveBlue(t *testing.T) {
 		Sinks:            []blip.Sink{},
 		TransformMetrics: xf,
 	})
-	stopChan := make(chan struct{}, 5)
+	stopChan := make(chan struct{})
 	doneChan := make(chan struct{})
-	for i := 0; i < 5; i++ {
-		stopChan <- struct{}{}
-	}
-	close(stopChan)
+	defer monitor.StopAfterTicks(5, stopChan)()
 
 	readyChan := planSet(nil)
 	defer event.RemoveSubscribers()
@@ -1002,12 +993,9 @@ func TestLevelCollector_RGB_Fault(t *testing.T) {
 		Sinks:            []blip.Sink{},
 		TransformMetrics: xf,
 	})
-	stopChan := make(chan struct{}, 3)
+	stopChan := make(chan struct{})
 	doneChan := make(chan struct{})
-	for i := 0; i < 3; i++ {
-		stopChan <- struct{}{}
-	}
-	close(stopChan)
+	defer monitor.StopAfterTicks(3, stopChan)()
 
 	// Record events to check that 2 event.COLLECTOR_FAULT events are sent
 	events := []string{}
diff --git a/monitor/monitor.go b/monitor/monitor.go
index 11dd544..eb7ef50 100644
--- a/monitor/monitor.go
+++ b/monitor/monitor.go
@@ -21,6 +21,7 @@ import (
 	"github.com/cashapp/blip/v2/event"
 	"github.com/cashapp/blip/v2/ha"
 	"github.com/cashapp/blip/v2/heartbeat"
+	"github.com/cashapp/blip/v2/internal/stopchan"
 	"github.com/cashapp/blip/v2/plan"
 	"github.com/cashapp/blip/v2/prom"
 	"github.com/cashapp/blip/v2/sink"
@@ -236,7 +237,7 @@ func (m *Monitor) runLoop(stopChan, doneChan chan struct{}) {
 
 		m.setErr(err, false)
 		if err != nil {
-			if !sleep(stopChan, m.retry.NextBackOff()) {
+			if !stopchan.Sleep(stopChan, m.retry.NextBackOff()) {
 				return
 			}
 			continue
@@ -245,8 +246,9 @@ func (m *Monitor) runLoop(stopChan, doneChan chan struct{}) {
 		// Monitor is running. Wait for either Stop (which closes m.runLoopChan)
 		// or one of the monitor subsystems to return/panic (which closes m.runChan).
 		// On Stop, return immediately: user is stopping the monitor completely.
-		// On m.runChan close (via stop func), we restart almost immediately because
-		// Blip never stops trying to send metrics.
+		// On m.runChan close (via stop func), we restart almost immediately to
+		// resume collecting metrics. Like a stop, the restart drops the collection
+		// in progress and metrics not yet sent to sinks (see lco.Run).
 		m.retry.Reset()
 		status.Monitor(m.monitorId, status.MONITOR, "running since %s", blip.FormatTime(time.Now()))
 		select {
@@ -254,26 +256,13 @@ func (m *Monitor) runLoop(stopChan, doneChan chan struct{}) {
 			return
 		case <-runChan: // internal failure
 			blip.Debug("%s: runChan closed; restarting", m.monitorId)
-			if !sleep(stopChan, 1*time.Second) { // between monitor restarts
+			if !stopchan.Sleep(stopChan, 1*time.Second) { // between monitor restarts
 				return
 			}
 		}
 	}
 }
 
-// sleep waits for d or until stopChan is closed. It returns false if stopChan
-// was closed.
-func sleep(stopChan chan struct{}, d time.Duration) bool {
-	t := time.NewTimer(d)
-	defer t.Stop()
-	select {
-	case <-stopChan:
-		return false
-	case <-t.C:
-		return true
-	}
-}
-
 // startup starts the monitor subsystems, which are optional depending on config:
 // heartbeat writer, exporter API (Prometheus emulation), PCH, and LCO.
 // The monitor is running once these have started. If any subsystem crashes
@@ -321,7 +310,7 @@ func (m *Monitor) startup(stopChan chan struct{}) (err error) {
 			break
 		}
 		status.Monitor(m.monitorId, status.MONITOR, "error making DB/DSN, sleep and retry: %s", err)
-		if !sleep(stopChan, m.retry.NextBackOff()) {
+		if !stopchan.Sleep(stopChan, m.retry.NextBackOff()) {
 			return nil // runLoop stopped (Stop called)
 		}
 	}
@@ -337,7 +326,7 @@ func (m *Monitor) startup(stopChan chan struct{}) (err error) {
 			break
 		}
 		status.Monitor(m.monitorId, status.MONITOR, "error loading plans, sleep and retry: %s", err)
-		if !sleep(stopChan, m.retry.NextBackOff()) {
+		if !stopchan.Sleep(stopChan, m.retry.NextBackOff()) {
 			return nil // runLoop stopped (Stop called)
 		}
 	}
diff --git a/monitor/plan_changer.go b/monitor/plan_changer.go
index 9d3b9c4..655af9f 100644
--- a/monitor/plan_changer.go
+++ b/monitor/plan_changer.go
@@ -13,6 +13,7 @@ import (
 	"github.com/cashapp/blip/v2"
 	"github.com/cashapp/blip/v2/event"
 	"github.com/cashapp/blip/v2/ha"
+	"github.com/cashapp/blip/v2/internal/stopchan"
 	"github.com/cashapp/blip/v2/status"
 )
 
@@ -121,39 +122,51 @@ func (pch *planChanger) setErr(err error) {
 	}
 }
 
-// Run calls CheckState every second; or, if offline, it uses an exponential
-// backoff up to 60 seconds until back online (MySQL connection ok). There is
-// no logic in this function; it's just a timed loop to call CheckState. It's
-// run as a goroutine from Monitor.Run only if config.monitors.plans.adjust
-// is enabled (blip.ConfigPlanChange.Enabled returns true).
+// Run calls checkState every second; or, if offline, it uses an exponential
+// backoff up to about 90 seconds (60 seconds plus jitter) until back online
+// (MySQL connection ok). There is no logic in this function; it's just a timed
+// loop to call checkState. It's run as a goroutine from Monitor.Run only if
+// config.monitors.plans.adjust is enabled (blip.ConfigPlanChange.Enabled
+// returns true).
 func (pch *planChanger) Run(stopChan, doneChan chan struct{}) error {
 	defer close(doneChan)
 	defer status.Monitor(pch.monitorId, status.PLAN_CHANGER, "not running")
 
 	status.Monitor(pch.monitorId, status.PLAN_CHANGER, "running")
 
-	for {
-		select {
-		case <-stopChan:
-			return nil
-		default:
-		}
+	// ctx is canceled on stop, so a stop doesn't wait for the read-only query
+	// to a hung or unreachable MySQL, or for the sleep between checks
+	ctx, cancel := stopchan.Context(stopChan)
+	defer cancel()
 
-		pch.CheckState()
+	for {
+		pch.checkState(ctx)
 
-		if pch.curr.state != blip.STATE_OFFLINE {
-			time.Sleep(1 * time.Second)
-		} else {
-			time.Sleep(pch.retry.NextBackOff())
+		wait := 1 * time.Second
+		if pch.curr.state == blip.STATE_OFFLINE {
+			wait = pch.retry.NextBackOff()
+		}
+		if !stopchan.Sleep(ctx.Done(), wait) {
+			return nil // stopped
 		}
 	}
 }
 
 // CheckState checks the current state and changes state when necessary.
-// This is the main logic, called periodically by Run.
+// It's the same check that Run does periodically, but it can't be canceled.
 func (pch *planChanger) CheckState() {
+	pch.checkState(context.Background())
+}
+
+// checkState is the main logic of CheckState, called periodically by Run. It
+// does nothing if ctx is canceled (the PCH is stopping) while it checks the
+// state.
+func (pch *planChanger) checkState(ctx context.Context) {
 	now := Now()
-	obsv := pch.state()
+	obsv := pch.state(ctx)
+	if ctx.Err() != nil {
+		return // stopped
+	}
 
 	pch.Lock()
 	defer pch.Unlock()
@@ -237,8 +250,9 @@ func (pch *planChanger) lcoChangePlan(state, planName string) error {
 
 const readOnlyQuery = "SELECT @@read_only, @@super_read_only"
 
-// state queries MySQL to ascertain the HA and read-only state.
-func (pch *planChanger) state() string {
+// state queries MySQL to ascertain the HA and read-only state. If ctx is
+// canceled, it returns STATE_NONE, which the caller must ignore.
+func (pch *planChanger) state(ctx context.Context) string {
 	status.Monitor(pch.monitorId, status.PLAN_CHANGER, "checking HA standby")
 	if pch.ha.Standby() {
 		return blip.STATE_STANDBY
@@ -248,9 +262,12 @@ func (pch *planChanger) state() string {
 	status.Monitor(pch.monitorId, status.PLAN_CHANGER, "checking MySQL read-only")
 
 	var ro, sro int
-	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
-	err := pch.db.QueryRowContext(ctx, readOnlyQuery).Scan(&ro, &sro)
+	qctx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	err := pch.db.QueryRowContext(qctx, readOnlyQuery).Scan(&ro, &sro)
 	cancel()
+	if ctx.Err() != nil {
+		return blip.STATE_NONE // stopped
+	}
 	pch.setErr(err)
 	if err != nil {
 		blip.Debug(err.Error())
diff --git a/monitor/plan_changer_stop_test.go b/monitor/plan_changer_stop_test.go
new file mode 100644
index 0000000..778fde6
--- /dev/null
+++ b/monitor/plan_changer_stop_test.go
@@ -0,0 +1,119 @@
+// Copyright 2026 Block, Inc.
+
+package monitor
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/cashapp/blip/v2/ha"
+	"github.com/cashapp/blip/v2/status"
+)
+
+// offlineTestDriver is a fake SQL driver that can't connect, so the PCH sees
+// MySQL offline.
+const offlineTestDriver = "blip-monitor-offline-test"
+
+// hungTestDriver is a fake SQL driver whose queries block until their context
+// is canceled, like a hung or partitioned MySQL.
+const hungTestDriver = "blip-monitor-hung-test"
+
+func init() {
+	sql.Register(offlineTestDriver, offlineTestSQLDriver{})
+	sql.Register(hungTestDriver, hungTestSQLDriver{})
+}
+
+type hungTestSQLDriver struct{}
+
+func (hungTestSQLDriver) Open(string) (driver.Conn, error) { return hungTestConn{}, nil }
+
+type hungTestConn struct{}
+
+func (hungTestConn) Prepare(string) (driver.Stmt, error) {
+	return nil, errors.New("no prepared statements")
+}
+func (hungTestConn) Close() error              { return nil }
+func (hungTestConn) Begin() (driver.Tx, error) { return nil, errors.New("no transactions") }
+func (hungTestConn) QueryContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Rows, error) {
+	hungTestQueries <- struct{}{}
+	<-ctx.Done()
+	return nil, ctx.Err()
+}
+
+var hungTestQueries = make(chan struct{}, 10) // a query has started
+
+type offlineTestSQLDriver struct{}
+
+func (offlineTestSQLDriver) Open(string) (driver.Conn, error) { return nil, errors.New("offline") }
+
+// stopPCH runs a PCH on a database opened with driverName, waits until ready
+// returns true, then stops the PCH and fails the test if the stop takes 1s or
+// longer. While MySQL is offline, the PCH waits 1 minute between checks.
+func stopPCH(t *testing.T, monitorId, driverName string, ready func() bool) {
+	t.Helper()
+	db, err := sql.Open(driverName, "")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	pch := NewPlanChanger(PlanChangerArgs{MonitorId: monitorId, DB: db, HA: ha.Disabled})
+	pch.retry.InitialInterval = time.Minute
+	pch.retry.MaxInterval = time.Minute
+	pch.retry.Reset()
+
+	stopChan := make(chan struct{})
+	doneChan := make(chan struct{})
+	go pch.Run(stopChan, doneChan)
+	deadline := time.Now().Add(5 * time.Second)
+	for !ready() {
+		if time.Now().After(deadline) {
+			close(stopChan)
+			t.Fatal("timeout waiting for the PCH")
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+
+	t0 := time.Now()
+	close(stopChan)
+	select {
+	case <-doneChan:
+	case <-time.After(time.Minute):
+		t.Fatal("PCH did not stop")
+	}
+	if d := time.Since(t0); d >= time.Second {
+		t.Errorf("stop took %s, expected less than 1s", d)
+	}
+}
+
+func TestPlanChangerStopsDuringOfflineWait(t *testing.T) {
+	// While MySQL is offline, the PCH waits its retry backoff (up to about 90s:
+	// 60s plus jitter) between state checks. Stopping it must not wait for that.
+	const monitorId = "pch-stop"
+	defer status.RemoveMonitor(monitorId)
+	stopPCH(t, monitorId, offlineTestDriver, func() bool {
+		return status.ReportMonitors(monitorId)[monitorId]["error:"+status.PLAN_CHANGER] != ""
+	})
+}
+
+func TestPlanChangerStopsDuringStateQuery(t *testing.T) {
+	// The read-only query can wait up to 10s on a hung or partitioned MySQL.
+	// Stopping the PCH cancels it instead of waiting.
+	const monitorId = "pch-stop-query"
+	defer status.RemoveMonitor(monitorId)
+	stopPCH(t, monitorId, hungTestDriver, func() bool {
+		select {
+		case <-hungTestQueries:
+			return true
+		default:
+			return false
+		}
+	})
+	// A canceled query isn't an error: the PCH didn't see MySQL offline
+	if e := status.ReportMonitors(monitorId)[monitorId]["error:"+status.PLAN_CHANGER]; e != "" {
+		t.Errorf("PCH error status %q after stop, expected none", e)
+	}
+}
diff --git a/server/api_test.go b/server/api_test.go
index fafab56..81c5f88 100644
--- a/server/api_test.go
+++ b/server/api_test.go
@@ -123,6 +123,7 @@ func TestAPIMonitorsStart(t *testing.T) {
 	// running or the StartMonitor plugin declines to start it. A monitor
 	// stopped by POST /monitors/stop stays stopped across a reload.
 	cfg := blip.DefaultConfig()
+	cfg.MonitorLoader.StopTimeout = "200ms"
 	dbf := blockingDbFactory{
 		DbFactory: dbconn.NewConnFactory(nil, nil),
 		monitorId: "api-pending",
@@ -231,7 +232,7 @@ func TestAPIMonitorsStart(t *testing.T) {
 	}
 
 	// Start api-pending and stop it while it's stuck in startup (making its
-	// DB), so the stop times out after the default stop timeout (3s). Starting
+	// DB), so the stop times out after the stop timeout (200ms). Starting
 	// it while it's still stopping is pending, and starting it again is still
 	// pending, not already running.
 	if got := post("/monitors/start?id=api-pending"); got != http.StatusOK {