/* * SPDX-License-Identifier: MIT * MeshTimeSync - mesh clock consensus from Ed25519-signed advert timestamps. * See MeshTimeSync.h for the role contract, ARCHITECTURE.md for the design. */ #include "MeshTimeSync.h" #include #include void MeshTimeSync::reset(uint32_t build_epoch) { memset(_slots, 0, sizeof(_slots)); _build_epoch = build_epoch; _next_eval_uptime = 0; _suppress_uptime = 0; _suppressed = false; _pedigree_uptime = 0; _pedigree = false; _last_step_uptime = 0; _stepped_once = false; _evals = _abstains = _steps = _bootstrap_steps = _backward_skips = 0; _last_step_delta = 0; _last_step_wall = 0; } MeshTimeSync::Slot *MeshTimeSync::findSlot(const uint8_t *prefix) { for (int i = 0; i < MESHTIMESYNC_TABLE_SIZE; i++) { if (_slots[i].used && memcmp(_slots[i].prefix, prefix, sizeof(_slots[i].prefix)) == 0) { return &_slots[i]; } } return nullptr; } bool MeshTimeSync::slotTenured(const Slot &s, uint32_t uptime_secs) const { return (uptime_secs - s.first_uptime) >= TENURE_SECS && s.count >= TENURE_MIN_ADVERTS; } bool MeshTimeSync::slotEligible(const Slot &s, uint32_t uptime_secs) const { return s.used && slotTenured(s, uptime_secs) && (uptime_secs - s.arrival_uptime) <= MAX_SAMPLE_AGE_SECS; } int64_t MeshTimeSync::slotSkew(const Slot &s, uint32_t local_time, uint32_t uptime_secs) const { /* Project the sender's clock forward by our elapsed uptime, then compare * with our wall clock. Uptime anchor keeps this valid across own steps. */ return (int64_t)s.advert_ts + (int64_t)(uptime_secs - s.arrival_uptime) - (int64_t)local_time; } void MeshTimeSync::onAdvertHeard(const uint8_t *pubkey, uint32_t advert_ts, uint8_t hops, uint32_t uptime_secs) { if (hops > HOP_CAP) return; Slot *s = findSlot(pubkey); if (s) { if (advert_ts <= s->advert_ts) return; /* per-sender monotonic dedup */ uint32_t d_up = uptime_secs - s->arrival_uptime; uint32_t d_ts = advert_ts - s->advert_ts; int64_t err = (int64_t)d_ts - (int64_t)d_up; int64_t tol = CONSISTENCY_BASE_SECS + ((int64_t)d_up * CONSISTENCY_PPM) / 1000000; if (err > tol || err < -tol) { /* Sender rebooted, corrected itself, or is lying — re-earn * tenure from scratch. >150 ppm is beyond crystal physics. */ s->first_uptime = uptime_secs; s->count = 1; } else if (s->count < 0xFFFF) { s->count++; } s->advert_ts = advert_ts; s->arrival_uptime = uptime_secs; s->hops = hops; return; } for (int i = 0; i < MESHTIMESYNC_TABLE_SIZE; i++) { if (!_slots[i].used) { s = &_slots[i]; break; } } if (!s) { /* Hop-priority admission: a new sender may only displace a young * (not tenure-eligible) entry farther than it; mature entries are * protected unless silent > 24 h. Among candidates evict farthest, * then lowest count, then stalest. (Naive LRU churns hub nodes to * zero eligible voters.) */ for (int i = 0; i < MESHTIMESYNC_TABLE_SIZE; i++) { Slot &c = _slots[i]; bool young = !slotTenured(c, uptime_secs); bool silent = (uptime_secs - c.arrival_uptime) > MATURE_SILENT_EVICT_SECS; if (!young && !silent) continue; if (c.hops <= hops) continue; if (s == nullptr || c.hops > s->hops || (c.hops == s->hops && c.count < s->count) || (c.hops == s->hops && c.count == s->count && c.arrival_uptime < s->arrival_uptime)) { s = &c; } } if (!s) return; /* table full of closer/protected senders — drop */ } memcpy(s->prefix, pubkey, sizeof(s->prefix)); s->advert_ts = advert_ts; s->arrival_uptime = uptime_secs; s->first_uptime = uptime_secs; s->count = 1; s->hops = hops; s->used = 1; } MeshTimeSync::Consensus MeshTimeSync::computeConsensus(uint32_t local_time, uint32_t uptime_secs, bool bootstrap) const { Consensus c; memset(&c, 0, sizeof(c)); /* Collect eligible votes. Bootstrap relaxes tenure: any sender with a * fresh sample may vote (the table is RAM-only, so after the reboot that * dead-ended the clock everything in it is freshly heard anyway). */ int64_t skews[MESHTIMESYNC_TABLE_SIZE]; int32_t radii[MESHTIMESYNC_TABLE_SIZE]; int n = 0; for (int i = 0; i < MESHTIMESYNC_TABLE_SIZE; i++) { const Slot &s = _slots[i]; if (!s.used) continue; if ((uptime_secs - s.arrival_uptime) > MAX_SAMPLE_AGE_SECS) continue; if (!bootstrap && !slotTenured(s, uptime_secs)) continue; skews[n] = slotSkew(s, local_time, uptime_secs); radii[n] = RADIUS_BASE_SECS + RADIUS_PER_HOP_SECS * s.hops; n++; } c.eligible = (uint8_t)n; if (n == 0) return c; /* Marzullo endpoint sweep — interval intersection with the most votes. * No absolute outlier thresholds against our own clock: clustering does * the rejection, so an epoch-0 local clock still finds the true cluster. */ int64_t val[2 * MESHTIMESYNC_TABLE_SIZE]; int8_t typ[2 * MESHTIMESYNC_TABLE_SIZE]; /* +1 = start, -1 = end */ int m = 0; for (int i = 0; i < n; i++) { val[m] = skews[i] - radii[i]; typ[m] = 1; m++; val[m] = skews[i] + radii[i]; typ[m] = -1; m++; } /* Insertion sort by value; starts before ends at equal values so * touching intervals count as overlapping. */ for (int i = 1; i < m; i++) { int64_t v = val[i]; int8_t t = typ[i]; int j = i - 1; while (j >= 0 && (val[j] > v || (val[j] == v && typ[j] < t))) { val[j + 1] = val[j]; typ[j + 1] = typ[j]; j--; } val[j + 1] = v; typ[j + 1] = t; } int cur = 0, best = 0, best_idx = 0; for (int i = 0; i < m; i++) { cur += typ[i]; if (typ[i] > 0 && cur > best) { best = cur; best_idx = i; } } int64_t lo = val[best_idx]; int64_t hi = val[best_idx + 1]; /* next endpoint always exists and, at the * maximum, is an end — a start would have * raised the count past `best` */ c.valid = true; c.votes_for = (uint8_t)best; c.votes_against = (uint8_t)(n - best); c.mid = (lo + hi) / 2; c.radius = (int32_t)((hi - lo) / 2); return c; } MeshTimeSync::Verdict MeshTimeSync::evaluateNow(uint32_t local_time, uint32_t uptime_secs) const { Verdict v; memset(&v, 0, sizeof(v)); bool boot = isBootstrap(local_time); v.bootstrap = boot; v.consensus = computeConsensus(local_time, uptime_secs, boot); if (!v.consensus.valid) { v.type = VERDICT_ABSTAIN; v.reason = REASON_NO_DATA; return v; } if (boot) { if (v.consensus.votes_for < BOOTSTRAP_QUORUM) { v.type = VERDICT_ABSTAIN; v.reason = REASON_NO_QUORUM; return v; } /* Step to the cluster's low edge: from below, all later refinement * is forward steps, which are always monotonicity-safe. */ int64_t delta = v.consensus.mid - RADIUS_BASE_SECS; if (delta < STEP_TRIGGER_SECS) { v.type = VERDICT_NONE; v.reason = REASON_IN_BAND; return v; } if (isSuppressed(uptime_secs)) { /* suppression gates bootstrap too */ v.type = VERDICT_NONE; v.reason = REASON_SUPPRESSED; return v; } if (_stepped_once && (uptime_secs - _last_step_uptime) < STEP_INTERVAL_SECS) { v.type = VERDICT_NONE; v.reason = REASON_RATE_LIMITED; return v; } v.type = VERDICT_STEP; v.delta = delta; /* bootstrap exempt from the ±1 h cap */ return v; } if (v.consensus.eligible < MESHTIMESYNC_QUORUM) { v.type = VERDICT_ABSTAIN; v.reason = REASON_NO_QUORUM; return v; } if ((int)v.consensus.votes_for * 2 <= (int)v.consensus.eligible) { v.type = VERDICT_ABSTAIN; v.reason = REASON_NO_MAJORITY; return v; } int64_t mag = v.consensus.mid < 0 ? -v.consensus.mid : v.consensus.mid; if (mag < STEP_TRIGGER_SECS) { v.type = VERDICT_NONE; v.reason = REASON_IN_BAND; return v; } if (isSuppressed(uptime_secs)) { v.type = VERDICT_NONE; v.reason = REASON_SUPPRESSED; return v; } if (_stepped_once && (uptime_secs - _last_step_uptime) < STEP_INTERVAL_SECS) { v.type = VERDICT_NONE; v.reason = REASON_RATE_LIMITED; return v; } if (_pedigree) { /* Physics veto: with a trusted sync + continuous uptime since, a * real crystal cannot have drifted further than 300 ppm allows. */ int64_t envelope = ((int64_t)(uptime_secs - _pedigree_uptime) * PEDIGREE_PPM) / 1000000 + PEDIGREE_BASE_SECS; if (mag > envelope) { v.type = VERDICT_ABSTAIN; v.reason = REASON_PEDIGREE; return v; } } int64_t delta = v.consensus.mid; if (delta > STEP_CAP_SECS) delta = STEP_CAP_SECS; if (delta < -STEP_CAP_SECS) delta = -STEP_CAP_SECS; v.type = VERDICT_STEP; v.delta = delta; return v; } MeshTimeSync::Verdict MeshTimeSync::tick(uint32_t local_time, uint32_t uptime_secs) { Verdict v; if (_next_eval_uptime == 0 || uptime_secs < _next_eval_uptime) { /* First call arms the pacing timer (skips a junk no-data abstain * at boot); later calls no-op between evaluations. */ if (_next_eval_uptime == 0) { _next_eval_uptime = uptime_secs + EVAL_INTERVAL_SECS; } memset(&v, 0, sizeof(v)); return v; } _next_eval_uptime = uptime_secs + EVAL_INTERVAL_SECS; _evals++; v = evaluateNow(local_time, uptime_secs); if (v.type == VERDICT_ABSTAIN) _abstains++; return v; } void MeshTimeSync::noteManualSync(uint32_t uptime_secs) { _suppress_uptime = uptime_secs; _suppressed = true; _pedigree_uptime = uptime_secs; _pedigree = true; } void MeshTimeSync::noteGPSSync(uint32_t uptime_secs) { _pedigree_uptime = uptime_secs; _pedigree = true; } void MeshTimeSync::noteStepApplied(int64_t delta, uint32_t local_time, uint32_t uptime_secs, bool bootstrap) { _last_step_uptime = uptime_secs; _stepped_once = true; _steps++; if (bootstrap) _bootstrap_steps++; _last_step_delta = delta; _last_step_wall = local_time; } bool MeshTimeSync::isSuppressed(uint32_t uptime_secs) const { return _suppressed && (uptime_secs - _suppress_uptime) < SUPPRESS_SECS; } uint32_t MeshTimeSync::suppressRemaining(uint32_t uptime_secs) const { if (!isSuppressed(uptime_secs)) return 0; return SUPPRESS_SECS - (uptime_secs - _suppress_uptime); } const char *MeshTimeSync::reasonStr(Reason r) { switch (r) { case REASON_IN_BAND: return "in-band"; case REASON_NO_DATA: return "no-data"; case REASON_NO_QUORUM: return "no-quorum"; case REASON_NO_MAJORITY: return "no-majority"; case REASON_SUPPRESSED: return "suppressed"; case REASON_RATE_LIMITED: return "rate-limited"; case REASON_PEDIGREE: return "pedigree-veto"; default: return "-"; } } /* Clamp an int64 skew to a printable long (display only). */ static long clampl(int64_t v) { if (v > 2000000000LL) return 2000000000L; if (v < -2000000000LL) return -2000000000L; return (long)v; } int MeshTimeSync::formatStatus(char *out, size_t cap, uint32_t local_time, uint32_t uptime_secs, bool enabled) const { Verdict v = evaluateNow(local_time, uptime_secs); const Consensus &c = v.consensus; char verdict[40]; if (v.type == VERDICT_STEP) { snprintf(verdict, sizeof(verdict), "step%+ld%s", clampl(v.delta), v.bootstrap ? " (bootstrap)" : ""); } else if (v.type == VERDICT_NONE && v.reason == REASON_IN_BAND) { int64_t mag = c.mid < 0 ? -c.mid : c.mid; snprintf(verdict, sizeof(verdict), "%s", mag <= DEAD_BAND_SECS ? "ok" : "in-band"); } else if (v.type == VERDICT_NONE) { snprintf(verdict, sizeof(verdict), "hold (%s)", reasonStr(v.reason)); } else { snprintf(verdict, sizeof(verdict), "abstain (%s)", reasonStr(v.reason)); } int pos = snprintf(out, cap, "%s%s eligible=%u votes=%u/%u skew=%+lds r=%lds -> %s", enabled ? "on" : "off", enabled ? "" : " (dry-run)", (unsigned)c.eligible, (unsigned)c.votes_for, (unsigned)c.votes_against, c.valid ? clampl(c.mid) : 0L, c.valid ? (long)c.radius : 0L, verdict); if (pos < 0 || (size_t)pos >= cap) goto full; if (_steps > 0) { pos += snprintf(out + pos, cap - pos, "; steps=%lu last=%+lds", (unsigned long)_steps, clampl(_last_step_delta)); if (pos < 0 || (size_t)pos >= cap) goto full; } if (_backward_skips > 0) { pos += snprintf(out + pos, cap - pos, " backskip=%lu", (unsigned long)_backward_skips); if (pos < 0 || (size_t)pos >= cap) goto full; } if (isSuppressed(uptime_secs)) { pos += snprintf(out + pos, cap - pos, " sup=%luh", (unsigned long)(suppressRemaining(uptime_secs) / 3600)); if (pos < 0 || (size_t)pos >= cap) goto full; } pos += snprintf(out + pos, cap - pos, " evals=%lu abst=%lu", (unsigned long)_evals, (unsigned long)_abstains); if (pos < 0 || (size_t)pos >= cap) goto full; /* Evidence table: one entry per used slot, as many as fit. */ for (int i = 0; i < MESHTIMESYNC_TABLE_SIZE; i++) { const Slot &s = _slots[i]; if (!s.used) continue; int w = snprintf(out + pos, cap - pos, "\r\n %02x%02x h%u n%u %+lds%s", s.prefix[0], s.prefix[1], (unsigned)s.hops, (unsigned)(s.count > 99 ? 99 : s.count), clampl(slotSkew(s, local_time, uptime_secs)), slotEligible(s, uptime_secs) ? " E" : ""); if (w < 0 || (size_t)(pos + w) >= cap) { out[pos] = 0; /* drop the partial entry */ return pos; } pos += w; } return pos; full: out[cap - 1] = 0; return (int)(cap - 1); }