fix(clock-skew): restrict per-node skew to self-originated adverts (#1816, #1818) (#1820)

Closes #1816. Closes #1818 (confirmed duplicate of #1816 by the triage
bot).

## Root cause

`byNode` is an involvement index (`indexResolvedPathHops`,
`store.go:1696-1705`, #1558/#1352): a transmission is indexed under
every relay-hop pubkey found in an observation's `resolved_path`, not
just its originator. `getNodeClockSkewLocked` (`clock_skew.go:489`)
iterated every ADVERT transaction under a pubkey without checking who
actually signed it, so a relay inherited the clock skew of every
broken-clock node it forwarded as if it were its own.

This produced:
- Fleet-wide false `no_clock`/`bimodal_clock` classifications on healthy
relays whose only "bad" samples were adverts they merely relayed.
- Bit-identical `RecentMedianSkewSec` "clusters" across unrelated relays
that all forwarded the same broken-clock originator.
- Single relays showing a multi-day skew even though their own
self-adverts are healthy, because 1-2 relayed adverts from a broken
originator landed in the tail of their small recent-window sample (the
#1818 "island" repro from @cwichura).

## Fix

Add `txOriginatedBy(tx, pubkey)`: ADVERTs are self-signed, so
`decoded["pubKey"]` is the originator per protocol (case-insensitive
compare as a defensive measure). Apply it as a guard in both the main
skew-aggregation loop and the per-hash evidence loop in
`getNodeClockSkewLocked`. `byNode` itself is untouched — #1558/#1352
still rely on the broader involvement index for other consumers.

## Tests

- Existing `clock_skew_test.go` / `clock_skew_issue1094_test.go` /
`clock_skew_issue1285_test.go` fixtures built synthetic ADVERT
transactions without a `pubKey` field and seeded `s.byNode` directly,
bypassing the normal `indexByNode` path where every real ADVERT carries
`pubKey`. Added `pubKey` to each fixture so it reflects a
self-originated advert, which is what these tests already intended to
represent. All pre-existing tests pass unchanged in behavior.
- New `clock_skew_issue1816_test.go`:
- `TestTxOriginatedBy` — unit coverage of the new guard (self, foreign,
missing pubKey, case-insensitivity).
- `TestIssue1816_RelayDoesNotInheritOriginatorSkew` — a relay with
healthy self-adverts plus relayed adverts from a broken-clock originator
(matching the report's +100.5k s band) must report `ok` severity based
only on its own adverts.
- `TestIssue1816_PureRelaysReportNoSkew_NoBitIdenticalCluster` — five
relay pubkeys that only ever forward a broken originator's advert (never
self-advert) must report `nil`, not a bit-identical copy of the
originator's skew.
- `TestIssue1818_TwoForeignAdvertsDoNotPoisonIslandNode` — reproduces
the cwichura island scenario: 8 healthy self-adverts + 2 foreign adverts
at ~10 days skew must not flip severity or pollute
`RecentMedianSkewSec`.

Full suite: `go test ./...` passes (one pre-existing, unrelated flaky
test — `TestHandleNodePaths_PrefixCollision_1352`, an index-loading race
— reproduces intermittently on unmodified `master` too).

Operator context: running CoreScope for SaarMesh (SaarLorLux, DE/FR/LU,
800+ nodes); this bug was surfacing as fleet-wide clock-skew false
positives on our infra nodes.

Co-Authored-By: Claude <noreply@anthropic.com>

Co-authored-by: Claude <noreply@anthropic.com>

cmd/server/clock_skew.go

@@ -3,6 +3,7 @@ package main
 import (
 	"math"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 )
@@ -484,6 +485,34 @@ func (s *PacketStore) GetNodeClockSkew(pubkey string) *NodeClockSkew {
 	return s.getNodeClockSkewLocked(pubkey)
 }
 
+// txOriginatedBy reports whether tx is an ADVERT self-signed by pubkey,
+// as opposed to a transmission merely indexed under pubkey because it was
+// resolved as a relay hop. byNode is an involvement index that
+// intentionally includes relay-hop transmissions for the activity
+// timeline (store.go:1696-1705, indexResolvedPathHops, #1558/#1352).
+// Clock-skew computation must restrict to self-originated adverts only:
+// ADVERTs are self-signed, so decoded["pubKey"] is the originator per
+// protocol (safe key, does not need path-hop resolution). Without this
+// guard, every relay that forwards a broken-clock node's advert inherits
+// that node's skew, producing fleet-wide false no_clock/bimodal
+// classifications and bit-identical skew "clusters" across unrelated
+// relays (#1816), including cases where a node's own adverts are healthy
+// but a couple of relayed adverts from a broken-clock originator dominate
+// its small recent-window sample (#1818). Comparison is case-insensitive
+// as a defensive measure; pubkeys are lowercase hex today (decoder.go),
+// but nothing enforces that at this boundary.
+func txOriginatedBy(tx *StoreTx, pubkey string) bool {
+	decoded := tx.ParsedDecoded()
+	if decoded == nil {
+		return false
+	}
+	pk, ok := decoded["pubKey"].(string)
+	if !ok || pk == "" {
+		return false
+	}
+	return strings.EqualFold(pk, pubkey)
+}
+
 // getNodeClockSkewLocked returns clock skew for a node.
 // Must be called with s.mu held (at least RLock).
 func (s *PacketStore) getNodeClockSkewLocked(pubkey string) *NodeClockSkew {
@@ -508,6 +537,12 @@ func (s *PacketStore) getNodeClockSkewLocked(pubkey string) *NodeClockSkew {
 		if tx.PayloadType == nil || *tx.PayloadType != PayloadADVERT {
 			continue
 		}
+		if !txOriginatedBy(tx, pubkey) {
+			// Skip adverts this node merely relayed (byNode is an
+			// involvement index, not an originator index) — see
+			// txOriginatedBy and #1816/#1818.
+			continue
+		}
 		cs, ok := s.clockSkew.nodeSkew[tx.Hash]
 		if !ok {
 			continue
@@ -672,6 +707,11 @@ func (s *PacketStore) getNodeClockSkewLocked(pubkey string) *NodeClockSkew {
 		if tx.PayloadType == nil || *tx.PayloadType != PayloadADVERT {
 			continue
 		}
+		if !txOriginatedBy(tx, pubkey) {
+			// Keep evidence consistent with the self-only skew stream
+			// above — don't show relayed adverts as this node's evidence.
+			continue
+		}
 		ev, ok := s.clockSkew.hashEvidence[tx.Hash]
 		if !ok || len(ev) == 0 {
 			continue

cmd/server/clock_skew_issue1094_test.go

@@ -36,7 +36,7 @@ func seedIssue1094Repro(t *testing.T) (*PacketStore, []string, []int64) {
 		txs = append(txs, &StoreTx{
 			Hash:        "healthy-1094-" + formatInt64(int64(i)),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"BADTS1094"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -52,7 +52,7 @@ func seedIssue1094Repro(t *testing.T) (*PacketStore, []string, []int64) {
 		txs = append(txs, &StoreTx{
 			Hash:        hash,
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"BADTS1094"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},

cmd/server/clock_skew_issue1285_test.go

@@ -41,7 +41,7 @@ func seedIssue1285Repro(t *testing.T) *PacketStore {
 		tx := &StoreTx{
 			Hash:        "healthy-" + formatInt64(int64(i)),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"RTCRESET"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -56,7 +56,7 @@ func seedIssue1285Repro(t *testing.T) *PacketStore {
 	rtcTx := &StoreTx{
 		Hash:        "rtc-reset-0001",
 		PayloadType: &pt,
-		DecodedJSON: `{"payload":{"timestamp":` + formatInt64(rtcResetAdv) + `}}`,
+		DecodedJSON: `{"payload":{"timestamp":` + formatInt64(rtcResetAdv) + `},"pubKey":"RTCRESET"}`,
 		Observations: []*StoreObs{
 			{ObserverID: "obs1", Timestamp: time.Unix(rtcResetObs, 0).UTC().Format(time.RFC3339)},
 		},

cmd/server/clock_skew_issue1816_test.go

@@ -0,0 +1,287 @@
+package main
+
+// Regression tests for #1816 / #1818:
+//
+// getNodeClockSkewLocked() iterated every ADVERT transaction indexed under
+// a pubkey in s.byNode WITHOUT checking whether the node actually
+// originated (self-signed) that advert. byNode is an involvement index
+// (store.go:1696-1705, indexResolvedPathHops, #1558/#1352): a transmission
+// is also indexed under every relay-hop pubkey extracted from an
+// observation's resolved_path. So a relay that merely forwarded a
+// broken-clock node's advert inherited that node's skew as if it were its
+// own — producing:
+//
+//   - Fleet-wide false no_clock / bimodal_clock classifications on healthy
+//     relays whose only "bad" samples are adverts they relayed (#1816).
+//   - Bit-identical RecentMedianSkewSec "clusters" across unrelated relays
+//     that all forwarded the same broken-clock originator (#1816).
+//   - A single relay showing a nonsense multi-day skew even though its own
+//     self-adverts are perfectly healthy, because 1-2 relayed adverts from
+//     a broken originator landed in the tail of its small recent-window
+//     sample (#1818 — the "island" repro from cwichura).
+//
+// The fix (txOriginatedBy in clock_skew.go) restricts the ADVERT filter in
+// getNodeClockSkewLocked to self-originated adverts only (decoded pubKey
+// == pubkey), leaving byNode itself untouched (#1558/#1352 still depend on
+// the broader involvement index for other consumers).
+
+import (
+	"testing"
+	"time"
+)
+
+// ── txOriginatedBy ───────────────────────────────────────────────────────────
+
+func TestTxOriginatedBy(t *testing.T) {
+	pt := 4 // ADVERT
+	selfTx := &StoreTx{
+		Hash:        "self-1",
+		PayloadType: &pt,
+		DecodedJSON: `{"payload":{"timestamp":1700000000},"pubKey":"AABBCC"}`,
+	}
+	foreignTx := &StoreTx{
+		Hash:        "foreign-1",
+		PayloadType: &pt,
+		DecodedJSON: `{"payload":{"timestamp":1700000000},"pubKey":"DDEEFF"}`,
+	}
+	noPubkeyTx := &StoreTx{
+		Hash:        "nopk-1",
+		PayloadType: &pt,
+		DecodedJSON: `{"payload":{"timestamp":1700000000}}`,
+	}
+	mixedCaseTx := &StoreTx{
+		Hash:        "case-1",
+		PayloadType: &pt,
+		DecodedJSON: `{"payload":{"timestamp":1700000000},"pubKey":"aabbcc"}`,
+	}
+
+	if !txOriginatedBy(selfTx, "AABBCC") {
+		t.Error("expected self-signed advert to match its own pubkey")
+	}
+	if txOriginatedBy(foreignTx, "AABBCC") {
+		t.Error("expected advert from a different originator NOT to match")
+	}
+	if txOriginatedBy(noPubkeyTx, "AABBCC") {
+		t.Error("expected advert with no decoded pubKey NOT to match anything")
+	}
+	if !txOriginatedBy(mixedCaseTx, "AABBCC") {
+		t.Error("expected case-insensitive pubkey match to succeed")
+	}
+}
+
+// ── #1816: relay must not inherit the originator's skew ───────────────────────
+
+// A relay ("RELAY001") only ever transmits healthy self-adverts (skew ~0s).
+// It also forwards adverts from a broken-clock originator ("BROKENORIG",
+// skew ~+100,500s, matching the #1816 report's +100.5k band) — those
+// transmissions are indexed under RELAY001 in byNode too, exactly as
+// indexResolvedPathHops does for real relay-hop traffic. Before the fix,
+// these foreign adverts polluted RELAY001's skew stream and could flip its
+// severity to no_clock/bimodal even though every advert RELAY001 itself
+// sent is fine.
+func TestIssue1816_RelayDoesNotInheritOriginatorSkew(t *testing.T) {
+	ps := NewPacketStore(nil, nil)
+	pt := 4 // ADVERT
+	baseObs := int64(1700000000)
+
+	var relayTxs []*StoreTx
+
+	// RELAY001's own healthy adverts: skew ~0s.
+	for i := 0; i < 5; i++ {
+		obsTS := baseObs + int64(i)*300
+		advTS := obsTS - 2 // 2s behind, trivially healthy
+		tx := &StoreTx{
+			Hash:        "relay-self-" + formatInt64(int64(i)),
+			PayloadType: &pt,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"RELAY001"}`,
+			Observations: []*StoreObs{
+				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
+			},
+		}
+		relayTxs = append(relayTxs, tx)
+	}
+
+	// BROKENORIG's adverts: skew ~+100,500s (matches the #1816 band).
+	// These are indexed under RELAY001 too (simulating indexResolvedPathHops
+	// adding the transmission under the relay-hop pubkey), but their
+	// decoded pubKey is BROKENORIG, not RELAY001.
+	for i := 0; i < 5; i++ {
+		obsTS := baseObs + int64(100+i)*300
+		advTS := obsTS + 100500
+		tx := &StoreTx{
+			Hash:        "relayed-foreign-" + formatInt64(int64(i)),
+			PayloadType: &pt,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"BROKENORIG"}`,
+			Observations: []*StoreObs{
+				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
+			},
+		}
+		relayTxs = append(relayTxs, tx)
+	}
+
+	ps.mu.Lock()
+	ps.byNode["RELAY001"] = relayTxs
+	for _, tx := range relayTxs {
+		ps.byPayloadType[4] = append(ps.byPayloadType[4], tx)
+	}
+	ps.clockSkew.computeInterval = 0
+	ps.mu.Unlock()
+
+	r := ps.GetNodeClockSkew("RELAY001")
+	if r == nil {
+		t.Fatal("expected clock skew result for RELAY001")
+	}
+	if r.Severity != SkewOK {
+		t.Errorf("severity = %v, want ok — relay's own adverts are healthy; "+
+			"relayed foreign adverts must not be attributed to it", r.Severity)
+	}
+	if abs(r.RecentMedianSkewSec) > 5 {
+		t.Errorf("recentMedianSkewSec = %v, want ~0 (only RELAY001's own adverts "+
+			"should count, not the relayed +100.5k s originator adverts)", r.RecentMedianSkewSec)
+	}
+	// Sample count must reflect only the 5 self-originated adverts.
+	if r.RecentSampleCount != 5 {
+		t.Errorf("recentSampleCount = %d, want 5 (relayed adverts must be excluded)", r.RecentSampleCount)
+	}
+}
+
+// ── #1816: bit-identical clusters across relays disappear ─────────────────────
+
+// Five distinct relay pubkeys never self-advert at all — they only forward
+// a single broken-clock originator's advert. Before the fix, each relay's
+// GetNodeClockSkew would report the SAME bit-identical skew as the
+// originator (a false "cluster"). After the fix, a node with zero
+// self-originated adverts must report no clock-skew data at all (nil),
+// since it never actually said anything about its own clock.
+func TestIssue1816_PureRelaysReportNoSkew_NoBitIdenticalCluster(t *testing.T) {
+	ps := NewPacketStore(nil, nil)
+	pt := 4 // ADVERT
+	baseObs := int64(1700000000)
+
+	const originator = "BROKENORIG2"
+	obsTS := baseObs
+	advTS := obsTS + 201000 // matches the #1816 report's +201k s band
+
+	originatorTx := &StoreTx{
+		Hash:        "originator-advert-1",
+		PayloadType: &pt,
+		DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"` + originator + `"}`,
+		Observations: []*StoreObs{
+			{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
+		},
+	}
+
+	relayKeys := []string{"RLY-A", "RLY-B", "RLY-C", "RLY-D", "RLY-E"}
+
+	ps.mu.Lock()
+	// The originator's own byNode entry (legitimate — it self-signed this advert).
+	ps.byNode[originator] = []*StoreTx{originatorTx}
+	// Each relay is ALSO indexed under this same transmission (as
+	// indexResolvedPathHops does for every relay hop in the resolved path),
+	// but none of them ever self-advert.
+	for _, rk := range relayKeys {
+		ps.byNode[rk] = []*StoreTx{originatorTx}
+	}
+	ps.byPayloadType[4] = []*StoreTx{originatorTx}
+	ps.clockSkew.computeInterval = 0
+	ps.mu.Unlock()
+
+	// The originator itself should still report its own (bad) skew.
+	origResult := ps.GetNodeClockSkew(originator)
+	if origResult == nil {
+		t.Fatal("expected clock skew result for the originator")
+	}
+	if abs(origResult.RecentMedianSkewSec-201000) > 5 {
+		t.Errorf("originator recentMedianSkewSec = %v, want ~201000", origResult.RecentMedianSkewSec)
+	}
+
+	// None of the pure relays ever said anything about their own clock —
+	// they must report nil, not a copy of the originator's skew.
+	for _, rk := range relayKeys {
+		r := ps.GetNodeClockSkew(rk)
+		if r != nil {
+			t.Errorf("relay %s: expected nil (no self-originated adverts), "+
+				"got skew=%v — this is the bit-identical false cluster from #1816",
+				rk, r.RecentMedianSkewSec)
+		}
+	}
+}
+
+// ── #1818: two foreign adverts must not poison an otherwise-healthy node ──────
+
+// Reproduces the cwichura "island" report: a node's own adverts are all
+// healthy (~0s skew), but two adverts from a broken-clock originator
+// (skew ~10 days) are present in its byNode entry — exactly the volume
+// described in the issue ("Only 2 adverts out of all the adverts seen are
+// from other nodes"). Because getNodeClockSkewLocked sorts by time and
+// takes only the last recentSkewWindowCount samples, even 2 catastrophic
+// foreign samples can dominate the tail of a small window. After the fix,
+// they must be excluded entirely since they aren't self-originated.
+func TestIssue1818_TwoForeignAdvertsDoNotPoisonIslandNode(t *testing.T) {
+	ps := NewPacketStore(nil, nil)
+	pt := 4 // ADVERT
+	baseObs := int64(1700000000)
+
+	const island = "ISLAND001"
+	var txs []*StoreTx
+
+	// Plenty of healthy self-adverts (island node mostly talks to itself).
+	for i := 0; i < 8; i++ {
+		obsTS := baseObs + int64(i)*300
+		advTS := obsTS - 1 // ~0s skew
+		tx := &StoreTx{
+			Hash:        "island-self-" + formatInt64(int64(i)),
+			PayloadType: &pt,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"` + island + `"}`,
+			Observations: []*StoreObs{
+				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
+			},
+		}
+		txs = append(txs, tx)
+	}
+
+	// Two foreign adverts from a broken-clock originator, landing at the
+	// tail of the recent window (most recent timestamps), skew ~10 days —
+	// matching the report's "off by nearly ten days".
+	for i := 0; i < 2; i++ {
+		obsTS := baseObs + int64(100+i)*300
+		advTS := obsTS + 10*86400
+		tx := &StoreTx{
+			Hash:        "island-foreign-" + formatInt64(int64(i)),
+			PayloadType: &pt,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"OFFCLOCKORIG"}`,
+			Observations: []*StoreObs{
+				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
+			},
+		}
+		txs = append(txs, tx)
+	}
+
+	ps.mu.Lock()
+	ps.byNode[island] = txs
+	for _, tx := range txs {
+		ps.byPayloadType[4] = append(ps.byPayloadType[4], tx)
+	}
+	ps.clockSkew.computeInterval = 0
+	ps.mu.Unlock()
+
+	r := ps.GetNodeClockSkew(island)
+	if r == nil {
+		t.Fatal("expected clock skew result for island node")
+	}
+	if r.Severity == SkewNoClock || r.Severity == SkewBimodalClock {
+		t.Errorf("severity = %v, want ok/warning — the node's own adverts are "+
+			"healthy; the two foreign adverts must not be attributed to it (#1818)",
+			r.Severity)
+	}
+	if abs(r.RecentMedianSkewSec) > 5 {
+		t.Errorf("recentMedianSkewSec = %v, want ~0 (foreign adverts excluded)", r.RecentMedianSkewSec)
+	}
+	// recentSkewWindowCount caps the window at 5 most-recent samples; with
+	// the 2 foreign adverts excluded, those 5 must all come from the
+	// island node's own healthy stream.
+	if r.RecentSampleCount != recentSkewWindowCount {
+		t.Errorf("recentSampleCount = %d, want %d (only self-originated adverts, "+
+			"windowed)", r.RecentSampleCount, recentSkewWindowCount)
+	}
+}

cmd/server/clock_skew_test.go

@@ -318,7 +318,7 @@ func TestGetNodeClockSkew_Integration(t *testing.T) {
 	tx1 := &StoreTx{
 		Hash:        "hash1",
 		PayloadType: &pt,
-		DecodedJSON: `{"payload":{"timestamp":1700002320}}`, // obs=1700002200, node ahead by 120s
+		DecodedJSON: `{"payload":{"timestamp":1700002320},"pubKey":"AABB"}`, // obs=1700002200, node ahead by 120s
 		Observations: []*StoreObs{
 			{ObserverID: "obs1", Timestamp: "2023-11-14T22:50:00Z"}, // 1700002200
 			{ObserverID: "obs2", Timestamp: "2023-11-14T22:50:00Z"}, // 1700002200
@@ -327,7 +327,7 @@ func TestGetNodeClockSkew_Integration(t *testing.T) {
 	tx2 := &StoreTx{
 		Hash:        "hash2",
 		PayloadType: &pt,
-		DecodedJSON: `{"payload":{"timestamp":1700005920}}`, // obs=1700005800, node ahead by 120s
+		DecodedJSON: `{"payload":{"timestamp":1700005920},"pubKey":"AABB"}`, // obs=1700005800, node ahead by 120s
 		Observations: []*StoreObs{
 			{ObserverID: "obs1", Timestamp: "2023-11-14T23:50:00Z"}, // 1700005800
 			{ObserverID: "obs2", Timestamp: "2023-11-14T23:50:00Z"}, // 1700005800
@@ -387,7 +387,7 @@ func TestGetNodeClockSkew_NoClock_EpochZero(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        "epoch-h" + string(rune('0'+i)),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":1577836800}}`, // Jan 1 2020 — valid but way off
+			DecodedJSON: `{"payload":{"timestamp":1577836800},"pubKey":"EPOCH"}`, // Jan 1 2020 — valid but way off
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -428,7 +428,7 @@ func TestGetNodeClockSkew_TooFewSamplesForDrift(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        "few-h" + string(rune('0'+i)),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"FEWSAMP"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -469,7 +469,7 @@ func TestGetNodeClockSkew_AbsurdDriftCapped(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        "wild-h" + string(rune('0'+i)),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"WILD"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -508,7 +508,7 @@ func TestGetNodeClockSkew_NormalNodeWithDrift(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        "norm-h" + string(rune('0'+i)),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"NORMAL"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -567,7 +567,7 @@ func TestSeverityUsesRecentNotMedian(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        fmt.Sprintf("recent-h%03d", i),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"RECENT"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -660,7 +660,7 @@ func TestReporterScenario_789(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        fmt.Sprintf("rep-%04d", i),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"REPNODE"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -717,7 +717,7 @@ func TestBimodalClock_845(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        fmt.Sprintf("bimodal-%04d", i),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"BIMODAL"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -763,7 +763,7 @@ func TestAllBad_NoClock_845(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        fmt.Sprintf("allbad-%04d", i),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"ALLBAD"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -807,7 +807,7 @@ func TestMostlyGood_OK_845(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        fmt.Sprintf("mostly-%04d", i),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"MOSTLY"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -844,7 +844,7 @@ func TestSingleSample_845(t *testing.T) {
 	tx := &StoreTx{
 		Hash:        "single-0001",
 		PayloadType: &pt,
-		DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `}}`,
+		DecodedJSON: `{"payload":{"timestamp":` + formatInt64(advTS) + `},"pubKey":"SINGLE"}`,
 		Observations: []*StoreObs{
 			{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 		},
@@ -887,7 +887,7 @@ func TestFiftyFifty_Bimodal_845(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        fmt.Sprintf("fifty-%04d", i),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(obsTS+skew) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(obsTS+skew) + `},"pubKey":"FIFTY"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -925,7 +925,7 @@ func TestAllGood_OK_845(t *testing.T) {
 		tx := &StoreTx{
 			Hash:        fmt.Sprintf("allgood-%04d", i),
 			PayloadType: &pt,
-			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(obsTS-3) + `}}`,
+			DecodedJSON: `{"payload":{"timestamp":` + formatInt64(obsTS-3) + `},"pubKey":"ALLGOOD"}`,
 			Observations: []*StoreObs{
 				{ObserverID: "obs1", Timestamp: time.Unix(obsTS, 0).UTC().Format(time.RFC3339)},
 			},
@@ -975,7 +975,7 @@ func TestNodeClockSkew_EvidencePayload(t *testing.T) {
 	tx1 := &StoreTx{
 		Hash:        "evidence_hash_1",
 		PayloadType: &pt,
-		DecodedJSON: `{"payload":{"timestamp":1700000060}}`,
+		DecodedJSON: `{"payload":{"timestamp":1700000060},"pubKey":"NODETEST"}`,
 		Observations: []*StoreObs{
 			{ObserverID: "obs1", ObserverName: "Observer Alpha", Timestamp: "2023-11-14T22:13:22Z"},
 			{ObserverID: "obs2", ObserverName: "Observer Beta", Timestamp: "2023-11-14T22:13:20Z"},
@@ -986,7 +986,7 @@ func TestNodeClockSkew_EvidencePayload(t *testing.T) {
 	tx2 := &StoreTx{
 		Hash:        "evidence_hash_2",
 		PayloadType: &pt,
-		DecodedJSON: `{"payload":{"timestamp":1700003660}}`,
+		DecodedJSON: `{"payload":{"timestamp":1700003660},"pubKey":"NODETEST"}`,
 		Observations: []*StoreObs{
 			{ObserverID: "obs1", ObserverName: "Observer Alpha", Timestamp: "2023-11-14T23:13:22Z"},
 			{ObserverID: "obs2", ObserverName: "Observer Beta", Timestamp: "2023-11-14T23:13:20Z"},