diff --git a/CHANGELOG.md b/CHANGELOG.md index 598ee9c83d..23043164e0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2611,6 +2611,19 @@ never reused, per the v1.1.1 precedent.) not to the pool alias it was dispatched to. Off by default: with no `claim.pools` configured, behavior is unchanged. +- **`bd monitor-commit-rate` — local backstop for the no-op-commit storm + (vp-5u7i deliverable 2, ADR-0023 L-A).** Samples `dolt_diff_issues` for the + current database over a trailing window (`--minutes`, default 1) and + reports the storm signature: no-op commits (`from_content_hash == + to_content_hash`) at or above `--commits` (default 10), concentrated on a + disproportionately small set of beads. This is independent of the no-op + gate above — it samples committed history directly, so it also catches a + recurrence from any writer, not just `bd`. Alert-only by default + (`--alert-only`, on by default; the `--alert-only=false` action path is + unimplemented in this release — see the flag's help text); exits 2 on + alert, 0 clean, so a caller can branch on the exit code without parsing + output. Intended to run on a schedule per rig/database (a Voxist city-pack + order does this outside this repo). - **Work leases: claim-TTL, heartbeat, and reclaim for dead-worker recovery** (schema v54, migration `0054`) ([#4537](https://github.com/gastownhall/beads/pull/4537)). A claim was diff --git a/cmd/bd/main.go b/cmd/bd/main.go index 57bd7d6b3b..8020792a0e 100644 --- a/cmd/bd/main.go +++ b/cmd/bd/main.go @@ -2315,6 +2315,9 @@ func main() { rootCmd.InitDefaultHelpCmd() registerHelpAllFlag() + // Add the monitor commit rate command + rootCmd.AddCommand(monitorCommitRateCmd) + executedCmd, err := rootCmd.ExecuteC() // Let this command's fire-and-forget hooks finish, for the same diff --git a/cmd/bd/monitor_commit_rate.go b/cmd/bd/monitor_commit_rate.go new file mode 100644 index 0000000000..e0f2220921 --- /dev/null +++ b/cmd/bd/monitor_commit_rate.go @@ -0,0 +1,273 @@ +package main + +import ( + "context" + "database/sql" + "fmt" + "os" + "strings" + "time" + + "github.com/spf13/cobra" + "github.com/steveyegge/beads/internal/storage" +) + +var ( + alertThresholdMin int + alertThresholdCommits int + alertOnly bool + dryRun bool +) + +func init() { + monitorCommitRateCmd.Flags().IntVar(&alertThresholdMin, "minutes", 1, "Time window in minutes to monitor for commit rate") + monitorCommitRateCmd.Flags().IntVar(&alertThresholdCommits, "commits", 10, "Alert threshold: number of no-op commits in the time window that triggers an alert") + monitorCommitRateCmd.Flags().BoolVar(&alertOnly, "alert-only", true, "Only alert when thresholds are exceeded (don't auto-take action)") + monitorCommitRateCmd.Flags().BoolVar(&dryRun, "dry-run", false, "Perform a dry run without taking any actions") +} + +var monitorCommitRateCmd = &cobra.Command{ + Use: "monitor-commit-rate", + Short: "Monitor Dolt commit rate and alert on excessive no-op commits (vp-5u7i deliverable 2)", + Long: `Monitor Dolt commit rate to detect excessive no-op commits that indicate the storm pattern. + +Samples dolt_diff_issues for the current database and alerts when the no-op +commit count in the time window is at or above the threshold AND those +commits are concentrated on a disproportionately small set of beads +(signature: many commits, few beads, identical content_hash — ADR-0023 L-A). +This is the local backstop that catches a recurrence independent of bd's own +write-path gate (e.g. a stray script or supervisor writing directly). + +This implements deliverable 2 from bead vp-5u7i.`, + SilenceUsage: true, + SilenceErrors: true, + RunE: func(cmd *cobra.Command, args []string) error { + return runMonitorCommitRate() + }, +} + +func runMonitorCommitRate() error { + if dryRun { + fmt.Printf("DRY RUN: Monitoring commit rate with threshold of %d no-op commits per %d minutes\n", alertThresholdCommits, alertThresholdMin) + return nil + } + + if store == nil { + return HandleErrorRespectJSON("no database connection available (%s)", diagHint()) + } + accessor, ok := storage.UnwrapStore(store).(storage.RawDBAccessor) + if !ok { + return HandleErrorRespectJSON("storage backend does not support raw DB access required for commit-rate monitoring") + } + db := accessor.UnderlyingDB() + if db == nil { + return HandleErrorRespectJSON("underlying database not available") + } + + ctx := rootCtx + var dbName string + if err := db.QueryRowContext(ctx, "SELECT DATABASE()").Scan(&dbName); err != nil { + return HandleErrorRespectJSON("resolving current database name: %v", err) + } + + result, err := analyzeCommitPatterns(ctx, db, dbName, alertThresholdMin, alertThresholdCommits) + if err != nil { + return HandleErrorRespectJSON("failed to analyze commit patterns: %v", err) + } + + if result.ExcessiveNoOpsDetected { + fmt.Fprintf(os.Stderr, "ALERT: Excessive no-op commit pattern detected!\n") + fmt.Fprintf(os.Stderr, " Database: %s\n", result.DatabaseName) + fmt.Fprintf(os.Stderr, " No-op commits: %d\n", result.CommitCount) + fmt.Fprintf(os.Stderr, " Distinct beads affected: %d\n", result.DistinctBeadCount) + fmt.Fprintf(os.Stderr, " Time window: %v\n", result.TimeWindow) + fmt.Fprintf(os.Stderr, " Content hash similarity: %.2f%% identical\n", result.PercentIdenticalContent) + + if !alertOnly { + fmt.Fprintf(os.Stderr, "Taking corrective action (disabled in this implementation)...\n") + } + // Exit code 2 (Nagios-style: 0 clean, 1 command error, 2 alert) gives + // a caller (the commit-rate-watchdog city-pack order, one invocation + // per rig) a reliable machine-readable signal without parsing stderr. + return &exitError{Code: 2} + } + + fmt.Printf("No excessive commit patterns detected within threshold (database: %s).\n", result.DatabaseName) + return nil +} + +// CommitAnalysisResult holds the results of a commit-rate analysis pass. +// CommitCount and DistinctBeadCount describe only the no-op subset of the +// commits examined in the window — the counts the storm signature is judged +// against — not the window's total commit volume. +type CommitAnalysisResult struct { + ExcessiveNoOpsDetected bool + DatabaseName string + CommitCount int // no-op commits in the window + DistinctBeadCount int // distinct beads touched by those no-op commits + TimeWindow time.Duration + PercentIdenticalContent float64 // % of all commits examined that were no-op +} + +// minCommitsPerBeadRatio is the "flat distinct-bead count" leg of the storm +// signature (ADR-0023 L-A): the measured incident averaged ~13 no-op commits +// per bead (va-wzio alone took 10 in ~2s). Organic churn averages close to +// one commit per bead, so requiring at least this many no-op commits per +// affected bead separates a storm (the same small set hammered repeatedly) +// from a legitimate burst of first-time no-op touches across many beads. +const minCommitsPerBeadRatio = 2.0 + +// analyzeCommitPatterns samples dolt_diff_issues for the current database and +// reports whether the no-op-commit storm signature (ADR-0023 L-A) is present +// in the trailing windowMinutes: no-op commit volume at or above +// alertThreshold, concentrated on a disproportionately small set of beads, +// every one of them value-identical (from_content_hash == to_content_hash). +// A commit counts as no-op only when every row it touched compares +// content-identical; any real content change (or an added row, which has no +// "from" state) marks the whole commit as real. +// noOpDiffExcludedColumns are the issues columns whose change does NOT make a +// diff row meaningful, so they are left out of the value comparison below. +// +// - updated_at is rewritten by every update, including one that changes +// nothing else — it is the thing a no-op commit consists of. +// - row_lock is rewritten by any write that can touch status/assignee, so a +// concurrent reclaim/close collides on the row (see freshRowLock); its +// change carries no user-visible content. +// - content_hash is NOT MAINTAINED on the update path. It is written by the +// upsert/import path only (issueUpsertColumns in issueops/helpers.go) and +// never recomputed by UpdateIssue, so from_content_hash == to_content_hash +// holds for every ordinary edit. Comparing it — which this command used to +// do, and only that — reported every real update as a no-op. +var noOpDiffExcludedColumns = map[string]bool{ + "updated_at": true, + "row_lock": true, + "content_hash": true, +} + +// noOpDiffPredicate builds the SQL that decides whether one dolt_diff_issues +// row changed anything meaningful. +// +// The column list is discovered from information_schema rather than hardcoded: +// the issues table gains columns regularly upstream, and a stale hardcoded list +// fails OPEN — an unlisted column's change would be invisible and the row would +// count as a no-op. Discovery makes a new column meaningful by default, which +// is the safe direction to be wrong in. +// +// <=> is MySQL's NULL-safe equality: NULL <=> NULL is true and NULL <=> 'x' is +// false, which is what "unchanged" means for a nullable column. +func noOpDiffPredicate(ctx context.Context, db *sql.DB, dbName string) (string, error) { + rows, err := db.QueryContext(ctx, ` + SELECT column_name + FROM information_schema.columns + WHERE table_schema = ? AND table_name = 'issues' + ORDER BY ordinal_position + `, dbName) + if err != nil { + return "", fmt.Errorf("reading issues columns: %w", err) + } + defer rows.Close() + + var terms []string + for rows.Next() { + var col string + if err := rows.Scan(&col); err != nil { + return "", fmt.Errorf("scanning column name: %w", err) + } + if noOpDiffExcludedColumns[col] { + continue + } + terms = append(terms, fmt.Sprintf("`from_%s` <=> `to_%s`", col, col)) + } + if err := rows.Err(); err != nil { + return "", fmt.Errorf("reading issues columns: %w", err) + } + if len(terms) == 0 { + return "", fmt.Errorf("issues table reported no comparable columns") + } + return strings.Join(terms, " AND "), nil +} + +func analyzeCommitPatterns(ctx context.Context, db *sql.DB, dbName string, windowMinutes, alertThreshold int) (*CommitAnalysisResult, error) { + unchanged, err := noOpDiffPredicate(ctx, db, dbName) + if err != nil { + return nil, err + } + + // Compute the cutoff in Go rather than relying on the server's NOW() with + // a bound INTERVAL — go-mysql-server's parser support for a placeholder + // inside INTERVAL is untested, whereas a literal DATETIME comparison is + // universally supported and keeps the window's meaning explicit. + cutoff := time.Now().UTC().Add(-time.Duration(windowMinutes) * time.Minute) + rows, err := db.QueryContext(ctx, fmt.Sprintf(` + SELECT to_commit, + COALESCE(to_id, '') AS to_id, + from_id IS NOT NULL AS is_update, + (%s) AS unchanged + FROM dolt_diff_issues + WHERE to_commit_date > ? + AND to_id IS NOT NULL + `, unchanged), cutoff) + if err != nil { + return nil, fmt.Errorf("querying dolt_diff_issues: %w", err) + } + defer rows.Close() + + type commitAgg struct { + allNoOp bool + beads map[string]struct{} + } + commits := make(map[string]*commitAgg) + for rows.Next() { + var commitHash, id string + var isUpdate, rowUnchanged bool + if err := rows.Scan(&commitHash, &id, &isUpdate, &rowUnchanged); err != nil { + return nil, fmt.Errorf("scanning dolt_diff_issues row: %w", err) + } + if id == "" { + continue + } + agg, ok := commits[commitHash] + if !ok { + agg = &commitAgg{allNoOp: true, beads: make(map[string]struct{})} + commits[commitHash] = agg + } + agg.beads[id] = struct{}{} + // An INSERT is never a no-op: it added a bead that was not there. Only + // a modification whose every compared column is unchanged qualifies. + if !isUpdate || !rowUnchanged { + agg.allNoOp = false + } + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("reading dolt_diff_issues: %w", err) + } + + noOpCommits := 0 + noOpBeads := make(map[string]struct{}) + for _, agg := range commits { + if agg.allNoOp { + noOpCommits++ + for id := range agg.beads { + noOpBeads[id] = struct{}{} + } + } + } + + result := &CommitAnalysisResult{ + DatabaseName: dbName, + CommitCount: noOpCommits, + DistinctBeadCount: len(noOpBeads), + TimeWindow: time.Duration(windowMinutes) * time.Minute, + } + if len(commits) > 0 { + result.PercentIdenticalContent = 100 * float64(noOpCommits) / float64(len(commits)) + } + + beadsPerCommit := 0.0 + if len(noOpBeads) > 0 { + beadsPerCommit = float64(noOpCommits) / float64(len(noOpBeads)) + } + result.ExcessiveNoOpsDetected = noOpCommits >= alertThreshold && beadsPerCommit >= minCommitsPerBeadRatio + + return result, nil +} diff --git a/cmd/bd/monitor_commit_rate_embedded_test.go b/cmd/bd/monitor_commit_rate_embedded_test.go new file mode 100644 index 0000000000..9f52523064 --- /dev/null +++ b/cmd/bd/monitor_commit_rate_embedded_test.go @@ -0,0 +1,150 @@ +//go:build cgo + +package main + +import ( + "context" + "database/sql" + "fmt" + "os" + "path/filepath" + "testing" + "time" + + "github.com/steveyegge/beads/internal/configfile" + "github.com/steveyegge/beads/internal/storage/embeddeddolt" +) + +// These tests pin the commit-rate watchdog (vp-5u7i deliverable 2, ADR-0023 +// L-A local backstop): analyzeCommitPatterns must fire on a synthetic no-op +// commit burst regardless of its source — the write-path gate (see +// update_noop_gate_embedded_test.go) covers bd's own no-op updates, so this +// backstop is deliberately exercised via raw SQL that bypasses bd entirely, +// the same way an unrelated writer (a stray script, a supervisor runaway) +// would produce the storm. It must also stay quiet on ordinary activity +// spread across many beads. + +func openEmbeddedRawDB(t *testing.T, beadsDir string) *sql.DB { + t.Helper() + dataDir := filepath.Join(beadsDir, "embeddeddolt") + cfg, _ := configfile.Load(beadsDir) + database := "" + if cfg != nil { + database = cfg.GetDoltDatabase() + } + db, cleanup, err := embeddeddolt.OpenSQL(t.Context(), dataDir, database, "main") + if err != nil { + t.Fatalf("OpenSQL: %v", err) + } + t.Cleanup(func() { _ = cleanup() }) + return db +} + +// commitNoOpChange bumps updated_at only — content_hash untouched — then +// commits it, reproducing the pre-gate storm shape (real row diff, identical +// content_hash) directly at the SQL layer. +func commitNoOpChange(t *testing.T, db *sql.DB, id string, at time.Time) { + t.Helper() + ctx := context.Background() + if _, err := db.ExecContext(ctx, "UPDATE issues SET updated_at = ? WHERE id = ?", at, id); err != nil { + t.Fatalf("synthetic no-op update: %v", err) + } + if _, err := db.ExecContext(ctx, "CALL DOLT_COMMIT('-Am', ?)", fmt.Sprintf("bd: update %s", id)); err != nil { + t.Fatalf("synthetic no-op commit: %v", err) + } +} + +func currentDatabase(t *testing.T, db *sql.DB) string { + t.Helper() + var name string + if err := db.QueryRowContext(t.Context(), "SELECT DATABASE()").Scan(&name); err != nil { + t.Fatalf("SELECT DATABASE(): %v", err) + } + return name +} + +func TestAnalyzeCommitPatternsDetectsNoOpStorm(t *testing.T) { + if os.Getenv("BEADS_TEST_EMBEDDED_DOLT") != "1" { + t.Skip("set BEADS_TEST_EMBEDDED_DOLT=1 to run embedded dolt monitor tests") + } + t.Parallel() + + bd := buildEmbeddedBD(t) + dir, beadsDir, _ := bdInit(t, bd, "--prefix", "mc") + issue := bdCreate(t, bd, dir, "Storm target", "--type", "task") + + db := openEmbeddedRawDB(t, beadsDir) + base := time.Now().UTC() + for i := 0; i < 12; i++ { + commitNoOpChange(t, db, issue.ID, base.Add(time.Duration(i)*time.Second)) + } + + result, err := analyzeCommitPatterns(t.Context(), db, currentDatabase(t, db), 5, 10) + if err != nil { + t.Fatalf("analyzeCommitPatterns: %v", err) + } + if !result.ExcessiveNoOpsDetected { + t.Fatalf("want storm detected, got %+v", result) + } + if result.DistinctBeadCount != 1 { + t.Fatalf("DistinctBeadCount = %d, want 1 (all 12 no-op commits hit the same bead)", result.DistinctBeadCount) + } + if result.CommitCount < 10 { + t.Fatalf("CommitCount = %d, want >= 10", result.CommitCount) + } +} + +func TestAnalyzeCommitPatternsIgnoresDistinctContentAcrossManyBeads(t *testing.T) { + if os.Getenv("BEADS_TEST_EMBEDDED_DOLT") != "1" { + t.Skip("set BEADS_TEST_EMBEDDED_DOLT=1 to run embedded dolt monitor tests") + } + t.Parallel() + + bd := buildEmbeddedBD(t) + dir, beadsDir, _ := bdInit(t, bd, "--prefix", "mn") + + for i := 0; i < 5; i++ { + issue := bdCreate(t, bd, dir, fmt.Sprintf("Normal bead %d", i), "--type", "task") + bdUpdate(t, bd, dir, issue.ID, "--title", fmt.Sprintf("Normal bead %d renamed", i)) + } + + db := openEmbeddedRawDB(t, beadsDir) + result, err := analyzeCommitPatterns(t.Context(), db, currentDatabase(t, db), 5, 10) + if err != nil { + t.Fatalf("analyzeCommitPatterns: %v", err) + } + if result.ExcessiveNoOpsDetected { + t.Fatalf("want no storm on distinct-content activity across many beads, got %+v", result) + } + if result.CommitCount != 0 { + t.Fatalf("CommitCount = %d, want 0 (every commit changed real content)", result.CommitCount) + } +} + +func TestAnalyzeCommitPatternsBelowThresholdNoAlert(t *testing.T) { + if os.Getenv("BEADS_TEST_EMBEDDED_DOLT") != "1" { + t.Skip("set BEADS_TEST_EMBEDDED_DOLT=1 to run embedded dolt monitor tests") + } + t.Parallel() + + bd := buildEmbeddedBD(t) + dir, beadsDir, _ := bdInit(t, bd, "--prefix", "mb") + issue := bdCreate(t, bd, dir, "Small burst", "--type", "task") + + db := openEmbeddedRawDB(t, beadsDir) + base := time.Now().UTC() + for i := 0; i < 3; i++ { + commitNoOpChange(t, db, issue.ID, base.Add(time.Duration(i)*time.Second)) + } + + result, err := analyzeCommitPatterns(t.Context(), db, currentDatabase(t, db), 5, 10) + if err != nil { + t.Fatalf("analyzeCommitPatterns: %v", err) + } + if result.ExcessiveNoOpsDetected { + t.Fatalf("want no alert below the commit-count threshold, got %+v", result) + } + if result.CommitCount != 3 { + t.Fatalf("CommitCount = %d, want 3", result.CommitCount) + } +}