Set up repl.lag once and apply options at every level

collector.go

@@ -21,11 +21,18 @@ type Collector interface {
 
 	// Prepare prepares a plan for future calls to Collect. The return function
 	// is called once when the collector is destroyed; it allows the collector
-	// to clean up. It's also called if Prepare returns an error with a non-nil
-	// function, and if a later collector in the plan fails to prepare, because
-	// then the engine discards the collectors it already prepared. If Prepare
-	// returns an error, Blip will retry preparing the plan. Therefore, Prepare
-	// should not retry on error (for example, if the database is not online yet).
+	// to clean up. It's also called if a later collector in the plan fails to
+	// prepare, because then the engine discards the collectors it already
+	// prepared. If Prepare returns an error, Blip will retry preparing the
+	// plan. Therefore, Prepare should not retry on error (for example, if the
+	// database is not online yet).
+	//
+	// If Prepare returns an error, Blip discards the collector without calling
+	// Collect, so Prepare must release any resources that it started, like
+	// goroutines, before it returns, and it should return a nil cleanup func.
+	// As a backstop, Blip calls a non-nil cleanup func returned with an error,
+	// so the cleanup func must be safe to call after Prepare released the
+	// resources.
 	Prepare(ctx context.Context, plan Plan) (func(), error)
 
 	// Collect collects metrics for the previously prepared plan. Collect is only

docs/content/about/release-notes.md

@@ -61,6 +61,10 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde
 * Added validation of the retry options: `buffer-size` must be from 1 to 10000, `send-timeout` must be greater than zero and at most 1m, and `send-retry-wait` must be greater than zero. Other values prevent the monitor from loading. Previously a built-in sink rejected any retry option, so no config that loaded before is affected.
 * Changed the retry sink to stop waiting `send-retry-wait` between sends when `send-timeout` expires. Previously a send cycle could run past `send-timeout` by up to one `send-retry-wait`.
 * Added `sink.RetryArgs.Validate` and the `sink.MAX_RETRY_BUFFER_SIZE` and `sink.MAX_RETRY_SEND_TIMEOUT` constants. `sink.NewRetry` doesn't call `Validate`, so custom sinks that build `RetryArgs` from user input should call it first.
+* Fixed `repl.lag` at multiple levels with the Blip heartbeat writer: `report-no-heartbeat` now applies at every level, not only at one random level. Levels that don't set `writer`, or set it to `auto`, use the writer set at another level. Previously each of these levels auto-detected its own writer, so it could use `pfs` while another level used `blip`. The writer options `repl-check`, `table`, `source-id`, `source-role`, and `network-latency` are read from one level: the first level, by name, that sets `writer` to `pfs` or `blip`, else the first level that collects `repl.lag`.
+* **Behavior change:** The `repl.lag` heartbeat reader now uses `repl-check`; previously it ignored it at every level. With the Blip heartbeat writer and `repl-check` set, an instance that isn't a replica now drops `repl.lag.current`, or reports -1 with `report-not-a-replica=yes`, instead of reporting heartbeat lag.
+* **Behavior change:** Plans that set the `repl.lag` `writer` to `pfs` at one level and `blip` at another now fail to prepare, and the monitor keeps retrying the plan. Previously each level used its own writer. Set the same `writer` at every level, or set it at one level only.
+* Changed `heartbeat.BlipReader.Stop` to wait for the reader to return, which takes up to the heartbeat read timeout (2s) if a heartbeat query is running. Previously the reader could report status after its monitor was stopped and removed, so the removed monitor reappeared in [`GET /status/monitors`]({{< ref "/api/status#get-statusmonitors" >}}).
 
 ### v2.1.0 (10 Aug 2026)
 

docs/content/develop/collectors.md

@@ -161,6 +161,7 @@ Then, when `Collect` is called, the collector quickly returns the latest value(s
 
 Using this method, the collector should return a cleanup function (callback) from `Prepare`.
 Blip calls collector cleanup functions if/when a monitor is stopped or restarted.
+If `Prepare` returns an error, Blip discards the collector, so `Prepare` must stop any background threads that it started before it returns.
 
 ### ErrMore
 

docs/content/metrics/domains/repl.lag/_index.md

@@ -71,6 +71,10 @@ Only available with MySQL 8.x Performance Schema.
 
 ## Options
 
+When `repl.lag` is collected at more than one level, `writer`, `repl-check`, `table`, `source-id`, `source-role`, and `network-latency` are read from one level only: the first level, by name, that sets `writer` to `pfs` or `blip`, else the first level that collects `repl.lag`.
+The options `report-no-heartbeat`, `report-not-a-replica`, and `default-channel-name` apply per level.
+If one level sets `writer: pfs` and another sets `writer: blip`, the plan fails to prepare.
+
 ### Common
 
 #### `repl-check`

heartbeat/heartbeat_test.go

@@ -189,11 +189,15 @@ func TestReader(t *testing.T) {
 	// the real lag (and wait time, which we ignore) that the real waiter
 	// returns to the reader.
 	hbChan := make(chan int64, 1)
+	done := make(chan struct{}) // closed when test returns, so a blocked send doesn't block Stop
 	realWaiter := heartbeat.SlowFastWaiter{NetworkLatency: 10 * time.Millisecond}
 	mockWaiter := mock.LagWaiter{
 		WaitFunc: func(now, then time.Time, f int, srcId string) (int64, time.Duration) {
 			lag, wait := realWaiter.Wait(now, then, f, srcId)
-			hbChan <- lag
+			select {
+			case hbChan <- lag:
+			case <-done:
+			}
 			return lag, wait
 		},
 	}
@@ -204,6 +208,8 @@ func TestReader(t *testing.T) {
 		Waiter:    mockWaiter,
 	})
 	hr.Start()
+	defer hr.Stop()   // before ReadErrorWait is reset: Stop waits for the reader to return
+	defer close(done) // before Stop: unblocks the reader if it's sending to hbChan
 
 	timeout := time.After(5 * time.Second) // this whole test should take <1s
 

heartbeat/reader.go

@@ -54,7 +54,8 @@ type BlipReader struct {
 	lag      int64
 	last     time.Time
 	stopChan chan struct{}
-	doneChan chan struct{}
+	doneChan chan struct{} // closed when run returns
+	started  bool          // run started, so Stop waits for doneChan
 	isRepl   bool
 	event    event.MonitorReceiver
 	query    string
@@ -113,7 +114,20 @@ func NewBlipReader(args BlipReaderArgs) *BlipReader {
 	return r
 }
 
+// Start starts the reader in a goroutine. It doesn't start the reader if Start
+// was called before, or if Stop was called first.
 func (r *BlipReader) Start() error {
+	r.Lock()
+	defer r.Unlock()
+	select {
+	case <-r.stopChan:
+		return nil // stopped before started
+	default:
+	}
+	if r.started {
+		return nil
+	}
+	r.started = true
 	go r.run()
 	return nil
 }
@@ -211,15 +225,24 @@ func (r *BlipReader) wait(d time.Duration) bool {
 	}
 }
 
+// Stop stops the reader and waits for it to return, so it doesn't report
+// status after Stop returns. If a heartbeat query is running, Stop waits for it
+// to return, which takes up to ReadTimeout. Stop can be called more than once,
+// concurrently, and before Start.
 func (r *BlipReader) Stop() {
 	r.Lock()
 	select {
 	case <-r.stopChan:
-	case <-r.doneChan:
 	default:
 		close(r.stopChan)
 	}
+	started := r.started
 	r.Unlock()
+
+	// Wait without the lock: run locks to update lag
+	if started {
+		<-r.doneChan
+	}
 }
 
 func (r *BlipReader) Lag(_ context.Context) (Lag, error) {

heartbeat/reader_query_test.go

@@ -0,0 +1,20 @@
+// Copyright 2026 Block, Inc.
+
+package heartbeat
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestBlipReaderQuerySelectsReplCheck(t *testing.T) {
+	r := NewBlipReader(BlipReaderArgs{MonitorId: "m1", Table: "blip.heartbeat", ReplCheck: "read_only"})
+	if !strings.Contains(r.query, "@@read_only") {
+		t.Errorf("query %q does not select @@read_only", r.query)
+	}
+
+	r = NewBlipReader(BlipReaderArgs{MonitorId: "m1", Table: "blip.heartbeat"})
+	if strings.Contains(r.query, "@@") {
+		t.Errorf("query %q selects a variable, expected none without repl check", r.query)
+	}
+}

heartbeat/reader_stop_test.go

@@ -0,0 +1,108 @@
+// Copyright 2026 Block, Inc.
+
+package heartbeat
+
+import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/cashapp/blip/v2/status"
+)
+
+// blockingDB is a database whose queries block until release is closed (or
+// the query context is done), then fail.
+type blockingDB struct {
+	running chan struct{} // receives when a query starts
+	release chan struct{}
+}
+
+func newBlockingDB(t *testing.T) (*sql.DB, *blockingDB) {
+	b := &blockingDB{running: make(chan struct{}, 1), release: make(chan struct{})}
+	db := sql.OpenDB(b)
+	t.Cleanup(func() { db.Close() })
+	return db, b
+}
+
+func (b *blockingDB) Connect(context.Context) (driver.Conn, error) { return blockingConn{b}, nil }
+func (b *blockingDB) Driver() driver.Driver                        { return nil }
+
+type blockingConn struct{ b *blockingDB }
+
+func (c blockingConn) QueryContext(ctx context.Context, _ string, _ []driver.NamedValue) (driver.Rows, error) {
+	select {
+	case c.b.running <- struct{}{}:
+	default:
+	}
+	select {
+	case <-c.b.release:
+	case <-ctx.Done():
+	}
+	return nil, errors.New("query failed")
+}
+
+func (blockingConn) Prepare(string) (driver.Stmt, error) { return nil, errors.New("not supported") }
+func (blockingConn) Close() error                        { return nil }
+func (blockingConn) Begin() (driver.Tx, error)           { return nil, errors.New("not supported") }
+
+func TestBlipReaderStopWaitsForRunningQuery(t *testing.T) {
+	// The monitor loader removes a monitor's status after stopping it. If Stop
+	// returned while a heartbeat query was running, the reader would report
+	// status after that, which makes the removed monitor reappear.
+	monitorId := "stop-waits"
+	db, d := newBlockingDB(t)
+	r := NewBlipReader(BlipReaderArgs{MonitorId: monitorId, DB: db, Table: "blip.heartbeat"})
+	r.Start()
+	select {
+	case <-d.running:
+	case <-time.After(5 * time.Second):
+		t.Fatal("heartbeat query didn't start")
+	}
+
+	// Stop while the query is running, which returns after Stop is called
+	time.AfterFunc(100*time.Millisecond, func() { close(d.release) })
+	var wg sync.WaitGroup
+	for i := 0; i < 2; i++ { // concurrent Stop calls
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			r.Stop()
+		}()
+	}
+	wg.Wait()
+	status.RemoveMonitor(monitorId)
+
+	<-r.doneChan // reader returned, so it reported any status it was going to
+	if got := status.ReportMonitors(monitorId)[monitorId]; len(got) != 0 {
+		t.Errorf("monitor status after Stop and RemoveMonitor: %v, expected none", got)
+	}
+	r.Stop() // again, after it stopped
+}
+
+func TestBlipReaderStopBeforeStart(t *testing.T) {
+	// Repl.lag calls Start in a goroutine, so Stop can be called first
+	db, _ := newBlockingDB(t)
+	r := NewBlipReader(BlipReaderArgs{MonitorId: "stop-before-start", DB: db, Table: "blip.heartbeat"})
+	stopped := make(chan struct{})
+	go func() {
+		r.Stop()
+		r.Start() // doesn't start a stopped reader
+		r.Stop()
+		close(stopped)
+	}()
+	select {
+	case <-stopped:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Stop blocked for a reader that wasn't started")
+	}
+	r.Lock()
+	started := r.started
+	r.Unlock()
+	if started {
+		t.Error("Start started a stopped reader")
+	}
+}