feat: Clock skew detection — backend computation (M1) (#746)

## Summary Implements **Milestone 1** of #690 — backend clock skew computation for nodes and observers. ## What's New ### Clock Skew Engine (`clock_skew.go`) **Phase 1 — Raw Skew Calculation:** For every ADVERT observation: `raw_skew = advert_timestamp - observation_timestamp` **Phase 2 — Observer Calibration:** Same packet seen by multiple observers → compute each observer's clock offset as the median deviation from the per-packet median observation timestamp. This identifies observers with their own clock drift. **Phase 3 — Corrected Node Skew:** `corrected_skew = raw_skew + observer_offset` — compensates for observer clock error. **Phase 4 — Trend Analysis:** Linear regression over time-ordered skew samples estimates drift rate in seconds/day. Detects crystal drift vs stable offset vs sudden jumps. ### Severity Classification | Level | Threshold | Meaning | |-------|-----------|---------| | ✅ OK | < 5 min | Normal | | ⚠️ Warning | 5 min – 1 hour | Clock drifting | | 🔴 Critical | 1 hour – 30 days | Likely no time source | | 🟣 Absurd | > 30 days | Firmware default or epoch 0 | ### New API Endpoints - `GET /api/nodes/{pubkey}/clock-skew` — per-node skew data (mean, median, last, drift, severity) - `GET /api/observers/clock-skew` — observer calibration offsets - Clock skew also included in `GET /api/nodes/{pubkey}/analytics` response as `clockSkew` field ### Performance - 30-second compute cache avoids reprocessing on every request - Operates on in-memory `byPayloadType[ADVERT]` index — no DB queries - O(n) in total ADVERT observations, O(m log m) for median calculations ## Tests 15 unit tests covering: - Severity classification at all thresholds - Median/mean math helpers - ISO timestamp parsing - Timestamp extraction from decoded JSON (nested and top-level) - Observer calibration with single and multi-observer scenarios - Observer offset correction direction (verified the sign is `+obsOffset`) - Drift estimation: stable, linear, insufficient data, short time span - JSON number extraction edge cases ## What's NOT in This PR - No UI changes (M2–M4) - No customizer integration (M5) - Thresholds are hardcoded constants (will be configurable in M5) Implements #690 M1. --------- Co-authored-by: you <you@example.com>
2026-10-23 09:12:29 +00:00 · 2026-04-14 23:22:35 -07:00
1 parent aa84ce1e6a
commit a815e70975
@@ -0,0 +1,510 @@
+package main
+
+import (
+	"math"
+	"sort"
+	"sync"
+	"time"
+)
+
+// ── Clock Skew Severity ────────────────────────────────────────────────────────
+
+type SkewSeverity string
+
+const (
+	SkewOK       SkewSeverity = "ok"       // < 5 min
+	SkewWarning  SkewSeverity = "warning"  // 5 min – 1 hour
+	SkewCritical SkewSeverity = "critical" // 1 hour – 30 days
+	SkewAbsurd   SkewSeverity = "absurd"   // > 30 days
+)
+
+// Default thresholds in seconds.
+const (
+	skewThresholdWarnSec     = 5 * 60        // 5 minutes
+	skewThresholdCriticalSec = 60 * 60       // 1 hour
+	skewThresholdAbsurdSec   = 30 * 24 * 3600 // 30 days
+)
+
+// classifySkew maps absolute skew (seconds) to a severity level.
+// Float64 comparison is safe: inputs are rounded to 1 decimal via round(),
+// and thresholds are integer multiples of 60 — no rounding artifacts.
+func classifySkew(absSkewSec float64) SkewSeverity {
+	switch {
+	case absSkewSec >= skewThresholdAbsurdSec:
+		return SkewAbsurd
+	case absSkewSec >= skewThresholdCriticalSec:
+		return SkewCritical
+	case absSkewSec >= skewThresholdWarnSec:
+		return SkewWarning
+	default:
+		return SkewOK
+	}
+}
+
+// ── Data Types ─────────────────────────────────────────────────────────────────
+
+// skewSample is a single raw skew measurement from one advert observation.
+type skewSample struct {
+	advertTS    int64  // node's advert Unix timestamp
+	observedTS  int64  // observation Unix timestamp
+	observerID  string // which observer saw this
+	hash        string // transmission hash (for multi-observer grouping)
+}
+
+// ObserverCalibration holds the computed clock offset for an observer.
+type ObserverCalibration struct {
+	ObserverID string  `json:"observerID"`
+	OffsetSec  float64 `json:"offsetSec"`  // positive = observer clock ahead
+	Samples    int     `json:"samples"`    // number of multi-observer packets used
+}
+
+// NodeClockSkew is the API response for a single node's clock skew data.
+type NodeClockSkew struct {
+	Pubkey          string       `json:"pubkey"`
+	MeanSkewSec     float64      `json:"meanSkewSec"`     // corrected mean skew (positive = node ahead)
+	MedianSkewSec   float64      `json:"medianSkewSec"`   // corrected median skew
+	LastSkewSec     float64      `json:"lastSkewSec"`     // most recent corrected skew
+	DriftPerDaySec  float64      `json:"driftPerDaySec"`  // linear drift rate (sec/day)
+	Severity        SkewSeverity `json:"severity"`
+	SampleCount     int          `json:"sampleCount"`
+	Calibrated      bool         `json:"calibrated"`      // true if observer calibration was applied
+	LastAdvertTS    int64        `json:"lastAdvertTS"`     // most recent advert timestamp
+	LastObservedTS  int64        `json:"lastObservedTS"`   // most recent observation timestamp
+}
+
+// txSkewResult maps tx hash → per-transmission skew stats. This is an
+// intermediate result keyed by hash (not pubkey); the store maps hash → pubkey
+// when building the final per-node view.
+type txSkewResult = map[string]*NodeClockSkew
+
+// ── Clock Skew Engine ──────────────────────────────────────────────────────────
+
+// ClockSkewEngine computes and caches clock skew data for nodes and observers.
+type ClockSkewEngine struct {
+	mu               sync.RWMutex
+	observerOffsets  map[string]float64 // observerID → calibrated offset (seconds)
+	observerSamples  map[string]int     // observerID → number of multi-observer packets used
+	nodeSkew         txSkewResult
+	lastComputed     time.Time
+	computeInterval  time.Duration
+}
+
+func NewClockSkewEngine() *ClockSkewEngine {
+	return &ClockSkewEngine{
+		observerOffsets:  make(map[string]float64),
+		observerSamples: make(map[string]int),
+		nodeSkew:       make(txSkewResult),
+		computeInterval: 30 * time.Second,
+	}
+}
+
+// Recompute recalculates all clock skew data from the packet store.
+// Called periodically or on demand. Holds store RLock externally.
+// Uses read-copy-update: heavy computation runs outside the write lock,
+// then results are swapped in under a brief lock.
+func (e *ClockSkewEngine) Recompute(store *PacketStore) {
+	// Fast path: check under read lock if recompute is needed.
+	e.mu.RLock()
+	fresh := time.Since(e.lastComputed) < e.computeInterval
+	e.mu.RUnlock()
+	if fresh {
+		return
+	}
+
+	// Phase 1: Collect skew samples from ADVERT packets (store RLock held by caller).
+	samples := collectSamples(store)
+
+	// Phase 2–3: Compute outside the write lock.
+	var newOffsets map[string]float64
+	var newSamples map[string]int
+	var newNodeSkew txSkewResult
+
+	if len(samples) > 0 {
+		newOffsets, newSamples = calibrateObservers(samples)
+		newNodeSkew = computeNodeSkew(samples, newOffsets)
+	} else {
+		newOffsets = make(map[string]float64)
+		newSamples = make(map[string]int)
+		newNodeSkew = make(txSkewResult)
+	}
+
+	// Swap results under brief write lock.
+	e.mu.Lock()
+	// Re-check: another goroutine may have computed while we were working.
+	if time.Since(e.lastComputed) < e.computeInterval {
+		e.mu.Unlock()
+		return
+	}
+	e.observerOffsets = newOffsets
+	e.observerSamples = newSamples
+	e.nodeSkew = newNodeSkew
+	e.lastComputed = time.Now()
+	e.mu.Unlock()
+}
+
+// collectSamples extracts skew samples from ADVERT packets in the store.
+// Must be called with store.mu held (at least RLock).
+func collectSamples(store *PacketStore) []skewSample {
+	adverts := store.byPayloadType[PayloadADVERT]
+	if len(adverts) == 0 {
+		return nil
+	}
+
+	samples := make([]skewSample, 0, len(adverts)*2)
+	for _, tx := range adverts {
+		decoded := tx.ParsedDecoded()
+		if decoded == nil {
+			continue
+		}
+		// Extract advert timestamp from decoded JSON.
+		advertTS := extractTimestamp(decoded)
+		if advertTS <= 0 {
+			continue
+		}
+		// Sanity: skip timestamps before year 2020 or after year 2100.
+		if advertTS < 1577836800 || advertTS > 4102444800 {
+			continue
+		}
+
+		for _, obs := range tx.Observations {
+			obsTS := parseISO(obs.Timestamp)
+			if obsTS <= 0 {
+				continue
+			}
+			samples = append(samples, skewSample{
+				advertTS:   advertTS,
+				observedTS: obsTS,
+				observerID: obs.ObserverID,
+				hash:       tx.Hash,
+			})
+		}
+	}
+	return samples
+}
+
+// extractTimestamp gets the Unix timestamp from a decoded ADVERT payload.
+func extractTimestamp(decoded map[string]interface{}) int64 {
+	// Try payload.timestamp first (nested in "payload" key).
+	if payload, ok := decoded["payload"]; ok {
+		if pm, ok := payload.(map[string]interface{}); ok {
+			if ts := jsonNumber(pm, "timestamp"); ts > 0 {
+				return ts
+			}
+		}
+	}
+	// Fallback: top-level timestamp.
+	if ts := jsonNumber(decoded, "timestamp"); ts > 0 {
+		return ts
+	}
+	return 0
+}
+
+// jsonNumber extracts an int64 from a JSON-parsed map (handles float64 and json.Number).
+func jsonNumber(m map[string]interface{}, key string) int64 {
+	v, ok := m[key]
+	if !ok || v == nil {
+		return 0
+	}
+	switch n := v.(type) {
+	case float64:
+		return int64(n)
+	case int64:
+		return n
+	case int:
+		return int64(n)
+	}
+	return 0
+}
+
+// parseISO parses an ISO 8601 timestamp string to Unix seconds.
+func parseISO(s string) int64 {
+	if s == "" {
+		return 0
+	}
+	t, err := time.Parse(time.RFC3339, s)
+	if err != nil {
+		// Try with fractional seconds.
+		t, err = time.Parse("2006-01-02T15:04:05.999999999Z07:00", s)
+		if err != nil {
+			return 0
+		}
+	}
+	return t.Unix()
+}
+
+// ── Phase 2: Observer Calibration ──────────────────────────────────────────────
+
+// calibrateObservers computes each observer's clock offset using multi-observer
+// packets. Returns offset map and sample count map.
+func calibrateObservers(samples []skewSample) (map[string]float64, map[string]int) {
+	// Group observations by packet hash.
+	byHash := make(map[string][]skewSample)
+	for _, s := range samples {
+		byHash[s.hash] = append(byHash[s.hash], s)
+	}
+
+	// For each multi-observer packet, compute per-observer deviation from median.
+	deviations := make(map[string][]float64) // observerID → list of deviations
+	for _, group := range byHash {
+		if len(group) < 2 {
+			continue // single-observer packet, can't calibrate
+		}
+		// Compute median observation timestamp for this packet.
+		obsTimes := make([]float64, len(group))
+		for i, s := range group {
+			obsTimes[i] = float64(s.observedTS)
+		}
+		medianObs := median(obsTimes)
+		for _, s := range group {
+			dev := float64(s.observedTS) - medianObs
+			deviations[s.observerID] = append(deviations[s.observerID], dev)
+		}
+	}
+
+	// Each observer's offset = median of its deviations.
+	offsets := make(map[string]float64, len(deviations))
+	counts := make(map[string]int, len(deviations))
+	for obsID, devs := range deviations {
+		offsets[obsID] = median(devs)
+		counts[obsID] = len(devs)
+	}
+	return offsets, counts
+}
+
+// ── Phase 3: Per-Node Skew ─────────────────────────────────────────────────────
+
+// computeNodeSkew calculates corrected skew statistics for each node.
+func computeNodeSkew(samples []skewSample, obsOffsets map[string]float64) txSkewResult {
+	// Compute corrected skew per sample, grouped by hash (each hash = one
+	// node's advert transmission). The caller maps hash → pubkey via byNode.
+	type correctedSample struct {
+		skew       float64
+		observedTS int64
+		calibrated bool
+	}
+
+	byHash := make(map[string][]correctedSample)
+	hashAdvertTS := make(map[string]int64)
+
+	for _, s := range samples {
+		obsOffset, hasCal := obsOffsets[s.observerID]
+		rawSkew := float64(s.advertTS - s.observedTS)
+		corrected := rawSkew
+		if hasCal {
+			// Observer offset = obs_ts - median(all_obs_ts). If observer is ahead,
+			// its obs_ts is inflated, making raw_skew too low. Add offset to correct.
+			corrected = rawSkew + obsOffset
+		}
+		byHash[s.hash] = append(byHash[s.hash], correctedSample{
+			skew:       corrected,
+			observedTS: s.observedTS,
+			calibrated: hasCal,
+		})
+		hashAdvertTS[s.hash] = s.advertTS
+	}
+
+	// Each hash represents one advert from one node. Compute median corrected
+	// skew per hash (across multiple observers).
+
+	result := make(map[string]*NodeClockSkew) // keyed by hash for now
+	for hash, cs := range byHash {
+		skews := make([]float64, len(cs))
+		for i, c := range cs {
+			skews[i] = c.skew
+		}
+		medSkew := median(skews)
+		meanSkew := mean(skews)
+
+		// Find latest observation.
+		var latestObsTS int64
+		var anyCal bool
+		for _, c := range cs {
+			if c.observedTS > latestObsTS {
+				latestObsTS = c.observedTS
+			}
+			if c.calibrated {
+				anyCal = true
+			}
+		}
+
+		absMedian := math.Abs(medSkew)
+		result[hash] = &NodeClockSkew{
+			MeanSkewSec:    round(meanSkew, 1),
+			MedianSkewSec:  round(medSkew, 1),
+			LastSkewSec:    round(cs[len(cs)-1].skew, 1),
+			Severity:       classifySkew(absMedian),
+			SampleCount:    len(cs),
+			Calibrated:     anyCal,
+			LastAdvertTS:   hashAdvertTS[hash],
+			LastObservedTS: latestObsTS,
+		}
+	}
+	return result
+}
+
+// ── Integration with PacketStore ───────────────────────────────────────────────
+
+// GetNodeClockSkew returns the clock skew data for a specific node (acquires RLock).
+func (s *PacketStore) GetNodeClockSkew(pubkey string) *NodeClockSkew {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+	return s.getNodeClockSkewLocked(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 {
+	s.clockSkew.Recompute(s)
+
+	txs := s.byNode[pubkey]
+	if len(txs) == 0 {
+		return nil
+	}
+
+	s.clockSkew.mu.RLock()
+	defer s.clockSkew.mu.RUnlock()
+
+	var allSkews []float64
+	var lastSkew float64
+	var lastObsTS, lastAdvTS int64
+	var totalSamples int
+	var anyCal bool
+	var tsSkews []tsSkewPair
+
+	for _, tx := range txs {
+		if tx.PayloadType == nil || *tx.PayloadType != PayloadADVERT {
+			continue
+		}
+		cs, ok := s.clockSkew.nodeSkew[tx.Hash]
+		if !ok {
+			continue
+		}
+		allSkews = append(allSkews, cs.MedianSkewSec)
+		totalSamples += cs.SampleCount
+		if cs.Calibrated {
+			anyCal = true
+		}
+		if cs.LastObservedTS > lastObsTS {
+			lastObsTS = cs.LastObservedTS
+			lastSkew = cs.LastSkewSec
+			lastAdvTS = cs.LastAdvertTS
+		}
+		tsSkews = append(tsSkews, tsSkewPair{ts: cs.LastObservedTS, skew: cs.MedianSkewSec})
+	}
+
+	if len(allSkews) == 0 {
+		return nil
+	}
+
+	medSkew := median(allSkews)
+	meanSkew := mean(allSkews)
+	absMedian := math.Abs(medSkew)
+	drift := computeDrift(tsSkews)
+
+	return &NodeClockSkew{
+		Pubkey:         pubkey,
+		MeanSkewSec:    round(meanSkew, 1),
+		MedianSkewSec:  round(medSkew, 1),
+		LastSkewSec:    round(lastSkew, 1),
+		DriftPerDaySec: round(drift, 2),
+		Severity:       classifySkew(absMedian),
+		SampleCount:    totalSamples,
+		Calibrated:     anyCal,
+		LastAdvertTS:   lastAdvTS,
+		LastObservedTS: lastObsTS,
+	}
+}
+
+// GetObserverCalibrations returns the current observer clock offsets.
+func (s *PacketStore) GetObserverCalibrations() []ObserverCalibration {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	s.clockSkew.Recompute(s)
+
+	s.clockSkew.mu.RLock()
+	defer s.clockSkew.mu.RUnlock()
+
+	result := make([]ObserverCalibration, 0, len(s.clockSkew.observerOffsets))
+	for obsID, offset := range s.clockSkew.observerOffsets {
+		result = append(result, ObserverCalibration{
+			ObserverID: obsID,
+			OffsetSec:  round(offset, 1),
+			Samples:    s.clockSkew.observerSamples[obsID],
+		})
+	}
+	// Sort by absolute offset descending.
+	sort.Slice(result, func(i, j int) bool {
+		return math.Abs(result[i].OffsetSec) > math.Abs(result[j].OffsetSec)
+	})
+	return result
+}
+
+// ── Math Helpers ───────────────────────────────────────────────────────────────
+
+func median(vals []float64) float64 {
+	if len(vals) == 0 {
+		return 0
+	}
+	sorted := make([]float64, len(vals))
+	copy(sorted, vals)
+	sort.Float64s(sorted)
+	n := len(sorted)
+	if n%2 == 0 {
+		return (sorted[n/2-1] + sorted[n/2]) / 2
+	}
+	return sorted[n/2]
+}
+
+func mean(vals []float64) float64 {
+	if len(vals) == 0 {
+		return 0
+	}
+	sum := 0.0
+	for _, v := range vals {
+		sum += v
+	}
+	return sum / float64(len(vals))
+}
+
+// tsSkewPair is a (timestamp, skew) pair for drift estimation.
+type tsSkewPair struct {
+	ts   int64
+	skew float64
+}
+
+// computeDrift estimates linear drift in seconds per day from time-ordered
+// (timestamp, skew) pairs using simple linear regression.
+func computeDrift(pairs []tsSkewPair) float64 {
+	if len(pairs) < 2 {
+		return 0
+	}
+	// Sort by timestamp.
+	sort.Slice(pairs, func(i, j int) bool {
+		return pairs[i].ts < pairs[j].ts
+	})
+
+	// Time span too short? Skip.
+	spanSec := float64(pairs[len(pairs)-1].ts - pairs[0].ts)
+	if spanSec < 3600 { // need at least 1 hour of data
+		return 0
+	}
+
+	// Simple linear regression: skew = a + b*t
+	n := float64(len(pairs))
+	var sumX, sumY, sumXY, sumX2 float64
+	for _, p := range pairs {
+		x := float64(p.ts - pairs[0].ts) // normalize to avoid large numbers
+		y := p.skew
+		sumX += x
+		sumY += y
+		sumXY += x * y
+		sumX2 += x * x
+	}
+	denom := n*sumX2 - sumX*sumX
+	if denom == 0 {
+		return 0
+	}
+	slope := (n*sumXY - sumX*sumY) / denom // seconds of drift per second
+	return slope * 86400                     // convert to seconds per day
+}
@@ -0,0 +1,369 @@
+package main
+
+import (
+	"math"
+	"testing"
+)
+
+// ── classifySkew ───────────────────────────────────────────────────────────────
+
+func TestClassifySkew(t *testing.T) {
+	tests := []struct {
+		absSkew  float64
+		expected SkewSeverity
+	}{
+		{0, SkewOK},
+		{60, SkewOK},           // 1 min
+		{299, SkewOK},          // just under 5 min
+		{300, SkewWarning},     // exactly 5 min
+		{1800, SkewWarning},    // 30 min
+		{3599, SkewWarning},    // just under 1 hour
+		{3600, SkewCritical},   // exactly 1 hour
+		{86400, SkewCritical},  // 1 day
+		{2592000 - 1, SkewCritical}, // just under 30 days
+		{2592000, SkewAbsurd},  // exactly 30 days
+		{86400 * 365, SkewAbsurd}, // 1 year
+	}
+	for _, tc := range tests {
+		got := classifySkew(tc.absSkew)
+		if got != tc.expected {
+			t.Errorf("classifySkew(%v) = %v, want %v", tc.absSkew, got, tc.expected)
+		}
+	}
+}
+
+// ── median ─────────────────────────────────────────────────────────────────────
+
+func TestMedian(t *testing.T) {
+	tests := []struct {
+		vals     []float64
+		expected float64
+	}{
+		{nil, 0},
+		{[]float64{}, 0},
+		{[]float64{5}, 5},
+		{[]float64{1, 3}, 2},
+		{[]float64{3, 1, 2}, 2},
+		{[]float64{4, 1, 3, 2}, 2.5},
+		{[]float64{-10, 0, 10}, 0},
+	}
+	for _, tc := range tests {
+		got := median(tc.vals)
+		if got != tc.expected {
+			t.Errorf("median(%v) = %v, want %v", tc.vals, got, tc.expected)
+		}
+	}
+}
+
+func TestMean(t *testing.T) {
+	tests := []struct {
+		vals     []float64
+		expected float64
+	}{
+		{nil, 0},
+		{[]float64{10}, 10},
+		{[]float64{2, 4, 6}, 4},
+	}
+	for _, tc := range tests {
+		got := mean(tc.vals)
+		if got != tc.expected {
+			t.Errorf("mean(%v) = %v, want %v", tc.vals, got, tc.expected)
+		}
+	}
+}
+
+// ── parseISO ───────────────────────────────────────────────────────────────────
+
+func TestParseISO(t *testing.T) {
+	tests := []struct {
+		input    string
+		expected int64
+	}{
+		{"", 0},
+		{"garbage", 0},
+		{"2026-04-15T12:00:00Z", 1776254400},
+		{"2026-04-15T12:00:00+00:00", 1776254400},
+	}
+	for _, tc := range tests {
+		got := parseISO(tc.input)
+		if got != tc.expected {
+			t.Errorf("parseISO(%q) = %v, want %v", tc.input, got, tc.expected)
+		}
+	}
+}
+
+// ── extractTimestamp ────────────────────────────────────────────────────────────
+
+func TestExtractTimestamp(t *testing.T) {
+	// Nested payload.timestamp
+	decoded := map[string]interface{}{
+		"payload": map[string]interface{}{
+			"timestamp": float64(1776340800),
+		},
+	}
+	got := extractTimestamp(decoded)
+	if got != 1776340800 {
+		t.Errorf("extractTimestamp (nested) = %v, want 1776340800", got)
+	}
+
+	// Top-level timestamp
+	decoded2 := map[string]interface{}{
+		"timestamp": float64(1776340900),
+	}
+	got2 := extractTimestamp(decoded2)
+	if got2 != 1776340900 {
+		t.Errorf("extractTimestamp (top-level) = %v, want 1776340900", got2)
+	}
+
+	// No timestamp
+	decoded3 := map[string]interface{}{"foo": "bar"}
+	got3 := extractTimestamp(decoded3)
+	if got3 != 0 {
+		t.Errorf("extractTimestamp (missing) = %v, want 0", got3)
+	}
+}
+
+// ── calibrateObservers ─────────────────────────────────────────────────────────
+
+func TestCalibrateObservers_SingleObserver(t *testing.T) {
+	// Single-observer packets can't calibrate — should return empty.
+	samples := []skewSample{
+		{advertTS: 1000, observedTS: 1000, observerID: "obs1", hash: "h1"},
+		{advertTS: 2000, observedTS: 2000, observerID: "obs1", hash: "h2"},
+	}
+	offsets, _ := calibrateObservers(samples)
+	if len(offsets) != 0 {
+		t.Errorf("expected no offsets for single-observer, got %v", offsets)
+	}
+}
+
+func TestCalibrateObservers_MultiObserver(t *testing.T) {
+	// Packet h1 seen by 3 observers: obs1 at t=100, obs2 at t=110, obs3 at t=100.
+	// Median observation = 100. obs1=0, obs2=+10, obs3=0
+	// Packet h2 seen by 3 observers: obs1 at t=200, obs2 at t=210, obs3 at t=200.
+	// Median observation = 200. obs1=0, obs2=+10, obs3=0
+	samples := []skewSample{
+		{advertTS: 100, observedTS: 100, observerID: "obs1", hash: "h1"},
+		{advertTS: 100, observedTS: 110, observerID: "obs2", hash: "h1"},
+		{advertTS: 100, observedTS: 100, observerID: "obs3", hash: "h1"},
+		{advertTS: 200, observedTS: 200, observerID: "obs1", hash: "h2"},
+		{advertTS: 200, observedTS: 210, observerID: "obs2", hash: "h2"},
+		{advertTS: 200, observedTS: 200, observerID: "obs3", hash: "h2"},
+	}
+	offsets, _ := calibrateObservers(samples)
+	if offsets["obs1"] != 0 {
+		t.Errorf("obs1 offset = %v, want 0", offsets["obs1"])
+	}
+	if offsets["obs2"] != 10 {
+		t.Errorf("obs2 offset = %v, want 10", offsets["obs2"])
+	}
+	if offsets["obs3"] != 0 {
+		t.Errorf("obs3 offset = %v, want 0", offsets["obs3"])
+	}
+}
+
+// ── computeNodeSkew ────────────────────────────────────────────────────────────
+
+func TestComputeNodeSkew_BasicCorrection(t *testing.T) {
+	// Validates observer offset correction direction.
+	//
+	// Setup: node is 60s ahead, obs1 accurate, obs2 is 10s ahead.
+	// With 2 observers, median obs_ts = 1005.
+	//   obs1 offset = 1000 - 1005 = -5
+	//   obs2 offset = 1010 - 1005 = +5
+	// Correction: corrected = raw_skew + obsOffset
+	//   obs1: raw=60, corrected = 60 + (-5) = 55
+	//   obs2: raw=50, corrected = 50 + 5 = 55
+	// Both converge to 55 (not exact 60 because with only 2 observers,
+	// the median can't fully distinguish which observer is drifted).
+
+	samples := []skewSample{
+		// Same packet seen by accurate obs1 and obs2 (+10s ahead)
+		{advertTS: 1060, observedTS: 1000, observerID: "obs1", hash: "h1"},
+		{advertTS: 1060, observedTS: 1010, observerID: "obs2", hash: "h1"},
+	}
+	offsets, _ := calibrateObservers(samples)
+	// median obs = 1005, obs1 offset = -5, obs2 offset = +5
+	// So the median approach finds obs2 is +5 ahead (relative to median)
+
+	// Now compute node skew with those offsets:
+	nodeSkew := computeNodeSkew(samples, offsets)
+	cs, ok := nodeSkew["h1"]
+	if !ok {
+		t.Fatal("expected skew data for hash h1")
+	}
+	// With only 2 observers, median obs_ts = 1005.
+	// obs1 offset = 1000-1005 = -5, obs2 offset = 1010-1005 = +5
+	// raw from obs1 = 60, corrected = 60 + (-5) = 55
+	// raw from obs2 = 50, corrected = 50 + 5 = 55
+	// median = 55
+	if cs.MedianSkewSec != 55 {
+		t.Errorf("median skew = %v, want 55", cs.MedianSkewSec)
+	}
+}
+
+func TestComputeNodeSkew_ThreeObservers(t *testing.T) {
+	// Node is exactly 60s ahead. obs1 accurate, obs2 accurate, obs3 +30s ahead.
+	// advertTS = 1060, real time = 1000
+	samples := []skewSample{
+		{advertTS: 1060, observedTS: 1000, observerID: "obs1", hash: "h1"},
+		{advertTS: 1060, observedTS: 1000, observerID: "obs2", hash: "h1"},
+		{advertTS: 1060, observedTS: 1030, observerID: "obs3", hash: "h1"},
+	}
+	offsets, _ := calibrateObservers(samples)
+	// median obs_ts = 1000. obs1=0, obs2=0, obs3=+30
+	if offsets["obs3"] != 30 {
+		t.Errorf("obs3 offset = %v, want 30", offsets["obs3"])
+	}
+
+	nodeSkew := computeNodeSkew(samples, offsets)
+	cs := nodeSkew["h1"]
+	if cs == nil {
+		t.Fatal("expected skew data for h1")
+	}
+	// raw from obs1 = 60, corrected = 60 + 0 = 60
+	// raw from obs2 = 60, corrected = 60 + 0 = 60
+	// raw from obs3 = 30, corrected = 30 + 30 = 60
+	// All three converge to 60. 
+	if cs.MedianSkewSec != 60 {
+		t.Errorf("median skew = %v, want 60 (node is 60s ahead)", cs.MedianSkewSec)
+	}
+}
+
+// ── computeDrift ───────────────────────────────────────────────────────────────
+
+func TestComputeDrift_Stable(t *testing.T) {
+	// Constant skew = no drift.
+	pairs := []tsSkewPair{
+		{ts: 0, skew: 60},
+		{ts: 7200, skew: 60},
+		{ts: 14400, skew: 60},
+	}
+	drift := computeDrift(pairs)
+	if drift != 0 {
+		t.Errorf("drift = %v, want 0 for stable skew", drift)
+	}
+}
+
+func TestComputeDrift_LinearDrift(t *testing.T) {
+	// 1 second drift per hour = 24 sec/day.
+	pairs := []tsSkewPair{
+		{ts: 0, skew: 0},
+		{ts: 3600, skew: 1},
+		{ts: 7200, skew: 2},
+	}
+	drift := computeDrift(pairs)
+	expected := 24.0
+	if math.Abs(drift-expected) > 0.1 {
+		t.Errorf("drift = %v, want ~%v", drift, expected)
+	}
+}
+
+func TestComputeDrift_TooFewSamples(t *testing.T) {
+	pairs := []tsSkewPair{{ts: 0, skew: 10}}
+	if computeDrift(pairs) != 0 {
+		t.Error("expected 0 drift for single sample")
+	}
+}
+
+func TestComputeDrift_TooShortSpan(t *testing.T) {
+	// Less than 1 hour apart.
+	pairs := []tsSkewPair{
+		{ts: 0, skew: 0},
+		{ts: 1800, skew: 10},
+	}
+	if computeDrift(pairs) != 0 {
+		t.Error("expected 0 drift for short time span")
+	}
+}
+
+// ── jsonNumber ─────────────────────────────────────────────────────────────────
+
+func TestJsonNumber(t *testing.T) {
+	m := map[string]interface{}{
+		"a": float64(42),
+		"b": int64(99),
+		"c": "not a number",
+		"d": nil,
+	}
+	if jsonNumber(m, "a") != 42 {
+		t.Error("float64 case failed")
+	}
+	if jsonNumber(m, "b") != 99 {
+		t.Error("int64 case failed")
+	}
+	if jsonNumber(m, "c") != 0 {
+		t.Error("string case should return 0")
+	}
+	if jsonNumber(m, "d") != 0 {
+		t.Error("nil case should return 0")
+	}
+	if jsonNumber(m, "missing") != 0 {
+		t.Error("missing key should return 0")
+	}
+}
+
+// ── Integration: GetNodeClockSkew via PacketStore ──────────────────────────────
+
+func TestGetNodeClockSkew_Integration(t *testing.T) {
+	ps := NewPacketStore(nil, nil)
+
+	// Simulate two ADVERT transmissions for the same node, seen by 2 observers each.
+	// Node "AABB" has clock 120s ahead.
+	pt := 4 // ADVERT
+	tx1 := &StoreTx{
+		Hash:        "hash1",
+		PayloadType: &pt,
+		DecodedJSON: `{"payload":{"timestamp":1700002320}}`, // 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
+		},
+	}
+	tx2 := &StoreTx{
+		Hash:        "hash2",
+		PayloadType: &pt,
+		DecodedJSON: `{"payload":{"timestamp":1700005920}}`, // 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
+		},
+	}
+
+	ps.mu.Lock()
+	ps.byNode["AABB"] = []*StoreTx{tx1, tx2}
+	ps.byPayloadType[4] = []*StoreTx{tx1, tx2}
+	// Force recompute by setting interval to 0.
+	ps.clockSkew.computeInterval = 0
+	ps.mu.Unlock()
+
+	result := ps.GetNodeClockSkew("AABB")
+	if result == nil {
+		t.Fatal("expected clock skew result for node AABB")
+	}
+	if result.Pubkey != "AABB" {
+		t.Errorf("pubkey = %q, want AABB", result.Pubkey)
+	}
+	// Both transmissions show 120s skew, so median should be 120.
+	if result.MedianSkewSec != 120 {
+		t.Errorf("median skew = %v, want 120", result.MedianSkewSec)
+	}
+	if result.SampleCount < 2 {
+		t.Errorf("sample count = %v, want >= 2", result.SampleCount)
+	}
+	if result.Severity != SkewOK {
+		t.Errorf("severity = %v, want ok (120s < 5min)", result.Severity)
+	}
+	// Drift should be ~0 since skew is constant.
+	if math.Abs(result.DriftPerDaySec) > 1 {
+		t.Errorf("drift = %v, want ~0 for constant skew", result.DriftPerDaySec)
+	}
+}
+
+func TestGetNodeClockSkew_NoData(t *testing.T) {
+	ps := NewPacketStore(nil, nil)
+	result := ps.GetNodeClockSkew("nonexistent")
+	if result != nil {
+		t.Error("expected nil for nonexistent node")
+	}
+}
@@ -166,7 +166,7 @@ func BuildFromStoreWithLog(store *PacketStore, enableLog bool) *NeighborGraph {

 	// Phase 1: Extract edges from every transmission + observation.
 	for _, tx := range packets {
-		isAdvert := tx.PayloadType != nil && *tx.PayloadType == 4
+		isAdvert := tx.PayloadType != nil && *tx.PayloadType == PayloadADVERT
 		fromNode := extractFromNode(tx)
 		// Pre-compute lowered originator once per tx (not per observation).
 		fromLower := ""
@@ -525,7 +525,7 @@ type edgeCandidate struct {
 // For ADVERTs: originator↔path[0] (if unambiguous). For ALL types: observer↔path[last] (if unambiguous).
 // Also handles zero-hop ADVERTs (originator↔observer direct link).
 func extractEdgesFromObs(obs *StoreObs, tx *StoreTx, pm *prefixMap) []edgeCandidate {
-	isAdvert := tx.PayloadType != nil && *tx.PayloadType == 4
+	isAdvert := tx.PayloadType != nil && *tx.PayloadType == PayloadADVERT
 	fromNode := extractFromNode(tx)
 	path := parsePathJSON(obs.PathJSON)
 	observerPK := strings.ToLower(obs.ObserverID)
@@ -142,6 +142,8 @@ func (s *Server) RegisterRoutes(r *mux.Router) {
 	r.HandleFunc("/api/nodes/{pubkey}/health", s.handleNodeHealth).Methods("GET")
 	r.HandleFunc("/api/nodes/{pubkey}/paths", s.handleNodePaths).Methods("GET")
 	r.HandleFunc("/api/nodes/{pubkey}/analytics", s.handleNodeAnalytics).Methods("GET")
+	r.HandleFunc("/api/nodes/{pubkey}/clock-skew", s.handleNodeClockSkew).Methods("GET")
+	r.HandleFunc("/api/observers/clock-skew", s.handleObserverClockSkew).Methods("GET")
 	r.HandleFunc("/api/nodes/{pubkey}/neighbors", s.handleNodeNeighbors).Methods("GET")
 	r.HandleFunc("/api/nodes/{pubkey}", s.handleNodeDetail).Methods("GET")
 	r.HandleFunc("/api/nodes", s.handleNodes).Methods("GET")
@@ -1315,6 +1317,28 @@ func (s *Server) handleNodeAnalytics(w http.ResponseWriter, r *http.Request) {
 	writeError(w, 404, "Not found")
 }

+func (s *Server) handleNodeClockSkew(w http.ResponseWriter, r *http.Request) {
+	pubkey := mux.Vars(r)["pubkey"]
+	if s.store == nil {
+		writeError(w, 404, "Not found")
+		return
+	}
+	result := s.store.GetNodeClockSkew(pubkey)
+	if result == nil {
+		writeError(w, 404, "No clock skew data for this node")
+		return
+	}
+	writeJSON(w, result)
+}
+
+func (s *Server) handleObserverClockSkew(w http.ResponseWriter, r *http.Request) {
+	if s.store == nil {
+		writeJSON(w, []ObserverCalibration{})
+		return
+	}
+	writeJSON(w, s.store.GetObserverCalibrations())
+}
+
 // --- Analytics Handlers ---

 func (s *Server) handleAnalyticsRF(w http.ResponseWriter, r *http.Request) {
@@ -200,6 +200,9 @@ type PacketStore struct {
 	// Persisted neighbor graph for hop resolution at ingest time.
 	graph *NeighborGraph

+	// Clock skew detection engine.
+	clockSkew *ClockSkewEngine
+
 	// Async backfill state: set after backfillResolvedPathsAsync completes.
 	backfillComplete atomic.Bool
 	// Progress tracking for async backfill (total pending and processed so far).
@@ -304,6 +307,7 @@ func NewPacketStore(db *DB, cfg *PacketStoreConfig, cacheTTLs ...map[string]inte
 		spTxIndex:     make(map[string][]*StoreTx, 4096),
 		advertPubkeys:   make(map[string]int),
 		lastSeenTouched: make(map[string]time.Time),
+		clockSkew:       NewClockSkewEngine(),
 	}
 	if cfg != nil {
 		ps.retentionHours = cfg.RetentionHours
@@ -640,7 +644,7 @@ func (s *PacketStore) touchRelayLastSeen(tx *StoreTx, now time.Time) {
 // trackAdvertPubkey increments the advertPubkeys refcount for ADVERT packets.
 // Must be called under s.mu write lock.
 func (s *PacketStore) trackAdvertPubkey(tx *StoreTx) {
-	if tx.PayloadType == nil || *tx.PayloadType != 4 || tx.DecodedJSON == "" {
+	if tx.PayloadType == nil || *tx.PayloadType != PayloadADVERT || tx.DecodedJSON == "" {
 		return
 	}
 	d := tx.ParsedDecoded()
@@ -661,7 +665,7 @@ func (s *PacketStore) trackAdvertPubkey(tx *StoreTx) {
 // untrackAdvertPubkey decrements the advertPubkeys refcount for ADVERT packets.
 // Must be called under s.mu write lock.
 func (s *PacketStore) untrackAdvertPubkey(tx *StoreTx) {
-	if tx.PayloadType == nil || *tx.PayloadType != 4 || tx.DecodedJSON == "" {
+	if tx.PayloadType == nil || *tx.PayloadType != PayloadADVERT || tx.DecodedJSON == "" {
 		return
 	}
 	var d map[string]interface{}
@@ -5134,7 +5138,7 @@ func (s *PacketStore) computeAnalyticsHashSizes(region string) map[string]interf

 		// Track originator from advert packets (including zero-hop adverts,
 		// keyed by pubKey so same-name nodes don't merge).
-		if tx.PayloadType != nil && *tx.PayloadType == 4 && tx.DecodedJSON != "" {
+		if tx.PayloadType != nil && *tx.PayloadType == PayloadADVERT && tx.DecodedJSON != "" {
 			var d map[string]interface{}
 			if json.Unmarshal([]byte(tx.DecodedJSON), &d) == nil {
 				pk := ""
@@ -6621,6 +6625,9 @@ func (s *PacketStore) GetNodeAnalytics(pubkey string, days int) (*NodeAnalyticsR
 		relayPct = round(float64(relayedCount)*100.0/float64(totalWithPath), 1)
 	}

+	// Compute clock skew (already under RLock).
+	clockSkew := s.getNodeClockSkewLocked(pubkey)
+
 	return &NodeAnalyticsResponse{
 		Node:                node,
 		TimeRange:           TimeRangeResp{From: fromISO, To: toISO, Days: days},
@@ -6644,6 +6651,7 @@ func (s *PacketStore) GetNodeAnalytics(pubkey string, days int) (*NodeAnalyticsR
 			UniquePeers:         len(peerSlice),
 			AvgPacketsPerDay:    avgPacketsPerDay,
 		},
+		ClockSkew: clockSkew,
 	}, nil
 }

@@ -468,6 +468,7 @@ type NodeAnalyticsResponse struct {
 	PeerInteractions    []PeerInteraction       `json:"peerInteractions"`
 	UptimeHeatmap       []HeatmapCell           `json:"uptimeHeatmap"`
 	ComputedStats       ComputedNodeStats       `json:"computedStats"`
+	ClockSkew           *NodeClockSkew          `json:"clockSkew,omitempty"`
 }

 // ─── Analytics — RF ────────────────────────────────────────────────────────────