Stop the level collector without waiting for the next tick

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.

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.
 
 ---
 

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,29 +101,28 @@ 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
 		}
 
 		// 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
-		}
 	}
 }

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)
+		}
+	}
+}

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
+}

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")
+	}
+}