From 90361c4554f13e1781119dc17d99c6c7a37cce5b Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 22 Oct 2026 13:53:23 -0700 Subject: [PATCH 1/8] Set up repl.lag once and apply options at every level Lag.Prepare never assigned `configured`, so every level re-ran the full writer setup and mixed writers were never rejected. With the blip writer, only the first level (in random map order) got report-no-heartbeat, so other levels reported -1 instead of dropping it. And the heartbeat reader was made before repl-check was set, so it never selected @@ and report-not-a-replica did nothing. Prepare now sets up the writer once, from the first level, rejects a different writer at another level (an unset writer is auto), and sets the per-level options at every level. The reader gets repl-check from the level options. Document on blip.Collector.Prepare that a collector must release what it started before it returns an error, with the engine calling a returned cleanup func only as a backstop. Lag already did this. Co-Authored-By: Claude Opus 5.5 --- collector.go | 17 +++- docs/content/about/release-notes.md | 1 + docs/content/develop/collectors.md | 1 + heartbeat/reader_query_test.go | 20 ++++ metrics/repl.lag/lag.go | 121 +++++++++++----------- metrics/repl.lag/lag_test.go | 151 ++++++++++++++++++++++++++++ monitor/engine.go | 3 + 7 files changed, 253 insertions(+), 61 deletions(-) create mode 100644 heartbeat/reader_query_test.go diff --git a/collector.go b/collector.go index 4a53dff..1b6ee97 100644 --- a/collector.go +++ b/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 diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 9b9b89d..6f8354a 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -61,6 +61,7 @@ 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` and `report-not-a-replica` now apply at every level, not only at one random level, and the heartbeat reader now uses `repl-check`. Plans that set a different `writer` at different levels (an unset `writer` is `auto`) now fail to prepare. ### v2.1.0 (10 Aug 2026) diff --git a/docs/content/develop/collectors.md b/docs/content/develop/collectors.md index 2483886..ab3dc59 100644 --- a/docs/content/develop/collectors.md +++ b/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 diff --git a/heartbeat/reader_query_test.go b/heartbeat/reader_query_test.go new file mode 100644 index 0000000..4914b53 --- /dev/null +++ b/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) + } +} diff --git a/metrics/repl.lag/lag.go b/metrics/repl.lag/lag.go index ed35a50..2c8fd66 100644 --- a/metrics/repl.lag/lag.go +++ b/metrics/repl.lag/lag.go @@ -33,7 +33,8 @@ const ( type Lag struct { db *sql.DB lagReader heartbeat.Reader - stopReader func() // stops lagReader; returned by Prepare as cleanup + stopReader func() // stops lagReader; returned by Prepare as cleanup + newReader func(heartbeat.BlipReaderArgs) heartbeat.Reader // tests replace it to inspect the args lagWriterIn map[string]string dropNoHeartbeat map[string]bool dropNotAReplica map[string]bool @@ -45,9 +46,14 @@ type Lag struct { var _ blip.Collector = &Lag{} +func newBlipReader(args heartbeat.BlipReaderArgs) heartbeat.Reader { + return heartbeat.NewBlipReader(args) +} + func NewLag(db *sql.DB) *Lag { return &Lag{ db: db, + newReader: newBlipReader, lagWriterIn: map[string]string{}, dropNoHeartbeat: map[string]bool{}, dropNotAReplica: map[string]bool{}, @@ -149,10 +155,16 @@ func (c *Lag) Help() blip.CollectorHelp { // domain collects only one metric (repl.lag.current), there's no need to // collect different metrics at different frequencies. // +// The first level sets up the writer. Every level must use the same writer, +// but the report-no-heartbeat, report-not-a-replica, and default-channel-name +// options are per level. +// // If lag is collected from a Blip heartbeat, the returned cleanup func stops -// the heartbeat reader; else it's nil. +// the heartbeat reader; else it's nil. On error, Prepare stops the reader +// and returns a nil cleanup func, as required by blip.Collector. func (c *Lag) Prepare(ctx context.Context, plan blip.Plan) (cleanup func(), err error) { - configured := "" // set after first level to its writer value + configured := "" // writer option of first level, set after it's configured + writer := "" // writer that configured resolved to: pfs or blip // The engine discards this collector if Prepare fails, so stop the // reader if it was started before the error. @@ -169,60 +181,63 @@ LEVEL: continue LEVEL // not collected in this level } - writer := dom.Options[OPT_WRITER] - - // Already configured? If yes and same writer, that's ok and expected - // (lag collected at multiple levels). But if writer is different, that's - // and error. - if configured != "" { - if configured != writer { - return nil, fmt.Errorf("different writer configuration: %s != %s", configured, writer) - } - c.lagWriterIn[levelName] = writer // collect at this level - continue LEVEL + opt := dom.Options[OPT_WRITER] + if opt == "" { + opt = "auto" // default } - blip.Debug("repl.lag: config from level %s", levelName) - switch writer { - case LAG_WRITER_PFS: - // Try collecting, discard metrics - if _, err = c.collectPFS(ctx, levelName); err != nil { + // Set up the writer at the first level. At other levels, the same writer + // is ok and expected (lag collected at multiple levels). But if writer + // is different, that's an error. + if configured == "" { + if writer, err = c.configure(ctx, levelName, plan, opt, dom.Options); err != nil { return nil, err } - case LAG_WRITER_BLIP: - cleanup, err = c.prepareBlip(levelName, plan.MonitorId, plan.Name, dom.Options) - if err != nil { - return nil, err - } - case "auto", "": // default - // Try PFS first - if _, err = c.collectPFS(ctx, levelName); err == nil { - blip.Debug("repl.lag auto-detected PFS") - writer = LAG_WRITER_PFS - } else { - // then Blip HeartBeat - if cleanup, err = c.prepareBlip(levelName, plan.MonitorId, plan.Name, dom.Options); err == nil { - blip.Debug("repl.lag auto-detected Blip heartbeat") - writer = LAG_WRITER_BLIP - } else { - return nil, fmt.Errorf("failed to auto-detect source, set %s manually", OPT_WRITER) - } - } - default: - return nil, fmt.Errorf("invalid lag writer: %q; valid values: auto, pfs, blip", writer) + configured = opt + } else if opt != configured { + return nil, fmt.Errorf("different writer configuration: %s != %s", configured, opt) } c.lagWriterIn[levelName] = writer // collect at this level - + c.dropNoHeartbeat[levelName] = !blip.Bool(dom.Options[OPT_REPORT_NO_HEARTBEAT]) c.dropNotAReplica[levelName] = !blip.Bool(dom.Options[OPT_REPORT_NOT_A_REPLICA]) c.defaultChannelNameOverrides[levelName] = dom.Options[OPT_DEFAULT_CHANNEL_NAME] - if err := c.verifyReplCheck(ctx, dom.Options[OPT_REPL_CHECK]); err != nil { - return nil, err + } + + return c.stopReader, nil +} + +// configure sets up the writer from the options of the first level that +// collects lag. It returns the writer, which is never auto. +func (c *Lag) configure(ctx context.Context, levelName string, plan blip.Plan, writer string, options map[string]string) (string, error) { + blip.Debug("repl.lag: config from level %s", levelName) + switch writer { + case LAG_WRITER_PFS: + // Try collecting, discard metrics + if _, err := c.collectPFS(ctx, levelName); err != nil { + return "", err + } + case LAG_WRITER_BLIP: + c.prepareBlip(levelName, plan.MonitorId, plan.Name, options) + case "auto": + // Try PFS first, then Blip heartbeat + if _, err := c.collectPFS(ctx, levelName); err == nil { + blip.Debug("repl.lag auto-detected PFS") + writer = LAG_WRITER_PFS + } else { + c.prepareBlip(levelName, plan.MonitorId, plan.Name, options) + blip.Debug("repl.lag auto-detected Blip heartbeat") + writer = LAG_WRITER_BLIP } - c.replCheck = dom.Options[OPT_REPL_CHECK] + default: + return "", fmt.Errorf("invalid lag writer: %q; valid values: auto, pfs, blip", writer) } - return cleanup, nil + if err := c.verifyReplCheck(ctx, options[OPT_REPL_CHECK]); err != nil { + return "", err + } + c.replCheck = options[OPT_REPL_CHECK] + return writer, nil } func (c *Lag) verifyReplCheck(ctx context.Context, variable string) error { @@ -253,15 +268,12 @@ func (c *Lag) Collect(ctx context.Context, levelName string) ([]blip.MetricValue // Internal methods // ////////////////////////////////////////////////////////////////////////// -func (c *Lag) prepareBlip(levelName string, monitorID string, planName string, options map[string]string) (func(), error) { - // Prepare calls this func for each level that collects lag. Return the - // same cleanup func for every level, else the reader is never stopped. +// prepareBlip starts the heartbeat reader. Only 1 reader per collector. +func (c *Lag) prepareBlip(levelName string, monitorID string, planName string, options map[string]string) { if c.lagReader != nil { - return c.stopReader, nil + return } - c.dropNoHeartbeat[levelName] = !blip.Bool(options[OPT_REPORT_NO_HEARTBEAT]) - table := options[OPT_HEARTBEAT_TABLE] if table == "" { table = blip.DEFAULT_HEARTBEAT_TABLE @@ -275,14 +287,13 @@ func (c *Lag) prepareBlip(levelName string, monitorID string, planName string, o netLatency = time.Duration(n) * time.Millisecond } } - // Only 1 reader per plan - c.lagReader = heartbeat.NewBlipReader(heartbeat.BlipReaderArgs{ + c.lagReader = c.newReader(heartbeat.BlipReaderArgs{ MonitorId: monitorID, DB: c.db, Table: table, SourceId: options[OPT_HEARTBEAT_SOURCE_ID], SourceRole: options[OPT_HEARTBEAT_SOURCE_ROLE], - ReplCheck: c.replCheck, + ReplCheck: options[OPT_REPL_CHECK], // not c.replCheck: it's set after the reader starts Waiter: heartbeat.SlowFastWaiter{ MonitorId: monitorID, NetworkLatency: netLatency, @@ -290,12 +301,10 @@ func (c *Lag) prepareBlip(levelName string, monitorID string, planName string, o }) go c.lagReader.Start() blip.Debug("%s: started reader: %s/%s (network latency: %s)", monitorID, planName, levelName, netLatency) - c.lagWriterIn[levelName] = LAG_WRITER_BLIP c.stopReader = func() { blip.Debug("%s: stopping reader", monitorID) c.lagReader.Stop() } - return c.stopReader, nil } func (c *Lag) collectBlip(ctx context.Context, levelName string) ([]blip.MetricValue, error) { diff --git a/metrics/repl.lag/lag_test.go b/metrics/repl.lag/lag_test.go index 16f8824..df187f7 100644 --- a/metrics/repl.lag/lag_test.go +++ b/metrics/repl.lag/lag_test.go @@ -12,6 +12,7 @@ import ( "github.com/stretchr/testify/assert" "github.com/cashapp/blip/v2" + "github.com/cashapp/blip/v2/heartbeat" "github.com/cashapp/blip/v2/test" "github.com/cashapp/blip/v2/test/mock" ) @@ -223,3 +224,153 @@ func TestCollectPFSReturnsDeadlineExceeded(t *testing.T) { t.Errorf("Collect returned error %v, expected an error wrapping context.DeadlineExceeded", err) } } + +// fakeReader is a heartbeat reader that reports lag without a database. +type fakeReader struct { + lag heartbeat.Lag +} + +func (r fakeReader) Start() error { return nil } +func (r fakeReader) Stop() {} +func (r fakeReader) Lag(context.Context) (heartbeat.Lag, error) { return r.lag, nil } + +func TestPrepareBlipAtMultipleLevelsDropsNoHeartbeat(t *testing.T) { + // report-no-heartbeat defaults to no (drop), and that applies to every + // level, not only to the level that started the reader. The reader can't + // reach the database, so it reports no heartbeat. + for _, report := range []string{"", "no", "yes"} { + t.Run("report-no-heartbeat="+report, func(t *testing.T) { + c := NewLag(unreachableDB(t)) + plan := blipPlan("fast", "medium", "slow") + for _, level := range plan.Levels { + if report != "" { + level.Collect[DOMAIN].Options[OPT_REPORT_NO_HEARTBEAT] = report + } + } + cleanup, err := c.Prepare(context.Background(), plan) + stopReader(t, cleanup) + if err != nil { + t.Fatal(err) + } + for level := range plan.Levels { + metrics, err := c.Collect(context.Background(), level) + if err != nil { + t.Fatalf("%s: %s", level, err) + } + if report == "yes" { + if len(metrics) != 1 || metrics[0].Value != -1 { + t.Errorf("%s: got %+v, expected repl.lag.current = -1", level, metrics) + } + } else if len(metrics) != 0 { + t.Errorf("%s: got %+v, expected no metrics (dropped)", level, metrics) + } + } + }) + } +} + +func TestPrepareBlipPassesReplCheckToReader(t *testing.T) { + // The reader selects @@ to report whether the instance is a + // replica, so it must be made with the repl-check option. The repl check + // is verified after the reader starts, and that fails because the database + // is unreachable, but by then the reader has been made. + c := NewLag(unreachableDB(t)) + var args []heartbeat.BlipReaderArgs + c.newReader = func(a heartbeat.BlipReaderArgs) heartbeat.Reader { + args = append(args, a) + return fakeReader{} + } + plan := blipPlan("fast", "slow") + for _, level := range plan.Levels { + level.Collect[DOMAIN].Options[OPT_REPL_CHECK] = "read_only" + } + if _, err := c.Prepare(context.Background(), plan); err == nil { + t.Fatal("Prepare returned nil error, expected repl check error") + } + if len(args) != 1 { + t.Fatalf("Prepare made %d heartbeat readers, expected 1", len(args)) + } + assert.Equal(t, "read_only", args[0].ReplCheck) +} + +func TestPrepareRejectsDifferentWriters(t *testing.T) { + // The writer must be the same at every level. With an unreachable database, + // auto resolves to blip, but it's still a different writer configuration. + // No writer is the same as auto. + tests := []struct { + fast, slow string + ok bool + }{ + {fast: LAG_WRITER_BLIP, slow: "auto", ok: false}, + {fast: "auto", slow: LAG_WRITER_BLIP, ok: false}, + {fast: LAG_WRITER_BLIP, slow: LAG_WRITER_BLIP, ok: true}, + {fast: "", slow: "auto", ok: true}, + } + for _, tt := range tests { + t.Run(tt.fast+"/"+tt.slow, func(t *testing.T) { + c := NewLag(unreachableDB(t)) + c.newReader = func(heartbeat.BlipReaderArgs) heartbeat.Reader { return fakeReader{} } + plan := blipPlan("fast", "slow") + plan.Levels["fast"].Collect[DOMAIN].Options[OPT_WRITER] = tt.fast + plan.Levels["slow"].Collect[DOMAIN].Options[OPT_WRITER] = tt.slow + _, err := c.Prepare(context.Background(), plan) + if tt.ok { + assert.NoError(t, err) + } else { + assert.ErrorContains(t, err, "different writer configuration") + } + }) + } +} + +func TestPrepareSetsOptionsAtEveryLevel(t *testing.T) { + // The writer is set up once, but every level has its own report and + // channel name options. + c := NewLag(unreachableDB(t)) + c.newReader = func(heartbeat.BlipReaderArgs) heartbeat.Reader { + return fakeReader{lag: heartbeat.Lag{Replica: false, Milliseconds: -1}} + } + plan := blipPlan("fast", "medium", "slow") + plan.Levels["fast"].Collect[DOMAIN].Options[OPT_REPORT_NO_HEARTBEAT] = "yes" + plan.Levels["medium"].Collect[DOMAIN].Options[OPT_REPORT_NOT_A_REPLICA] = "yes" + plan.Levels["slow"].Collect[DOMAIN].Options[OPT_DEFAULT_CHANNEL_NAME] = "main" + if _, err := c.Prepare(context.Background(), plan); err != nil { + t.Fatal(err) + } + assert.Equal(t, map[string]string{"fast": LAG_WRITER_BLIP, "medium": LAG_WRITER_BLIP, "slow": LAG_WRITER_BLIP}, c.lagWriterIn) + assert.Equal(t, map[string]bool{"fast": false, "medium": true, "slow": true}, c.dropNoHeartbeat) + assert.Equal(t, map[string]bool{"fast": true, "medium": false, "slow": true}, c.dropNotAReplica) + assert.Equal(t, map[string]string{"fast": "", "medium": "", "slow": "main"}, c.defaultChannelNameOverrides) + + // Not a replica: reported only at the level with report-not-a-replica=yes + for level := range plan.Levels { + metrics, err := c.Collect(context.Background(), level) + if err != nil { + t.Fatalf("%s: %s", level, err) + } + if level == "medium" { + assert.Len(t, metrics, 1, level) + } else { + assert.Empty(t, metrics, level) + } + } +} + +func TestPrepareRejectsPFSAndBlipWriters(t *testing.T) { + _, db, err := test.Connection("mysql80") + if err != nil { + if test.Build { + t.Skip("mysql80 not running") + } else { + t.Fatal(err) + } + } + defer db.Close() + + c := NewLag(db) + plan := blipPlan("fast", "slow") + plan.Levels["fast"].Collect[DOMAIN].Options[OPT_WRITER] = LAG_WRITER_PFS + cleanup, err := c.Prepare(context.Background(), plan) + stopReader(t, cleanup) + assert.ErrorContains(t, err, "different writer configuration") +} diff --git a/monitor/engine.go b/monitor/engine.go index 2b0c309..c955aa3 100644 --- a/monitor/engine.go +++ b/monitor/engine.go @@ -224,6 +224,9 @@ func (e *Engine) Prepare(ctx context.Context, plan blip.Plan, before, after func status.Monitor(e.monitorId, status.ENGINE_PREPARE, "%s: prepare collector %s", plan.Name, domain) cleanup, err := c.Prepare(ctx, plan) if err != nil { + // The collector should have released its resources and returned + // nil, but call a returned cleanup func as a backstop. See + // blip.Collector.Prepare. if cleanup != nil { cleanup() } From 230477a6932bf6419146c744e3449e5c2c72a202 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 22 Oct 2026 14:07:16 -0700 Subject: [PATCH 2/8] Let auto levels use the repl.lag writer set at another level Prepare compared each level's raw writer option with the first level's, so a plan with no writer (auto) at one level and writer: blip at another failed to prepare. Plans have that because levels inherit options only from lower levels whose frequency divides theirs. Which side the error named also depended on map order. Now Prepare sets up the writer from the first level, by name, that sets pfs or blip, else it auto-detects once from the first level. Levels with no writer or auto use that writer. Only pfs at one level and blip at another is an error. Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 2 +- metrics/repl.lag/lag.go | 77 +++++++++++++---------- metrics/repl.lag/lag_test.go | 94 +++++++++++++++++++++-------- 3 files changed, 114 insertions(+), 59 deletions(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 6f8354a..7723f15 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -61,7 +61,7 @@ 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` and `report-not-a-replica` now apply at every level, not only at one random level, and the heartbeat reader now uses `repl-check`. Plans that set a different `writer` at different levels (an unset `writer` is `auto`) now fail to prepare. +* Fixed `repl.lag` at multiple levels with the Blip heartbeat writer: `report-no-heartbeat` and `report-not-a-replica` now apply at every level, not only at one random level, and the heartbeat reader now uses `repl-check`. Levels that don't set `writer`, or set it to `auto`, use the writer set at another level. Plans that set `writer` to `pfs` at one level and `blip` at another now fail to prepare. ### v2.1.0 (10 Aug 2026) diff --git a/metrics/repl.lag/lag.go b/metrics/repl.lag/lag.go index 2c8fd66..25b15b4 100644 --- a/metrics/repl.lag/lag.go +++ b/metrics/repl.lag/lag.go @@ -6,6 +6,7 @@ import ( "context" "database/sql" "fmt" + "sort" "strconv" "time" @@ -155,17 +156,17 @@ func (c *Lag) Help() blip.CollectorHelp { // domain collects only one metric (repl.lag.current), there's no need to // collect different metrics at different frequencies. // -// The first level sets up the writer. Every level must use the same writer, -// but the report-no-heartbeat, report-not-a-replica, and default-channel-name -// options are per level. +// One level sets up the writer: the first level, by name, that sets the +// writer option to pfs or blip; else the first level that collects lag, which +// auto-detects the writer. Levels that don't set the writer, or set it to +// auto, use that writer. It's an error if one level sets pfs and another sets +// blip. The report-no-heartbeat, report-not-a-replica, and +// default-channel-name options are per level. // // If lag is collected from a Blip heartbeat, the returned cleanup func stops // the heartbeat reader; else it's nil. On error, Prepare stops the reader // and returns a nil cleanup func, as required by blip.Collector. func (c *Lag) Prepare(ctx context.Context, plan blip.Plan) (cleanup func(), err error) { - configured := "" // writer option of first level, set after it's configured - writer := "" // writer that configured resolved to: pfs or blip - // The engine discards this collector if Prepare fails, so stop the // reader if it was started before the error. defer func() { @@ -174,41 +175,55 @@ func (c *Lag) Prepare(ctx context.Context, plan blip.Plan) (cleanup func(), err } }() -LEVEL: + // Levels that collect lag, sorted so the result doesn't depend on map order + levels := []string{} for levelName, level := range plan.Levels { - dom, ok := level.Collect[DOMAIN] - if !ok { - continue LEVEL // not collected in this level + if _, ok := level.Collect[DOMAIN]; ok { + levels = append(levels, levelName) } + } + if len(levels) == 0 { + return nil, nil + } + sort.Strings(levels) - opt := dom.Options[OPT_WRITER] - if opt == "" { - opt = "auto" // default + // Find the level that sets up the writer + configLevel := levels[0] + writer := "auto" + for _, levelName := range levels { + opt := plan.Levels[levelName].Collect[DOMAIN].Options[OPT_WRITER] + switch opt { + case "", "auto": + continue + case LAG_WRITER_PFS, LAG_WRITER_BLIP: + default: + return nil, fmt.Errorf("invalid lag writer: %q; valid values: auto, pfs, blip", opt) } - - // Set up the writer at the first level. At other levels, the same writer - // is ok and expected (lag collected at multiple levels). But if writer - // is different, that's an error. - if configured == "" { - if writer, err = c.configure(ctx, levelName, plan, opt, dom.Options); err != nil { - return nil, err - } - configured = opt - } else if opt != configured { - return nil, fmt.Errorf("different writer configuration: %s != %s", configured, opt) + if writer == "auto" { + configLevel = levelName + writer = opt + } else if opt != writer { + return nil, fmt.Errorf("different writer configuration: %s at level %s != %s at level %s", writer, configLevel, opt, levelName) } + } + + if writer, err = c.configure(ctx, configLevel, plan, writer, plan.Levels[configLevel].Collect[DOMAIN].Options); err != nil { + return nil, err + } + for _, levelName := range levels { + options := plan.Levels[levelName].Collect[DOMAIN].Options c.lagWriterIn[levelName] = writer // collect at this level - c.dropNoHeartbeat[levelName] = !blip.Bool(dom.Options[OPT_REPORT_NO_HEARTBEAT]) - c.dropNotAReplica[levelName] = !blip.Bool(dom.Options[OPT_REPORT_NOT_A_REPLICA]) - c.defaultChannelNameOverrides[levelName] = dom.Options[OPT_DEFAULT_CHANNEL_NAME] + c.dropNoHeartbeat[levelName] = !blip.Bool(options[OPT_REPORT_NO_HEARTBEAT]) + c.dropNotAReplica[levelName] = !blip.Bool(options[OPT_REPORT_NOT_A_REPLICA]) + c.defaultChannelNameOverrides[levelName] = options[OPT_DEFAULT_CHANNEL_NAME] } return c.stopReader, nil } -// configure sets up the writer from the options of the first level that -// collects lag. It returns the writer, which is never auto. +// configure sets up the writer from the options of one level. It returns the +// writer, which is never auto. func (c *Lag) configure(ctx context.Context, levelName string, plan blip.Plan, writer string, options map[string]string) (string, error) { blip.Debug("repl.lag: config from level %s", levelName) switch writer { @@ -219,7 +234,7 @@ func (c *Lag) configure(ctx context.Context, levelName string, plan blip.Plan, w } case LAG_WRITER_BLIP: c.prepareBlip(levelName, plan.MonitorId, plan.Name, options) - case "auto": + default: // auto // Try PFS first, then Blip heartbeat if _, err := c.collectPFS(ctx, levelName); err == nil { blip.Debug("repl.lag auto-detected PFS") @@ -229,8 +244,6 @@ func (c *Lag) configure(ctx context.Context, levelName string, plan blip.Plan, w blip.Debug("repl.lag auto-detected Blip heartbeat") writer = LAG_WRITER_BLIP } - default: - return "", fmt.Errorf("invalid lag writer: %q; valid values: auto, pfs, blip", writer) } if err := c.verifyReplCheck(ctx, options[OPT_REPL_CHECK]); err != nil { diff --git a/metrics/repl.lag/lag_test.go b/metrics/repl.lag/lag_test.go index df187f7..5baddb9 100644 --- a/metrics/repl.lag/lag_test.go +++ b/metrics/repl.lag/lag_test.go @@ -6,6 +6,7 @@ import ( "context" "database/sql" "errors" + "strings" "testing" "time" @@ -293,31 +294,59 @@ func TestPrepareBlipPassesReplCheckToReader(t *testing.T) { assert.Equal(t, "read_only", args[0].ReplCheck) } -func TestPrepareRejectsDifferentWriters(t *testing.T) { - // The writer must be the same at every level. With an unreachable database, - // auto resolves to blip, but it's still a different writer configuration. - // No writer is the same as auto. +func TestPrepareWritersAtDifferentLevels(t *testing.T) { + // Levels that don't set the writer, or set auto, use the writer set at + // another level. Plans have them because levels inherit options only from + // some lower levels. But pfs at one level and blip at another is an error. + // Each pair is tested in both orders, and many times because Prepare + // ranges over a map. tests := []struct { fast, slow string - ok bool + writer string // expected at every level, or "" if Prepare fails + err string // expected error, if writer is "" }{ - {fast: LAG_WRITER_BLIP, slow: "auto", ok: false}, - {fast: "auto", slow: LAG_WRITER_BLIP, ok: false}, - {fast: LAG_WRITER_BLIP, slow: LAG_WRITER_BLIP, ok: true}, - {fast: "", slow: "auto", ok: true}, + {fast: "", slow: LAG_WRITER_BLIP, writer: LAG_WRITER_BLIP}, + {fast: LAG_WRITER_BLIP, slow: "", writer: LAG_WRITER_BLIP}, + {fast: "auto", slow: LAG_WRITER_BLIP, writer: LAG_WRITER_BLIP}, + {fast: LAG_WRITER_BLIP, slow: "auto", writer: LAG_WRITER_BLIP}, + {fast: LAG_WRITER_BLIP, slow: LAG_WRITER_BLIP, writer: LAG_WRITER_BLIP}, + {fast: "", slow: "auto", writer: LAG_WRITER_BLIP}, // auto-detected: PFS fails because database is unreachable + {fast: LAG_WRITER_PFS, slow: LAG_WRITER_BLIP, err: "different writer configuration"}, + {fast: LAG_WRITER_BLIP, slow: LAG_WRITER_PFS, err: "different writer configuration"}, + // pfs is set up, not auto-detected or blip, so it fails because the + // database is unreachable. See TestPrepareAutoAndPFSOnMySQL80. + {fast: "auto", slow: LAG_WRITER_PFS, err: "could not check replication lag"}, + {fast: LAG_WRITER_PFS, slow: "", err: "could not check replication lag"}, } for _, tt := range tests { t.Run(tt.fast+"/"+tt.slow, func(t *testing.T) { - c := NewLag(unreachableDB(t)) - c.newReader = func(heartbeat.BlipReaderArgs) heartbeat.Reader { return fakeReader{} } - plan := blipPlan("fast", "slow") - plan.Levels["fast"].Collect[DOMAIN].Options[OPT_WRITER] = tt.fast - plan.Levels["slow"].Collect[DOMAIN].Options[OPT_WRITER] = tt.slow - _, err := c.Prepare(context.Background(), plan) - if tt.ok { - assert.NoError(t, err) - } else { - assert.ErrorContains(t, err, "different writer configuration") + for i := 0; i < 20; i++ { + c := NewLag(unreachableDB(t)) + readers := 0 + c.newReader = func(heartbeat.BlipReaderArgs) heartbeat.Reader { + readers++ + return fakeReader{} + } + plan := blipPlan("fast", "slow") + plan.Levels["fast"].Collect[DOMAIN].Options[OPT_WRITER] = tt.fast + plan.Levels["slow"].Collect[DOMAIN].Options[OPT_WRITER] = tt.slow + _, err := c.Prepare(context.Background(), plan) + if tt.writer == "" { + if err == nil || !strings.Contains(err.Error(), tt.err) { + t.Fatalf("run %d: got error %v, expected %q", i, err, tt.err) + } + continue + } + if err != nil { + t.Fatalf("run %d: %s", i, err) + } + if readers != 1 { + t.Fatalf("run %d: Prepare made %d heartbeat readers, expected 1", i, readers) + } + want := map[string]string{"fast": tt.writer, "slow": tt.writer} + if !assert.Equal(t, want, c.lagWriterIn, "run %d", i) { + return + } } }) } @@ -356,7 +385,7 @@ func TestPrepareSetsOptionsAtEveryLevel(t *testing.T) { } } -func TestPrepareRejectsPFSAndBlipWriters(t *testing.T) { +func TestPrepareAutoAndPFSOnMySQL80(t *testing.T) { _, db, err := test.Connection("mysql80") if err != nil { if test.Build { @@ -367,10 +396,23 @@ func TestPrepareRejectsPFSAndBlipWriters(t *testing.T) { } defer db.Close() - c := NewLag(db) - plan := blipPlan("fast", "slow") - plan.Levels["fast"].Collect[DOMAIN].Options[OPT_WRITER] = LAG_WRITER_PFS - cleanup, err := c.Prepare(context.Background(), plan) - stopReader(t, cleanup) - assert.ErrorContains(t, err, "different writer configuration") + // Same as auto and blip in TestPrepareWritersAtDifferentLevels, but pfs + // needs MySQL + for _, writers := range [][2]string{{"auto", LAG_WRITER_PFS}, {LAG_WRITER_PFS, ""}} { + for i := 0; i < 20; i++ { + c := NewLag(db) + plan := blipPlan("fast", "slow") + plan.Levels["fast"].Collect[DOMAIN].Options[OPT_WRITER] = writers[0] + plan.Levels["slow"].Collect[DOMAIN].Options[OPT_WRITER] = writers[1] + cleanup, err := c.Prepare(context.Background(), plan) + stopReader(t, cleanup) + if err != nil { + t.Fatalf("%v run %d: %s", writers, i, err) + } + if c.lagReader != nil { + t.Fatalf("%v run %d: Prepare made a heartbeat reader for pfs", writers, i) + } + assert.Equal(t, map[string]string{"fast": LAG_WRITER_PFS, "slow": LAG_WRITER_PFS}, c.lagWriterIn) + } + } } From bc4b9b44b678c4a679783af5dba7aeddd497c2d3 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 22 Oct 2026 14:31:40 -0700 Subject: [PATCH 3/8] Wait for the heartbeat reader to return in Stop Stop closed stopChan but didn't wait for run to return. If a heartbeat query was running, run reported status after the query returned, which could be after the monitor loader removed the monitor's status. The removed monitor then reappeared in the monitor status with one stale heartbeat-reader component. Stop now waits for run to return, without holding the lock that run uses to update lag. That takes up to ReadTimeout if a query is running. Start doesn't start a stopped reader, so Stop doesn't wait for a reader that never started, which happens when repl.lag stops the reader before its Start goroutine runs. TestReader now stops its reader. It leaked the reader, which raced with the next run's ReadErrorWait write under -count. Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 1 + heartbeat/heartbeat_test.go | 1 + heartbeat/reader.go | 27 ++++++- heartbeat/reader_stop_test.go | 108 ++++++++++++++++++++++++++++ 4 files changed, 135 insertions(+), 2 deletions(-) create mode 100644 heartbeat/reader_stop_test.go diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 7723f15..abe3e52 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -62,6 +62,7 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * 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` and `report-not-a-replica` now apply at every level, not only at one random level, and the heartbeat reader now uses `repl-check`. Levels that don't set `writer`, or set it to `auto`, use the writer set at another level. Plans that set `writer` to `pfs` at one level and `blip` at another now fail to prepare. +* 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) diff --git a/heartbeat/heartbeat_test.go b/heartbeat/heartbeat_test.go index e167ddd..1bc3789 100644 --- a/heartbeat/heartbeat_test.go +++ b/heartbeat/heartbeat_test.go @@ -204,6 +204,7 @@ func TestReader(t *testing.T) { Waiter: mockWaiter, }) hr.Start() + defer hr.Stop() // before ReadErrorWait is reset: Stop waits for the reader to return timeout := time.After(5 * time.Second) // this whole test should take <1s diff --git a/heartbeat/reader.go b/heartbeat/reader.go index 725b7b8..28f69fd 100644 --- a/heartbeat/reader.go +++ b/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) { diff --git a/heartbeat/reader_stop_test.go b/heartbeat/reader_stop_test.go new file mode 100644 index 0000000..da6057d --- /dev/null +++ b/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") + } +} From e868f64010fcb94c6e812f3631ab730e89170678 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Wed, 28 Oct 2026 13:23:19 -0700 Subject: [PATCH 4/8] Document repl.lag single-level reader options and mixed-writer change Prepare reads the writer and heartbeat reader options (table, source-id, source-role, repl-check, network-latency) from one level only. Say so in the Prepare doc comment and the release note. List rejecting pfs at one level and blip at another as a behavior change: those plans prepared before, and now the monitor keeps retrying them. TestReader's mock waiter blocked on a full channel with no way out, so the deferred Stop, which now waits for the reader, hung the test until the go test timeout if the test failed while the reader was sending. The send now also ends when the test returns. Co-Authored-By: Claude Opus 5.5 --- docs/content/about/release-notes.md | 3 ++- heartbeat/heartbeat_test.go | 9 +++++++-- metrics/repl.lag/lag.go | 6 ++++-- 3 files changed, 13 insertions(+), 5 deletions(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index abe3e52..b4f0087 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -61,7 +61,8 @@ 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` and `report-not-a-replica` now apply at every level, not only at one random level, and the heartbeat reader now uses `repl-check`. Levels that don't set `writer`, or set it to `auto`, use the writer set at another level. Plans that set `writer` to `pfs` at one level and `blip` at another now fail to prepare. +* Fixed `repl.lag` at multiple levels with the Blip heartbeat writer: `report-no-heartbeat` and `report-not-a-replica` now apply at every level, not only at one random level, and the heartbeat reader now uses `repl-check`. Levels that don't set `writer`, or set it to `auto`, use the writer set at another level. The heartbeat reader options `table`, `source-id`, `source-role`, `repl-check`, 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:** 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) diff --git a/heartbeat/heartbeat_test.go b/heartbeat/heartbeat_test.go index 1bc3789..9c289b5 100644 --- a/heartbeat/heartbeat_test.go +++ b/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,7 +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 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 diff --git a/metrics/repl.lag/lag.go b/metrics/repl.lag/lag.go index 25b15b4..abeb544 100644 --- a/metrics/repl.lag/lag.go +++ b/metrics/repl.lag/lag.go @@ -160,8 +160,10 @@ func (c *Lag) Help() blip.CollectorHelp { // writer option to pfs or blip; else the first level that collects lag, which // auto-detects the writer. Levels that don't set the writer, or set it to // auto, use that writer. It's an error if one level sets pfs and another sets -// blip. The report-no-heartbeat, report-not-a-replica, and -// default-channel-name options are per level. +// blip. The writer and heartbeat reader options (table, source-id, +// source-role, repl-check, network-latency) are read only from that level. +// The report-no-heartbeat, report-not-a-replica, and default-channel-name +// options are per level. // // If lag is collected from a Blip heartbeat, the returned cleanup func stops // the heartbeat reader; else it's nil. On error, Prepare stops the reader From 820396c59a3cce2fc9cd441fed9e23e85d566c62 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Wed, 28 Oct 2026 13:40:58 -0700 Subject: [PATCH 5/8] Document repl.lag single-level options on the domain page List which repl.lag options are read from one level and which are per level on the repl.lag docs page, and describe repl-check as a writer option since it also applies to the pfs writer. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Ian Oberst --- docs/content/about/release-notes.md | 2 +- docs/content/metrics/domains/repl.lag/_index.md | 4 ++++ metrics/repl.lag/lag.go | 4 ++-- 3 files changed, 7 insertions(+), 3 deletions(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index b4f0087..e904e71 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -61,7 +61,7 @@ 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` and `report-not-a-replica` now apply at every level, not only at one random level, and the heartbeat reader now uses `repl-check`. Levels that don't set `writer`, or set it to `auto`, use the writer set at another level. The heartbeat reader options `table`, `source-id`, `source-role`, `repl-check`, 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`. +* Fixed `repl.lag` at multiple levels with the Blip heartbeat writer: `report-no-heartbeat` and `report-not-a-replica` now apply at every level, not only at one random level, and the heartbeat reader now uses `repl-check`. Levels that don't set `writer`, or set it to `auto`, use the writer set at another level. 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:** 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" >}}). diff --git a/docs/content/metrics/domains/repl.lag/_index.md b/docs/content/metrics/domains/repl.lag/_index.md index 3e5a10c..5c393e5 100644 --- a/docs/content/metrics/domains/repl.lag/_index.md +++ b/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` diff --git a/metrics/repl.lag/lag.go b/metrics/repl.lag/lag.go index abeb544..5972dbd 100644 --- a/metrics/repl.lag/lag.go +++ b/metrics/repl.lag/lag.go @@ -160,8 +160,8 @@ func (c *Lag) Help() blip.CollectorHelp { // writer option to pfs or blip; else the first level that collects lag, which // auto-detects the writer. Levels that don't set the writer, or set it to // auto, use that writer. It's an error if one level sets pfs and another sets -// blip. The writer and heartbeat reader options (table, source-id, -// source-role, repl-check, network-latency) are read only from that level. +// blip. The writer and its options (repl-check, table, source-id, +// source-role, network-latency) are read only from that level. // The report-no-heartbeat, report-not-a-replica, and default-channel-name // options are per level. // From 3a4f01f78b5f77ba2388284bfff498d94dd95884 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Wed, 28 Oct 2026 14:35:22 -0700 Subject: [PATCH 6/8] Note that the repl.lag heartbeat reader now uses repl-check The heartbeat reader ignored repl-check in every plan, not only in plans that collect repl.lag at more than one level, so using it is a behavior change for single-level plans too. Move it out of the multiple-levels note into its own behavior change note. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Ian Oberst --- docs/content/about/release-notes.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index e904e71..095751c 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -61,7 +61,8 @@ 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` and `report-not-a-replica` now apply at every level, not only at one random level, and the heartbeat reader now uses `repl-check`. Levels that don't set `writer`, or set it to `auto`, use the writer set at another level. 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`. +* Fixed `repl.lag` at multiple levels with the Blip heartbeat writer: `report-no-heartbeat` and `report-not-a-replica` now apply 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. 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" >}}). From e4ef50182fcd6eac0b50c3a41635ccdde61a95b7 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 29 Oct 2026 06:51:37 -0700 Subject: [PATCH 7/8] Scope the repl.lag multiple-levels note to report-no-heartbeat report-not-a-replica was already set at every level: only report-no-heartbeat was set at one level. With the Blip writer, report-not-a-replica had no effect because the reader ignored repl-check, which the repl-check behavior change note covers. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Ian Oberst --- docs/content/about/release-notes.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 095751c..f6b6763 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -61,7 +61,7 @@ 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` and `report-not-a-replica` now apply 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. 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`. +* 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. 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" >}}). From 5777c64db8fed2b686d0522605b2461f160171f5 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 29 Oct 2026 07:11:44 -0700 Subject: [PATCH 8/8] Note what repl.lag levels without a writer did before Levels that don't set writer, or set it to auto, used to auto-detect their own writer, so one could use pfs while another used blip. Say so in the release note, next to the new rule. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Ian Oberst --- docs/content/about/release-notes.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index f6b6763..10fca47 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -61,7 +61,7 @@ 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. 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`. +* 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" >}}).