*: add metric with sync message cohorts

app/health/checker.go

@@ -20,8 +20,10 @@
 	scrapePeriod = 30 * time.Second
 	// maxScrapes is the maximum number of scrapes to keep (5 minutes at 30s intervals).
 	maxScrapes = 10
-	// maxSSEScrapes is the number of scrapes kept for the SSE head delay check (1 hour at 30s intervals).
-	maxSSEScrapes = 120
+	// maxLongScrapes is the number of scrapes kept in the long-window buffer used by MetricsFunc
+	// checks that compute rates over a longer history, e.g. SSE head delay and sync message
+	// disagreement (1 hour at 30s intervals).
+	maxLongScrapes = 120
 	// seriesCardinalityThreshold is the maximum number of time series per metric family
 	// for a single validator; for N validators, the threshold is N * seriesCardinalityThreshold.
 	seriesCardinalityThreshold = 100
@@ -56,7 +58,7 @@
 		gatherer:           gatherer,
 		scrapePeriod:       scrapePeriod,
 		maxScrapes:         maxScrapes,
-		maxSSEScrapes:      maxSSEScrapes,
+		maxLongScrapes:     maxLongScrapes,
 		logFilter:          log.Filter(),
 		numValidators:      numValidators,
 		memorySamplePeriod: memorySamplePeriod,
@@ -66,14 +68,16 @@
 
 // Checker is a health checker.
 type Checker struct {
-	metadata           Metadata
-	checks             []check
-	metrics            [][]*pb.MetricFamily
-	sseMetrics         [][]*pb.MetricFamily
+	metadata Metadata
+	checks   []check
+	// metrics is the short-window scrape buffer (maxScrapes) used by query-based Func checks.
+	metrics [][]*pb.MetricFamily
+	// longMetrics is the long-window scrape buffer (maxLongScrapes) passed to all MetricsFunc checks.
+	longMetrics        [][]*pb.MetricFamily
 	gatherer           prometheus.Gatherer
 	scrapePeriod       time.Duration
 	maxScrapes         int
-	maxSSEScrapes      int
+	maxLongScrapes     int
 	logFilter          z.Field
 	numValidators      int
 	memorySnapshots    []memorySnapshot
@@ -120,7 +124,7 @@
 		case check.MemFunc != nil:
 			failing, err = check.MemFunc(c.memorySnapshots, c.metadata)
 		case check.MetricsFunc != nil:
-			failing, err = check.MetricsFunc(c.sseMetrics, c.metadata)
+			failing, err = check.MetricsFunc(c.longMetrics, c.metadata)
 		default:
 			failing, err = check.Func(newQueryFunc(c.metrics), c.metadata)
 		}
@@ -176,9 +180,9 @@
 		c.metrics = c.metrics[1:]
 	}
 
-	c.sseMetrics = append(c.sseMetrics, metrics)
-	if len(c.sseMetrics) > c.maxSSEScrapes {
-		c.sseMetrics = c.sseMetrics[1:]
+	c.longMetrics = append(c.longMetrics, metrics)
+	if len(c.longMetrics) > c.maxLongScrapes {
+		c.longMetrics = c.longMetrics[1:]
 	}
 
 	return nil
@@ -302,6 +306,81 @@
 	return false, nil
 }
 
+// syncMsgDisagreementCheck returns true if any peer signed a different sync committee head block
+// root than the largest cohort for more than syncMsgDisagreementThreshold of the sync messages it
+// submitted during the scrape window. It computes the per-peer increase of the cohort rank counter
+// (rank!="0" = disagreed, all ranks = total) across the window, mirroring the Grafana panel.
+func syncMsgDisagreementCheck(scrapes [][]*pb.MetricFamily, _ Metadata) (bool, error) {
+	const (
+		metricName = "core_tracker_parsig_cohort_rank_total"
+		peerLabel  = "peer_idx"
+		rankLabel  = "rank"
+		threshold  = 0.05
+		// minSamples guards against noisy ratios for peers that barely participated in the window.
+		// MetricsFunc checks run over the ~1-hour SSE scrape buffer (~300 sync slots), so 20 is a
+		// low participation floor that mainly rejects startup and mostly-offline peers.
+		minSamples = 20
+	)
+
+	if len(scrapes) < 2 {
+		return false, nil
+	}
+
+	type counts struct{ total, disagreed float64 }
+
+	collect := func(fams []*pb.MetricFamily) map[string]counts {
+		result := make(map[string]counts)
+
+		for _, fam := range fams {
+			if fam.GetName() != metricName {
+				continue
+			}
+
+			for _, metric := range fam.GetMetric() {
+				var peerIdx, rank string
+
+				for _, lbl := range metric.GetLabel() {
+					switch lbl.GetName() {
+					case peerLabel:
+						peerIdx = lbl.GetValue()
+					case rankLabel:
+						rank = lbl.GetValue()
+					default:
+					}
+				}
+
+				c := result[peerIdx]
+
+				c.total += metric.GetCounter().GetValue()
+				if rank != "0" {
+					c.disagreed += metric.GetCounter().GetValue()
+				}
+
+				result[peerIdx] = c
+			}
+		}
+
+		return result
+	}
+
+	first := collect(scrapes[0])
+	last := collect(scrapes[len(scrapes)-1])
+
+	for peerIdx, lastCounts := range last {
+		deltaTotal := lastCounts.total - first[peerIdx].total
+		if deltaTotal < minSamples {
+			continue
+		}
+
+		deltaDisagreed := lastCounts.disagreed - first[peerIdx].disagreed
+		if deltaDisagreed/deltaTotal > threshold {
+			return true, nil
+		}
+	}
+
+	return false, nil
+}
+
 // memoryLeakCheck returns true if average memory in the most recent 24h has grown by more than
 // memoryLeakThreshold compared to the previous 24h, ignoring samples taken within
 // memoryWarmupPeriod of a restart.

