*: add metric with sync message cohorts

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",

core/tracker/tracker.go

@@ -3,10 +3,13 @@
 package tracker
 
 import (
+	"bytes"
 	"context"
 	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"slices"
+	"strconv"
 
 	eth2api "github.com/attestantio/go-eth2-client/api"
 
@@ -290,7 +293,8 @@ func analyseDutyFailed(duty core.Duty, allEvents map[core.Duty][]event, msgRootC
 			reason = reasonInsufficientPeerSignatures
 		} else {
 			reason = reasonBugParSigDBInconsistent
-			if expectInconsistentParSigs(duty.Type) {
+			// Sync committee duties skip consensus, so inconsistent partial signatures are expected, not a bug.
+			if duty.Type == core.DutySyncMessage || duty.Type == core.DutySyncContribution {
 				reason = reasonParSigDBInconsistentSync
 			}
 		}
@@ -880,6 +884,14 @@ func (t *Tracker) InclusionChecked(duty core.Duty, key core.PubKey, _ core.Signe
 }
 
 func reportParSigs(ctx context.Context, duty core.Duty, parsigMsgs parsigsByMsg) {
+	// Sync committee messages don't go through consensus: each node signs its own beacon
+	// node's head block root, so peers legitimately split into cohorts by head. Report these
+	// per-slot (all validators share the head root) via a dedicated per-peer metric and log.
+	if duty.Type == core.DutySyncMessage {
+		reportSyncMessageCohorts(ctx, duty, parsigMsgs)
+		return
+	}
+
 	if parsigMsgs.MsgRootsConsistent() {
 		return // Nothing to report.
 	}
@@ -922,7 +934,9 @@ func reportParSigs(ctx context.Context, duty core.Duty, parsigMsgs parsigsByMsg)
 			entriesJSON = json.RawMessage(fmt.Sprintf("%q", err.Error()))
 		}
 
-		if expectInconsistentParSigs(duty.Type) {
+		// DutySyncMessage is handled earlier via reportSyncMessageCohorts. Sync contribution
+		// inconsistency is expected (per-validator, per-subcommittee) so log at debug.
+		if duty.Type == core.DutySyncContribution {
 			log.Debug(ctx, "Inconsistent sync committee partial signed data",
 				z.Any("pubkey", pubkey),
 				z.Any("duty", duty),
@@ -938,8 +952,106 @@ func reportParSigs(ctx context.Context, duty core.Duty, parsigMsgs parsigsByMsg)
 	}
 }
 
-// expectInconsistentParSigs returns true if the duty type is expected to sometimes
-// produce inconsistent partial signed data.
-func expectInconsistentParSigs(duty core.DutyType) bool {
-	return duty == core.DutySyncMessage || duty == core.DutySyncContribution
+// cohort is a group of share indices that signed the same message root for a slot.
+type cohort struct {
+	root      [32]byte
+	shareIdxs []int // sorted ascending
+}
+
+// computeSyncCohorts groups the share indices that signed each head block root for a sync
+// message duty into cohorts, collapsed across all validators (which share the head root).
+// Cohorts are sorted by size descending, then root ascending, so their slice index is a
+// deterministic rank (0 = largest cohort).
+func computeSyncCohorts(parsigMsgs parsigsByMsg) []cohort {
+	// Sync messages share one head root, so a share index appears once per validator per
+	// root; collect the distinct set of share indices per root.
+	idxsByRoot := make(map[[32]byte]map[int]bool)
+
+	for _, byRoot := range parsigMsgs {
+		for root, parsigs := range byRoot {
+			set, ok := idxsByRoot[root]
+			if !ok {
+				set = make(map[int]bool)
+				idxsByRoot[root] = set
+			}
+
+			for _, parsig := range parsigs {
+				set[parsig.ShareIdx] = true
+			}
+		}
+	}
+
+	cohorts := make([]cohort, 0, len(idxsByRoot))
+
+	for root, set := range idxsByRoot {
+		idxs := make([]int, 0, len(set))
+		for idx := range set {
+			idxs = append(idxs, idx)
+		}
+
+		slices.Sort(idxs)
+		cohorts = append(cohorts, cohort{root: root, shareIdxs: idxs})
+	}
+
+	slices.SortFunc(cohorts, func(a, b cohort) int {
+		if n := len(b.shareIdxs) - len(a.shareIdxs); n != 0 {
+			return n // Larger cohort first (rank ascending by size descending).
+		}
+
+		return bytes.Compare(a.root[:], b.root[:]) // Deterministic tie-break.
+	})
+
+	return cohorts
+}
+
+// reportSyncMessageCohorts instruments per-peer head disagreement for a sync message duty.
+// It emits the cohort rank metric for every peer that signed (on agreement all peers are
+// rank 0) and warns once per slot when peers signed more than one head block root.
+func reportSyncMessageCohorts(ctx context.Context, duty core.Duty, parsigMsgs parsigsByMsg) {
+	cohorts := computeSyncCohorts(parsigMsgs)
+	if len(cohorts) == 0 {
+		return
+	}
+
+	dutyLabel := duty.Type.String()
+
+	for rank, c := range cohorts {
+		rankLabel := strconv.Itoa(rank)
+		for _, shareIdx := range c.shareIdxs {
+			// peer_idx is 0-based (shareIdx-1) to match the core_parsigdb_store convention.
+			cohortRank.WithLabelValues(dutyLabel, strconv.Itoa(shareIdx-1), rankLabel).Inc()
+		}
+	}
+
+	if len(cohorts) <= 1 {
+		return // Consistent: metric emitted, no disagreement to warn about.
+	}
+
+	inconsistentCounter.WithLabelValues(dutyLabel).Inc()
+
+	// Build one entry per distinct head root showing which shares signed it. The signed data
+	// sample is omitted: the root and share indices are the per-slot cohort table.
+	type rootEntry struct {
+		MsgRoot   string `json:"msg_root"`
+		ShareIdxs []int  `json:"share_idxs"`
+	}
+
+	entries := make([]rootEntry, 0, len(cohorts))
+	for _, c := range cohorts {
+		entries = append(entries, rootEntry{
+			MsgRoot:   hex.EncodeToString(c.root[:]),
+			ShareIdxs: c.shareIdxs,
+		})
+	}
+
+	entriesJSON, err := json.Marshal(entries)
+	if err != nil {
+		entriesJSON = json.RawMessage(fmt.Sprintf("%q", err.Error()))
+	}
+
+	log.Warn(ctx, "Sync committee head disagreement", nil,
+		z.Any("duty", duty),
+		z.U64("slot", duty.Slot),
+		z.Int("distinct_roots", len(cohorts)),
+		z.Str("data_by_root", string(entriesJSON)))
 }

core/tracker/tracker_internal_test.go

@@ -7,6 +7,7 @@ import (
 	"math/rand"
 	"net/http"
 	"reflect"
+	"strconv"
 	"sync"
 	"testing"
 	"time"
@@ -15,6 +16,7 @@ import (
 	eth2v1 "github.com/attestantio/go-eth2-client/api/v1"
 	eth2spec "github.com/attestantio/go-eth2-client/spec"
 	eth2p0 "github.com/attestantio/go-eth2-client/spec/phase0"
+	promtestutil "github.com/prometheus/client_golang/prometheus/testutil"
 	"github.com/stretchr/testify/require"
 
 	"github.com/obolnetwork/charon/app/errors"
@@ -1459,3 +1461,160 @@ func TestCountAttestations(t *testing.T) {
 		require.Equal(t, 0, count)
 	})
 }
+
+func rootFrom(b byte) [32]byte {
+	var r [32]byte
+
+	r[0] = b
+
+	return r
+}
+
+func syncSigs(idxs ...int) []core.ParSignedData {
+	out := make([]core.ParSignedData, 0, len(idxs))
+	for _, i := range idxs {
+		out = append(out, core.ParSignedData{ShareIdx: i})
+	}
+
+	return out
+}
+
+func TestComputeSyncCohorts(t *testing.T) {
+	rootA := rootFrom(0x0a)
+	rootB := rootFrom(0x0b)
+	rootC := rootFrom(0x0c)
+
+	tests := []struct {
+		name string
+		in   parsigsByMsg
+		want [][]int // want[rank] = sorted share indices of the cohort at that rank
+	}{
+		{
+			name: "single cohort",
+			in:   parsigsByMsg{"pk": {rootA: syncSigs(1, 2, 3, 4, 5, 6, 7)}},
+			want: [][]int{{1, 2, 3, 4, 5, 6, 7}},
+		},
+		{
+			name: "five two split",
+			in:   parsigsByMsg{"pk": {rootA: syncSigs(1, 2, 3, 4, 5), rootB: syncSigs(6, 7)}},
+			want: [][]int{{1, 2, 3, 4, 5}, {6, 7}},
+		},
+		{
+			name: "three way split",
+			in:   parsigsByMsg{"pk": {rootA: syncSigs(1, 2, 3, 4), rootB: syncSigs(5, 6), rootC: syncSigs(7)}},
+			want: [][]int{{1, 2, 3, 4}, {5, 6}, {7}},
+		},
+		{
+			name: "equal size tie broken by root ascending",
+			in:   parsigsByMsg{"pk": {rootB: syncSigs(3, 4), rootA: syncSigs(1, 2)}},
+			want: [][]int{{1, 2}, {3, 4}}, // rootA (0x0a) ranks before rootB (0x0b)
+		},
+		{
+			name: "collapsed across validators",
+			in: parsigsByMsg{
+				"pk1": {rootA: syncSigs(1, 2, 3, 4, 5), rootB: syncSigs(6, 7)},
+				"pk2": {rootA: syncSigs(1, 2, 3, 4, 5), rootB: syncSigs(6, 7)},
+			},
+			want: [][]int{{1, 2, 3, 4, 5}, {6, 7}},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cohorts := computeSyncCohorts(tt.in)
+			require.Len(t, cohorts, len(tt.want))
+
+			for rank, wantIdxs := range tt.want {
+				require.Equal(t, wantIdxs, cohorts[rank].shareIdxs, "rank %d", rank)
+			}
+		})
+	}
+}
+
+func BenchmarkComputeSyncCohorts(b *testing.B) {
+	const (
+		validators = 500
+		shares     = 7
+	)
+
+	rootA := rootFrom(0x0a)
+	rootB := rootFrom(0x0b)
+
+	build := func(split bool) parsigsByMsg {
+		out := make(parsigsByMsg, validators)
+		for v := range validators {
+			byRoot := make(map[[32]byte][]core.ParSignedData)
+			if split {
+				// 5/2 head disagreement: exercises the multi-cohort sort and bytes.Compare tie-break.
+				byRoot[rootA] = syncSigs(1, 2, 3, 4, 5)
+				byRoot[rootB] = syncSigs(6, 7)
+			} else {
+				byRoot[rootA] = syncSigs(1, 2, 3, 4, 5, 6, 7)
+			}
+
+			out[core.PubKey(strconv.Itoa(v))] = byRoot
+		}
+
+		return out
+	}
+
+	for name, in := range map[string]parsigsByMsg{
+		"agreement": build(false),
+		"split_5_2": build(true),
+	} {
+		b.Run(name, func(b *testing.B) {
+			b.ReportAllocs()
+
+			for range b.N {
+				_ = computeSyncCohorts(in)
+			}
+		})
+	}
+}
+
+func TestReportSyncMessageCohorts(t *testing.T) {
+	ctx := context.Background()
+	rootA := rootFrom(0x1a)
+	rootB := rootFrom(0x1b)
+
+	duty := core.NewSyncMessageDuty(123)
+	dutyLabel := duty.Type.String()
+
+	cohortVal := func(peerIdx, rank int) float64 {
+		return promtestutil.ToFloat64(cohortRank.WithLabelValues(dutyLabel, strconv.Itoa(peerIdx), strconv.Itoa(rank)))
+	}
+	inconsistentVal := func() float64 {
+		return promtestutil.ToFloat64(inconsistentCounter.WithLabelValues(dutyLabel))
+	}
+
+	t.Run("agreement emits rank 0 only", func(t *testing.T) {
+		peer0, peer4, before := cohortVal(0, 0), cohortVal(4, 0), inconsistentVal()
+
+		reportSyncMessageCohorts(ctx, duty, parsigsByMsg{"pk": {rootA: syncSigs(1, 2, 3, 4, 5)}})
+
+		require.InDelta(t, peer0+1, cohortVal(0, 0), 0)
+		require.InDelta(t, peer4+1, cohortVal(4, 0), 0)
+		require.InDelta(t, before, inconsistentVal(), 0) // No disagreement flagged.
+	})
+
+	t.Run("split ranks peers and flags disagreement", func(t *testing.T) {
+		majority, minority, before := cohortVal(0, 0), cohortVal(5, 1), inconsistentVal()
+
+		// share idx 1-5 (peer_idx 0-4) sign rootA (rank 0), share idx 6-7 (peer_idx 5-6) sign rootB (rank 1).
+		reportSyncMessageCohorts(ctx, duty, parsigsByMsg{"pk": {rootA: syncSigs(1, 2, 3, 4, 5), rootB: syncSigs(6, 7)}})
+
+		require.InDelta(t, majority+1, cohortVal(0, 0), 0)
+		require.InDelta(t, minority+1, cohortVal(5, 1), 0)
+		require.InDelta(t, before+1, inconsistentVal(), 0)
+	})
+
+	t.Run("non-sync duty does not emit cohort rank", func(t *testing.T) {
+		attDuty := core.NewAttesterDuty(123)
+		attLabel := attDuty.Type.String()
+		before := promtestutil.ToFloat64(cohortRank.WithLabelValues(attLabel, "0", "0"))
+
+		reportParSigs(ctx, attDuty, parsigsByMsg{"pk": {rootA: syncSigs(1, 2, 3), rootB: syncSigs(4, 5)}})
+
+		require.InDelta(t, before, promtestutil.ToFloat64(cohortRank.WithLabelValues(attLabel, "0", "0")), 0)
+	})
+}

docs/metrics.md

@@ -85,6 +85,7 @@ when storing metrics from multiple nodes or clusters in one Prometheus instance.
 | `core_tracker_inclusion_delay` | Gauge | Cluster`s average attestation inclusion delay in slots. Available only when attestation_inclusion feature flag is enabled. |  |
 | `core_tracker_inclusion_missed_total` | Counter | Total number of broadcast duties never included in any block by type | `duty` |
 | `core_tracker_inconsistent_parsigs_total` | Counter | Total number of duties that contained inconsistent partial signed data by duty type | `duty` |
+| `core_tracker_parsig_cohort_rank_total` | Counter | Total sync committee partial signatures per peer per cohort rank (0=largest cohort), for detecting head disagreement | `duty, peer_idx, rank` |
 | `core_tracker_participation` | Gauge | Set to 1 if peer participated successfully for the given duty or else 0 | `duty, peer` |
 | `core_tracker_participation_expected_total` | Counter | Total number of expected participations (fail + success) by peer and duty type | `duty, peer` |
 | `core_tracker_participation_missed_total` | Counter | Total number of missed participations by peer and duty type | `duty, peer` |