diff --git a/cmd/server/clock_skew.go b/cmd/server/clock_skew.go new file mode 100644 index 00000000..a4df0f5a --- /dev/null +++ b/cmd/server/clock_skew.go @@ -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 +} diff --git a/cmd/server/clock_skew_test.go b/cmd/server/clock_skew_test.go new file mode 100644 index 00000000..d359b5fa --- /dev/null +++ b/cmd/server/clock_skew_test.go @@ -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") + } +} diff --git a/cmd/server/neighbor_graph.go b/cmd/server/neighbor_graph.go index 3cbde61f..c26813dd 100644 --- a/cmd/server/neighbor_graph.go +++ b/cmd/server/neighbor_graph.go @@ -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 := "" diff --git a/cmd/server/neighbor_persist.go b/cmd/server/neighbor_persist.go index 4539ec66..2bce39f7 100644 --- a/cmd/server/neighbor_persist.go +++ b/cmd/server/neighbor_persist.go @@ -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) diff --git a/cmd/server/routes.go b/cmd/server/routes.go index c14aa90e..814b6683 100644 --- a/cmd/server/routes.go +++ b/cmd/server/routes.go @@ -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) { diff --git a/cmd/server/store.go b/cmd/server/store.go index c9c38ee6..9ef65d22 100644 --- a/cmd/server/store.go +++ b/cmd/server/store.go @@ -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 } diff --git a/cmd/server/types.go b/cmd/server/types.go index 7ac25c8c..51af55a3 100644 --- a/cmd/server/types.go +++ b/cmd/server/types.go @@ -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 ────────────────────────────────────────────────────────────