app/health/checks.go

@@ -347,6 +347,16 @@ var checks = []check{
 			return maxAvg > 3.0, nil // 3s threshold
 		},
 	},
+	{
+		Name: "sync_message_head_disagreement",
+		Description: `A peer signed a different sync committee head block root than the rest of the cluster for >5% of sync messages in the past hour.
+		Sync messages don't go through consensus: each node signs the head its own beacon node reports, so a peer that consistently disagrees is likely running a beacon node that is lagging, on a minority fork, or otherwise out of sync.
+		This is not slashable and only costs sync committee rewards for the affected slots, but persistent disagreement points at an unhealthy beacon node.
+
+		Identify the drifting peer via core_tracker_parsig_cohort_rank_total (by peer_idx) and the "Sync committee head disagreement" logs, then coordinate with that operator to check their beacon node's sync status and peer count. If it is the local node, check this node's beacon node.`,
+		Severity:    severityWarning,
+		MetricsFunc: syncMsgDisagreementCheck,
+	},
 	{
 		Name:        "high_goroutine_count",
 		Description: `Goroutine count exceeds 1000. Possible leak. Report to Obol technical team.`,

app/health/checks_internal_test.go

@@ -1091,3 +1091,77 @@ func genGauge(labels []*pb.LabelPair, values ...int) []*pb.Metric {
 
 	return resp
 }
+
+func TestSyncMessageHeadDisagreementCheck(t *testing.T) {
+	// makeScrapes builds a two-scrape counter history for one peer's sync cohort ranks.
+	// total/disagreed are cumulative counts in scrape[0]; delta values are added in scrape[1].
+	// Disagreements are modelled as a rank="1" series, agreements as rank="0".
+	makeScrapes := func(peerIdx string, firstTotal, firstDisagreed, deltaTotal, deltaDisagreed float64) [][]*pb.MetricFamily {
+		mkFam := func(total, disagreed float64) []*pb.MetricFamily {
+			name := "core_tracker_parsig_cohort_rank_total"
+			typ := pb.MetricType_COUNTER
+			rank0 := total - disagreed
+
+			return []*pb.MetricFamily{{
+				Name: &name,
+				Type: &typ,
+				Metric: []*pb.Metric{
+					{Label: []*pb.LabelPair{l("peer_idx", peerIdx), l("rank", "0")}, Counter: &pb.Counter{Value: &rank0}},
+					{Label: []*pb.LabelPair{l("peer_idx", peerIdx), l("rank", "1")}, Counter: &pb.Counter{Value: &disagreed}},
+				},
+			}}
+		}
+
+		return [][]*pb.MetricFamily{
+			mkFam(firstTotal, firstDisagreed),
+			mkFam(firstTotal+deltaTotal, firstDisagreed+deltaDisagreed),
+		}
+	}
+
+	t.Run("no data", func(t *testing.T) {
+		failing, err := syncMsgDisagreementCheck(nil, Metadata{})
+		require.NoError(t, err)
+		require.False(t, failing)
+	})
+
+	t.Run("single scrape only", func(t *testing.T) {
+		failing, err := syncMsgDisagreementCheck([][]*pb.MetricFamily{{}}, Metadata{})
+		require.NoError(t, err)
+		require.False(t, failing)
+	})
+
+	t.Run("full agreement", func(t *testing.T) {
+		// 25 sync messages, none disagreed.
+		failing, err := syncMsgDisagreementCheck(makeScrapes("6", 0, 0, 25, 0), Metadata{})
+		require.NoError(t, err)
+		require.False(t, failing)
+	})
+
+	t.Run("below threshold", func(t *testing.T) {
+		// 100 messages, 3 disagreed -> 3%.
+		failing, err := syncMsgDisagreementCheck(makeScrapes("6", 0, 0, 100, 3), Metadata{})
+		require.NoError(t, err)
+		require.False(t, failing)
+	})
+
+	t.Run("above threshold triggers warning", func(t *testing.T) {
+		// 100 messages, 8 disagreed -> 8%.
+		failing, err := syncMsgDisagreementCheck(makeScrapes("6", 0, 0, 100, 8), Metadata{})
+		require.NoError(t, err)
+		require.True(t, failing)
+	})
+
+	t.Run("insufficient samples does not trigger", func(t *testing.T) {
+		// 10 messages, 5 disagreed -> 50% but below minSamples.
+		failing, err := syncMsgDisagreementCheck(makeScrapes("6", 0, 0, 10, 5), Metadata{})
+		require.NoError(t, err)
+		require.False(t, failing)
+	})
+
+	t.Run("historical dilution does not mask recent spike", func(t *testing.T) {
+		// Large clean history, then 100 recent messages with 8 disagreed -> still 8% in window.
+		failing, err := syncMsgDisagreementCheck(makeScrapes("6", 10000, 0, 100, 8), Metadata{})
+		require.NoError(t, err)
+		require.True(t, failing)
+	})
+}

core/tracker/metrics.go

@@ -89,6 +89,13 @@ var (
 		Help:      "Total number of duties that contained inconsistent partial signed data by duty type",
 	}, []string{"duty"})
 
+	cohortRank = promauto.NewCounterVec(prometheus.CounterOpts{
+		Namespace: "core",
+		Subsystem: "tracker",
+		Name:      "parsig_cohort_rank_total",
+		Help:      "Total sync committee partial signatures per peer per cohort rank (0=largest cohort), for detecting head disagreement",
+	}, []string{"duty", "peer_idx", "rank"})
+
 	inclusionDelay = promauto.NewGauge(prometheus.GaugeOpts{
 		Namespace: "core",
 		Subsystem: "tracker",