diff --git a/docs/cli_command_availability.md b/docs/cli_command_availability.md index 24d508ac..dcafbcb6 100644 --- a/docs/cli_command_availability.md +++ b/docs/cli_command_availability.md @@ -98,7 +98,7 @@ fix, no WiFi connection, an inactive bridge, or an nRF52 bootloader without | System | [`get role`](cli_commands.md#view-this-nodes-configured-role) | All text CLI roles | Yes | Yes | Yes | | System | [`powersaving`; `powersaving on/off`](cli_commands.md#view-or-change-this-nodes-power-saving-flag-repeater-only) | Supported repeater board | Feature | Feature | Feature | | System | [`get/set reboot.interval`](#reboot-interval) | Full parser | Yes | Yes | Yes | -| Clock sync | [`get/set clock.sync.*`; `clock.sync.mesh now`](cli_commands.md#estimate-and-correct-repeater-time-after-startup) | Repeater; `clock.sync.internet` needs MQTT | Yes | Yes | Yes | +| Clock sync | [`get/set clock.sync.*`; `clock.sync.mesh now`](cli_commands.md#estimate-and-correct-infrastructure-node-time-after-startup) | Repeater, sensor, and room server; `clock.sync.internet` needs MQTT repeater | Yes | Yes | Yes | | Routing | [`get/set repeat`](cli_commands.md#view-or-change-this-nodes-repeat-flag) | Forwarding-capable role | Yes | Yes | Yes | | Routing | [`get/set path.hash.mode`](cli_commands.md#view-or-change-this-nodes-advert-path-hash-size) | Role that supports path-hash selection | Yes | Yes | Yes | | Routing | [`get/set loop.detect`](cli_commands.md#view-or-change-this-nodes-loop-detection) | Repeater | Yes | Yes | Yes | @@ -254,7 +254,7 @@ fix, no WiFi connection, an inactive bridge, or an nRF52 bootloader without | System | [`get role`](cli_commands.md#view-this-nodes-configured-role) | All text CLI roles | Yes | Yes | Yes | Yes | Yes | Yes | Yes | | System | [`powersaving`; `powersaving on/off`](cli_commands.md#view-or-change-this-nodes-power-saving-flag-repeater-only) | Supported repeater board | Feature | Feature | Feature | No | No | Feature | Feature | | System | [`get/set reboot.interval`](#reboot-interval) | Full parser | Yes | Yes | Yes | No | No | Yes | Yes | -| Clock sync | [`get/set clock.sync.*`; `clock.sync.mesh now`](cli_commands.md#estimate-and-correct-repeater-time-after-startup) | Repeater; `clock.sync.internet` needs MQTT | Yes | Yes | Yes | No | Yes | Yes | Yes | +| Clock sync | [`get/set clock.sync.*`; `clock.sync.mesh now`](cli_commands.md#estimate-and-correct-infrastructure-node-time-after-startup) | Repeater, sensor, and room server; `clock.sync.internet` needs MQTT repeater | Yes | Yes | Yes | No | Yes | Yes | Yes | | Routing | [`get/set repeat`](cli_commands.md#view-or-change-this-nodes-repeat-flag) | Forwarding-capable role | Yes | Yes | Yes | Yes | Yes | Yes | Yes | | Routing | [`get/set path.hash.mode`](cli_commands.md#view-or-change-this-nodes-advert-path-hash-size) | Full parser | Yes | Yes | Yes | No | No | Yes | Yes | | Routing | [`get/set loop.detect`](cli_commands.md#view-or-change-this-nodes-loop-detection) | Repeater, full common parser | Yes | Yes | Yes | No | No | Yes | Yes | diff --git a/docs/cli_commands.md b/docs/cli_commands.md index 89a42114..7c9351cd 100644 --- a/docs/cli_commands.md +++ b/docs/cli_commands.md @@ -830,7 +830,7 @@ send text.flood checking ridge link --- -#### Estimate and correct repeater time after startup +#### Estimate and correct infrastructure-node time after startup **Usage:** - `get clock.sync` @@ -850,23 +850,23 @@ send text.flood checking ridge link - `set clock.sync.samples <3-16>` **Defaults:** -- `clock.sync.mesh`: `on` for nRF52 repeaters; `off` for other builds +- `clock.sync.mesh`: `on` for all repeater, sensor, and room-server builds - `clock.sync.mesh.edge`: `on` - `clock.sync.internet`: `off` - `clock.sync.drift`: `3600` seconds - `clock.sync.samples`: `9` -When either source is enabled, the repeater makes its first clock-bootstrap +When either source is enabled, the node makes its first clock-bootstrap attempt after 30 minutes of uptime, or immediately when the configured number of fresh evidence sources has been collected, whichever comes first. A successful estimate changes the RTC only when the absolute difference is **greater than** `clock.sync.drift`; correction can move the clock forward or backward. A valid estimate within the threshold counts as a successful sync without changing the clock. Seven days after each successful estimate, the -repeater evaluates time again; the seven-day deadline therefore starts from the +node evaluates time again; the seven-day deadline therefore starts from the last successful estimate rather than from boot. This is a lazy uptime deadline: the check runs on the first normal loop/wake after it becomes due and does not -wake the device by itself. If no source or consensus is available, the repeater +wake the device by itself. If no source or consensus is available, the node retries every 30 minutes, and newly collected evidence triggers another immediate evaluation once the configured source count is present. Every reboot starts with the initial bootstrap attempt. An existing saved setting always @@ -975,13 +975,15 @@ one sender can claim multiple names and inflate the edge-mode vote count. `clock.sync.internet` is available on WiFi MQTT repeater-observer builds. Its initial and seven-day queries run on the MQTT/WiFi task and are read-only until the repeater applies the configured drift test. Failed queries retry after 30 -minutes. On other repeater builds, the preference can be stored but status -reports that internet time is unavailable. MQTT builds retain their existing -startup NTP behavior required for MQTT/TLS/JWT operation; this setting controls -the additional delayed drift checks. Startup NTP is always preferred when it +minutes. On other infrastructure-node builds, the preference can be stored but +status reports that internet time is unavailable. MQTT builds retain their +existing startup NTP behavior required for MQTT/TLS/JWT operation; this setting +controls the additional delayed drift checks. Startup NTP is always preferred when it succeeds, regardless of this setting. -Changing any `clock.sync.*` setting starts a new attempt for the current boot. +Sensor and room-server builds support mesh clock consensus and report +`clock.sync.internet` as unavailable. Changing any `clock.sync.*` setting starts +a new attempt for the current boot. Settings are persistent in `/clock_sync`; samples and schedule state are not. A backward correction is intentionally allowed, but peers that already recorded diff --git a/docs/halo_keymind_settings.md b/docs/halo_keymind_settings.md index 4f553e62..7c5bbd4c 100644 --- a/docs/halo_keymind_settings.md +++ b/docs/halo_keymind_settings.md @@ -87,8 +87,8 @@ set flood.retry.ignore none | `flood.filter` | Persistent flood-route rules selected by payload type and optional hop range. Generalized repeaters have 63 FPF7 forward slots; FULL ESP32 room servers have 31. Repeaters store their scope-rewrite phase, shared unordered blacklist, and channel-data compatibility state in the same atomic FPF7 image. `path=blacklist` is intended for forwarding abuse containment, including bulk internet-to-mesh dumping, but truncated path IDs are not authenticated identities. Fixed 240 KB STM32WL repeaters retain compact FPF6 filtering and separate blacklist storage. New generalized repeater tables seed slot 1 with `ota all suspend=tempradio` and slot 2 with an authenticated `#wardriving hops=5+` drop. Direct routes and local receive/logging are unchanged. | `get flood.filter[.n]`, `set flood.filter[.n] [N|N+|N-M|all] [scope=] [require=region] [tx=slow] [suspend=tempradio]`, `del flood.filter.|all`; repeater only: `get/set/del flood.filter.blacklist[.n]` | `set flood.filter grp_txt all scope=local tx=slow` | | `flood.rule` | Live alias for extended `flood.filter` on rule-engine repeaters and FULL ESP32 room servers. A row can AND packet type, hop range, optional channel authentication, ordered 1/2/3-byte pbyte source prefix, and original scope/region conditions, then drop, rewrite to a direct scope or configured region, enforce a per-row rate, and/or stop lower-priority FPF7 rules. `channel=*` means no channel condition and `type=any channel=*` matches every flood payload type. `scope=BlackHole86` directly derives a regionless sink scope; `region=BlackHole86` would require a configured flood-allowed region. All rows match the original receive-time packet; higher `priority` runs first and lower slot breaks a tie. Repeated rows with one channel key share a per-packet authentication result. Persistent FPF7 stores canonical region names, so region ID reorder or reuse cannot retarget a rule. A missing saved region makes its `region=` rewrite and paired `stop` inert until the name returns, allowing lower safety rows to run. Fixed 240 KB STM32WL profiles keep FPF6 and do not expose this alias; partition sizes are unchanged. | `get flood.rule[.n]`, `set flood.rule[.n] type= [hops=...] [channel=...] [prefix=...] [in=...] [priority=0-255]`, `del flood.rule.|all` | `set flood.rule.2 type=grp_data hops=4+ channel=#rgdata in=none scope=BlackHole86` | | `flood.moderation` | Decrypts keyed `GRP_TXT` channels and applies drop, per-username messages/minute, and maximum-hop controls, optionally matched against the first 1-3 path hashes. Supports `public`, `#channel`, and 128/256-bit channel keys. Sender names and truncated path hashes are moderation hints, not authenticated identities. | `get flood.moderation[.n]`, `set flood.moderation[.n] [path=...]`, `del flood.moderation.|all` | `set flood.moderation public "Noisy User" rate=5/min hops=4` | -| `clock.sync.mesh` | Defaults on for nRF52 repeaters and off for other builds; a saved setting overrides that default. It estimates UTC as soon as the configured number of fresh signed-advert or valid Public-channel sources is collected, with a 30-minute bootstrap/retry timer when evidence is still insufficient, then repeats lazily seven days after each successful estimate. New evidence retriggers evaluation after a no-consensus result. `clock.sync.mesh now` queues an immediate LoRa-only attempt without bypassing quorum or source suppression. Only timestamps from firmware build time through build time plus ten years are recorded. Successful CLI, GPS, or WiFi/NTP clock updates suppress LoRa time collection until reboot; after reboot LoRa is the fallback if NTP cannot sync. Status reports the reason a clock was not set; its `.table` and `.1` through `.16` forms inspect collected samples. | `get clock.sync.mesh`, `set clock.sync.mesh `, `clock.sync.mesh now`, `get clock.sync.status[.table|.1-.16]` | `set clock.sync.mesh on` | -| `clock.sync.mesh.edge` | Defaults on so edge repeaters can collect clock evidence when all packets arrive through one relay path. Verified evidence is observed before the forwarding decision, so `repeat off` and forwarding filters do not prevent collection. Signed adverts are deduplicated by public key and Public-channel timestamps by case-insensitive display name; all may share one receive path. Public display names are unauthenticated and can be spoofed. Changing this setting clears current clock samples. | `get clock.sync.mesh.edge`, `set clock.sync.mesh.edge ` | `set clock.sync.mesh.edge on` | +| `clock.sync.mesh` | Defaults on for all repeater, sensor, and room-server builds; a saved setting overrides that default. It estimates UTC as soon as the configured number of fresh signed-advert or valid Public-channel sources is collected, with a 30-minute bootstrap/retry timer when evidence is still insufficient, then repeats lazily seven days after each successful estimate. New evidence retriggers evaluation after a no-consensus result. `clock.sync.mesh now` queues an immediate LoRa-only attempt without bypassing quorum or source suppression. Only timestamps from firmware build time through build time plus ten years are recorded. Successful CLI, GPS, or WiFi/NTP clock updates suppress LoRa time collection until reboot; after reboot LoRa is the fallback if NTP cannot sync. Status reports the reason a clock was not set; its `.table` and `.1` through `.16` forms inspect collected samples. | `get clock.sync.mesh`, `set clock.sync.mesh `, `clock.sync.mesh now`, `get clock.sync.status[.table|.1-.16]` | `set clock.sync.mesh on` | +| `clock.sync.mesh.edge` | Defaults on so edge infrastructure nodes can collect clock evidence when all packets arrive through one relay path. Verified evidence is observed before the forwarding decision, so disabled forwarding and forwarding filters do not prevent collection. Signed adverts are deduplicated by public key and Public-channel timestamps by case-insensitive display name; all may share one receive path. Public display names are unauthenticated and can be spoofed. Changing this setting clears current clock samples. | `get clock.sync.mesh.edge`, `set clock.sync.mesh.edge ` | `set clock.sync.mesh.edge on` | | `clock.sync.internet` | Adds a read-only internet/NTP estimate at the same initial and seven-day checks on WiFi MQTT repeater-observer builds. Other builds retain the setting but report internet unavailable. | `get clock.sync.internet`, `set clock.sync.internet ` | `set clock.sync.internet on` | | `clock.sync.drift` | Absolute correction threshold in seconds. The clock is moved forward or backward only when the estimate differs by more than this value. | `get clock.sync.drift`, `set clock.sync.drift <30-86400>` | `set clock.sync.drift 3600` | | `clock.sync.samples` | Minimum fresh evidence count before mesh time can be used: distinct receive paths in normal mode or distinct advert keys/Public-channel display names in edge mode. A strict majority of all fresh samples is also required. Range `3-16`; default `9`. | `get clock.sync.samples`, `set clock.sync.samples <3-16>` | `set clock.sync.samples 9` | diff --git a/examples/simple_repeater/MyMesh.h b/examples/simple_repeater/MyMesh.h index d8cc08c0..84ee08ca 100644 --- a/examples/simple_repeater/MyMesh.h +++ b/examples/simple_repeater/MyMesh.h @@ -212,11 +212,7 @@ struct NeighbourInfo { #define CLOCK_SYNC_SAMPLE_SLOTS 16 #endif #ifndef CLOCK_SYNC_MESH_DEFAULT_ENABLED - #if defined(NRF52_PLATFORM) - #define CLOCK_SYNC_MESH_DEFAULT_ENABLED 1 - #else - #define CLOCK_SYNC_MESH_DEFAULT_ENABLED 0 - #endif + #define CLOCK_SYNC_MESH_DEFAULT_ENABLED 1 #endif #ifndef CLOCK_SYNC_MESH_EDGE_DEFAULT_ENABLED #define CLOCK_SYNC_MESH_EDGE_DEFAULT_ENABLED 1 diff --git a/examples/simple_room_server/MyMesh.cpp b/examples/simple_room_server/MyMesh.cpp index 8edec37d..03f981a7 100644 --- a/examples/simple_room_server/MyMesh.cpp +++ b/examples/simple_room_server/MyMesh.cpp @@ -524,6 +524,7 @@ bool MyMesh::allowPacketForward(const mesh::Packet *packet) { _ms->getMillis()); } #endif + _clock_sync.observeAcceptedFlood(packet); return true; } @@ -992,6 +993,28 @@ void MyMesh::onAckRecv(mesh::Packet *packet, uint32_t ack_crc) { } } +void MyMesh::onAdvertRecv(mesh::Packet* packet, const mesh::Identity& id, + uint32_t timestamp, const uint8_t* app_data, + size_t app_data_len) { + mesh::Mesh::onAdvertRecv(packet, id, timestamp, app_data, app_data_len); + _clock_sync.observeVerifiedAdvert(packet, id, timestamp); + +#if defined(WITH_MQTT_NEIGHBORS) + bool is_share = packet->hasTransportCodes() + && packet->transport_codes[0] == 0 && packet->transport_codes[1] == 0; + if (packet->getPathHashCount() == 0 && !is_share) { + AdvertDataParser parser(app_data, app_data_len); + if (parser.isValid() && parser.getType() == ADV_TYPE_REPEATER) { + putNeighbour(id, timestamp, packet->getSNR()); + } + } +#endif +} + +void MyMesh::onGroupPacketRecv(mesh::Packet* packet) { + _clock_sync.observeGroupPacket(packet); +} + #if defined(WITH_MQTT_NEIGHBORS) #define CTL_TYPE_NODE_DISCOVER_REQ 0x80 @@ -1022,26 +1045,6 @@ void MyMesh::putNeighbour(const mesh::Identity &id, uint32_t timestamp, float sn neighbour->snr = (int8_t)(snr * 4); } -static bool isShare(const mesh::Packet *packet) { - if (packet->hasTransportCodes()) { - return packet->transport_codes[0] == 0 && packet->transport_codes[1] == 0; // codes { 0, 0 } means 'send to nowhere' - } - return false; -} - -void MyMesh::onAdvertRecv(mesh::Packet *packet, const mesh::Identity &id, uint32_t timestamp, - const uint8_t *app_data, size_t app_data_len) { - mesh::Mesh::onAdvertRecv(packet, id, timestamp, app_data, app_data_len); // chain to super impl - - // if this a zero hop advert (and not via 'Share'), add it to neighbours - if (packet->getPathHashCount() == 0 && !isShare(packet)) { - AdvertDataParser parser(app_data, app_data_len); - if (parser.isValid() && parser.getType() == ADV_TYPE_REPEATER) { // just keep neigbouring Repeaters - putNeighbour(id, timestamp, packet->getSNR()); - } - } -} - void MyMesh::onControlDataRecv(mesh::Packet* packet) { uint8_t type = packet->payload[0] & 0xF0; // just test upper 4 bits // A room server is ADV_TYPE_ROOM, so it does NOT answer node-discover requests @@ -1098,6 +1101,7 @@ MyMesh::MyMesh(mesh::MainBoard &board, mesh::Radio &radio, mesh::MillisecondCloc : mesh::Mesh(radio, ms, rng, rtc, *createObserverPacketManager(32), tables), region_map(key_store), temp_map(key_store), _cli(board, rtc, sensors, region_map, acl, &_prefs, this), + _clock_sync(radio, ms, rtc, acl, sensors, _prefs.tx_delay_factor, this), telemetry(MAX_PACKET_PAYLOAD - 4) #ifdef WITH_MQTT_BRIDGE , bridge(nullptr) @@ -1208,6 +1212,7 @@ void MyMesh::begin(FILESYSTEM *fs) { acl.load(_fs, self_id); region_map.load(_fs); + _clock_sync.begin(_fs); #if defined(MESHCORE_ESP32_FULL_PROFILE) flood_rules.begin(_fs, ®ion_map); #endif @@ -2039,7 +2044,9 @@ void MyMesh::handleCommand(uint32_t sender_timestamp, char *command, char *reply // handled by the FULL-profile persistent flood rule engine } #endif - else { + else if (_clock_sync.handleCommand(command, reply)) { + // handled by the role-independent mesh clock synchronizer + } else { _cli.handleCommand(sender_timestamp, command, reply); // common CLI commands } } @@ -2053,6 +2060,7 @@ void MyMesh::loop() { // MQTT processing can take time, so we prioritize radio reception mesh::Mesh::loop(); _cli.loop(); + _clock_sync.loop(); #ifdef WITH_MQTT_BRIDGE // bridge.loop() is now handled by FreeRTOS task on Core 0 - no need to call it here #endif diff --git a/examples/simple_room_server/MyMesh.h b/examples/simple_room_server/MyMesh.h index 5296060c..2067c605 100644 --- a/examples/simple_room_server/MyMesh.h +++ b/examples/simple_room_server/MyMesh.h @@ -19,6 +19,7 @@ #include #include #include +#include #if defined(ESP32_PLATFORM) || defined(USER_GPIO_CONTROL) #include #endif @@ -122,7 +123,8 @@ struct NeighbourInfo { int8_t snr; // multiplied by 4, user should divide to get float value }; -class MyMesh : public mesh::Mesh, public CommonCLICallbacks +class MyMesh : public mesh::Mesh, public CommonCLICallbacks, + public mesh::MeshClockSyncCallbacks #ifdef WITH_WEBCONFIG , public WebConfigServer::Callbacks #endif @@ -145,6 +147,7 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks mesh::LogicalMessageCache recent_room_posts; mesh::RemoteCliReplyCache remote_cli_reply_cache; CommonCLI _cli; + mesh::MeshClockSync _clock_sync; #if defined(ESP32_PLATFORM) || defined(USER_GPIO_CONTROL) UserGpioReplyTracker _gpio_reply_tracker; #endif @@ -299,8 +302,9 @@ protected: void onPeerDataRecv(mesh::Packet* packet, uint8_t type, int sender_idx, const uint8_t* secret, uint8_t* data, size_t len) override; bool onPeerPathRecv(mesh::Packet* packet, int sender_idx, const uint8_t* secret, uint8_t* path, uint8_t path_len, uint8_t extra_type, uint8_t* extra, uint8_t extra_len) override; void onAckRecv(mesh::Packet* packet, uint32_t ack_crc) override; -#if defined(WITH_MQTT_NEIGHBORS) void onAdvertRecv(mesh::Packet* packet, const mesh::Identity& id, uint32_t timestamp, const uint8_t* app_data, size_t app_data_len) override; + void onGroupPacketRecv(mesh::Packet* packet) override; +#if defined(WITH_MQTT_NEIGHBORS) void onControlDataRecv(mesh::Packet* packet) override; #endif @@ -330,6 +334,14 @@ public: PrefsSaveRouting::Scope scope = PrefsSaveRouting::Scope::Common) override { _cli.savePrefs(_fs, scope); } + void onManualClockSet() override { _clock_sync.onManualClockSet(); } + bool hasAuthoritativeClock() const override { +#ifdef WITH_MQTT_BRIDGE + return bridge != nullptr && bridge->hasNtpTime(); +#else + return false; +#endif + } bool sendFloodScoped(const TransportKey& scope, mesh::Packet* pkt, uint32_t delay_millis, uint8_t path_hash_size); diff --git a/examples/simple_sensor/SensorMesh.cpp b/examples/simple_sensor/SensorMesh.cpp index 8f7b3fb1..5850c330 100644 --- a/examples/simple_sensor/SensorMesh.cpp +++ b/examples/simple_sensor/SensorMesh.cpp @@ -350,6 +350,7 @@ bool SensorMesh::getCADEnabled() const { bool SensorMesh::allowPacketForward(const mesh::Packet* packet) { if (_prefs.disable_fwd) return false; if (packet->isRouteFlood() && packet->getPathHashCount() >= _prefs.flood_max) return false; + _clock_sync.observeAcceptedFlood(packet); return true; } @@ -629,12 +630,25 @@ void SensorMesh::handleCommand(uint32_t sender_timestamp, char* command, char* r board.setGpio(val); } sprintf(reply, "%x", board.getGpio()); + } else if (_clock_sync.handleCommand(command, reply)) { + // handled by the role-independent mesh clock synchronizer } else{ _cli.handleCommand(sender_timestamp, command, reply); // common CLI commands updateGpsTelemetryPolicy(); } } +void SensorMesh::onAdvertRecv(mesh::Packet* packet, const mesh::Identity& id, + uint32_t timestamp, const uint8_t* app_data, + size_t app_data_len) { + mesh::Mesh::onAdvertRecv(packet, id, timestamp, app_data, app_data_len); + _clock_sync.observeVerifiedAdvert(packet, id, timestamp); +} + +void SensorMesh::onGroupPacketRecv(mesh::Packet* packet) { + _clock_sync.observeGroupPacket(packet); +} + void SensorMesh::onAnonDataRecv(mesh::Packet* packet, const uint8_t* secret, const mesh::Identity& sender, uint8_t* data, size_t len) { if (packet->getPayloadType() == PAYLOAD_TYPE_ANON_REQ) { // received an initial request by a possible admin client (unknown at this stage) uint32_t timestamp; @@ -886,6 +900,7 @@ SensorMesh::SensorMesh(mesh::MainBoard& board, mesh::Radio& radio, mesh::Millise : mesh::Mesh(radio, ms, rng, rtc, *new StaticPoolPacketManager(32), tables), region_map(key_store), _cli(board, rtc, sensors, region_map, acl, &_prefs, this), + _clock_sync(radio, ms, rtc, acl, sensors, _prefs.tx_delay_factor, this), telemetry(MAX_PACKET_PAYLOAD - 4) { next_local_advert = next_flood_advert = 0; @@ -944,6 +959,7 @@ void SensorMesh::begin(FILESYSTEM* fs) { acl.load(_fs, self_id); region_map.load(_fs); + _clock_sync.begin(_fs); updateGpsTelemetryPolicy(); // establish default-scope @@ -1126,6 +1142,7 @@ bool SensorMesh::getGPS(uint8_t channel, float& lat, float& lon, float& alt) { void SensorMesh::loop() { _cli.loop(); mesh::Mesh::loop(); + _clock_sync.loop(); if (next_flood_advert && millisHasNowPassed(next_flood_advert)) { mesh::Packet* pkt = createSelfAdvert(); diff --git a/examples/simple_sensor/SensorMesh.h b/examples/simple_sensor/SensorMesh.h index 8f01accb..fb39b84d 100644 --- a/examples/simple_sensor/SensorMesh.h +++ b/examples/simple_sensor/SensorMesh.h @@ -20,6 +20,7 @@ #include #include #include +#include #if defined(ESP32_PLATFORM) || defined(USER_GPIO_CONTROL) #include #endif @@ -49,7 +50,8 @@ #define MAX_SEARCH_RESULTS 8 #define MAX_CONCURRENT_ALERTS 4 -class SensorMesh : public mesh::Mesh, public CommonCLICallbacks { +class SensorMesh : public mesh::Mesh, public CommonCLICallbacks, + public mesh::MeshClockSyncCallbacks { public: SensorMesh(mesh::MainBoard& board, mesh::Radio& radio, mesh::MillisecondClock& ms, mesh::RNG& rng, mesh::RTCClock& rtc, mesh::MeshTables& tables); void begin(FILESYSTEM* fs); @@ -68,6 +70,7 @@ public: PrefsSaveRouting::Scope scope = PrefsSaveRouting::Scope::Common) override { _cli.savePrefs(_fs, scope); } + void onManualClockSet() override { _clock_sync.onManualClockSet(); } bool formatFileSystem() override; void sendSelfAdvertisement(int delay_millis, bool flood) override; void updateAdvertTimer() override; @@ -172,6 +175,10 @@ protected: bool onPeerPathRecv(mesh::Packet* packet, int sender_idx, const uint8_t* secret, uint8_t* path, uint8_t path_len, uint8_t extra_type, uint8_t* extra, uint8_t extra_len) override; void onControlDataRecv(mesh::Packet* packet) override; void onAckRecv(mesh::Packet* packet, uint32_t ack_crc) override; + void onAdvertRecv(mesh::Packet* packet, const mesh::Identity& id, + uint32_t timestamp, const uint8_t* app_data, + size_t app_data_len) override; + void onGroupPacketRecv(mesh::Packet* packet) override; virtual bool handleIncomingMsg(ClientInfo& from, uint32_t timestamp, uint8_t* data, uint8_t flags, size_t len); void sendAckTo(const ClientInfo& dest, uint32_t ack_hash, uint8_t path_hash_size=1); private: @@ -180,6 +187,7 @@ private: NodePrefs _prefs; ClientACL acl; CommonCLI _cli; + mesh::MeshClockSync _clock_sync; #if defined(ESP32_PLATFORM) || defined(USER_GPIO_CONTROL) UserGpioReplyTracker _gpio_reply_tracker; #endif diff --git a/src/helpers/MeshClockSync.cpp b/src/helpers/MeshClockSync.cpp new file mode 100644 index 00000000..3f1cea80 --- /dev/null +++ b/src/helpers/MeshClockSync.cpp @@ -0,0 +1,908 @@ +#include "MeshClockSync.h" + +#include +#include +#include +#include +#include +#include + +#include +#include + +#ifndef CLOCK_SYNC_MESH_DEFAULT_ENABLED + #define CLOCK_SYNC_MESH_DEFAULT_ENABLED 1 +#endif +#ifndef CLOCK_SYNC_MESH_EDGE_DEFAULT_ENABLED + #define CLOCK_SYNC_MESH_EDGE_DEFAULT_ENABLED 1 +#endif +#ifndef FIRMWARE_BUILD_EPOCH + #define FIRMWARE_BUILD_EPOCH 0UL +#endif + +namespace { + +constexpr char PREFS_FILE[] = "/clock_sync"; +constexpr uint8_t REQUIRED_SAMPLES_MIN = 3; +constexpr uint8_t REQUIRED_SAMPLES_MAX = 16; +constexpr uint8_t REQUIRED_SAMPLES_DEFAULT = 9; +constexpr uint64_t STARTUP_DELAY_MILLIS = 30ULL * 60ULL * 1000ULL; +constexpr uint64_t RETRY_INTERVAL_MILLIS = 30ULL * 60ULL * 1000ULL; +constexpr uint64_t RESYNC_INTERVAL_MILLIS = 7ULL * 24ULL * 60ULL * 60ULL * 1000ULL; +constexpr uint32_t SAMPLE_MAX_AGE_MILLIS = 2UL * 60UL * 60UL * 1000UL; +constexpr uint32_t CONSENSUS_WINDOW_SECONDS = 600UL; +constexpr uint32_t DRIFT_MIN_SECONDS = 30UL; +constexpr uint32_t DRIFT_MAX_SECONDS = 86400UL; +constexpr uint32_t DRIFT_DEFAULT_SECONDS = 3600UL; +constexpr uint16_t VALID_YEARS = 10; + +// Public-channel AES key, zero-padded to the shared-secret buffer width used +// by Utils::MACThenDecrypt(). +const uint8_t PUBLIC_CHANNEL_SECRET[PUB_KEY_SIZE] = { + 0x8b, 0x33, 0x87, 0xe9, 0xc5, 0xcd, 0xea, 0x6a, + 0xc9, 0xe5, 0xed, 0xba, 0xa1, 0x15, 0xcd, 0x72, +}; + +bool leapYear(uint16_t year) { + return (year % 4U == 0 && year % 100U != 0) || year % 400U == 0; +} + +uint32_t minimumValidEpoch() { +#if FIRMWARE_BUILD_EPOCH > 0 + return (uint32_t)FIRMWARE_BUILD_EPOCH; +#else + static uint32_t minimum = 0; + if (minimum == 0) minimum = DateTime(__DATE__, __TIME__).unixtime(); + return minimum; +#endif +} + +uint32_t maximumValidEpoch() { + static uint32_t maximum = 0; + if (maximum == 0) { + DateTime built(minimumValidEpoch()); + uint16_t upper_year = built.year() + VALID_YEARS; + uint8_t upper_day = built.day(); + if (built.month() == 2 && upper_day == 29 && !leapYear(upper_year)) upper_day = 28; + maximum = DateTime(upper_year, built.month(), upper_day, + built.hour(), built.minute(), built.second()).unixtime(); + } + return maximum; +} + +bool validEpoch(uint32_t epoch) { + return epoch >= minimumValidEpoch() && epoch <= maximumValidEpoch(); +} + +File openRead(FILESYSTEM* fs, const char* filename) { +#if defined(RP2040_PLATFORM) + return fs->open(filename, "r"); +#else + return fs->open(filename); +#endif +} + +File openWrite(FILESYSTEM* fs, const char* filename) { +#if defined(NRF52_PLATFORM) || defined(STM32_PLATFORM) + fs->remove(filename); + return fs->open(filename, FILE_O_WRITE); +#elif defined(RP2040_PLATFORM) + return fs->open(filename, "w"); +#else + return fs->open(filename, "w", true); +#endif +} + +const char* skipSpaces(const char* text) { + while (text != nullptr && *text == ' ') text++; + return text; +} + +bool parseUnsigned(const char* text, uint32_t maximum, uint32_t& value) { + text = skipSpaces(text); + if (text == nullptr || *text < '0' || *text > '9') return false; + uint32_t parsed = 0; + while (*text >= '0' && *text <= '9') { + uint8_t digit = (uint8_t)(*text++ - '0'); + if (parsed > (maximum - digit) / 10U) return false; + parsed = parsed * 10U + digit; + } + text = skipSpaces(text); + if (*text != 0) return false; + value = parsed; + return true; +} + +bool parsePositiveSelector(const char* text, int& value) { + uint32_t parsed = 0; + if (!parseUnsigned(text, 255, parsed) || parsed == 0) return false; + value = (int)parsed; + return true; +} + +void derivePathId(const mesh::Packet* packet, uint8_t path_id[8]) { + uint8_t material[2 + MAX_PATH_SIZE]; + uint8_t count = packet->getPathHashCount(); + uint8_t path_bytes = packet->getPathByteLen(); + material[0] = count == 0 ? 0 : packet->getPathHashSize(); + material[1] = count; + if (path_bytes > 0) memcpy(&material[2], packet->path, path_bytes); + mesh::Utils::sha256(path_id, 8, material, 2 + path_bytes); +} + +bool decodePublicPlainText(const mesh::Packet* packet, uint32_t& timestamp, + char* sender, size_t sender_len) { + if (sender != nullptr && sender_len > 0) sender[0] = 0; + if (packet == nullptr || sender == nullptr || sender_len < 2 + || !packet->isRouteFlood() + || packet->getPayloadType() != PAYLOAD_TYPE_GRP_TXT + || packet->payload_len <= PATH_HASH_SIZE + CIPHER_MAC_SIZE) return false; + + uint8_t channel_hash = 0; + mesh::Utils::sha256(&channel_hash, sizeof(channel_hash), + PUBLIC_CHANNEL_SECRET, CIPHER_KEY_SIZE); + if (packet->payload[0] != channel_hash) return false; + + uint8_t data[MAX_PACKET_PAYLOAD]; + int len = mesh::Utils::MACThenDecrypt( + PUBLIC_CHANNEL_SECRET, data, &packet->payload[PATH_HASH_SIZE], + packet->payload_len - PATH_HASH_SIZE); + if (len <= 5 || (data[4] >> 2) != TXT_TYPE_PLAIN) return false; + + memcpy(×tamp, data, sizeof(timestamp)); + const uint8_t* text = &data[5]; + size_t text_len = (size_t)len - 5; + const uint8_t* colon = (const uint8_t*)memchr(text, ':', text_len); + if (colon == nullptr) return false; + + size_t parsed_len = (size_t)(colon - text); + while (parsed_len > 0 && text[parsed_len - 1] == ' ') parsed_len--; + if (parsed_len == 0 || parsed_len >= sender_len) return false; + for (size_t i = 0; i < parsed_len; i++) { + if (text[i] == '\r' || text[i] == '\n' || text[i] < 0x20) return false; + } + memcpy(sender, text, parsed_len); + sender[parsed_len] = 0; + return true; +} + +uint32_t rebaseTimestamp(uint32_t timestamp, uint32_t old_now, + uint32_t new_now) { + if (timestamp == 0) return 0; + uint32_t age = old_now >= timestamp ? old_now - timestamp : 0; + return new_now > age ? new_now - age : 1; +} + +const char* sampleKindName(uint8_t source_kind) { + if (source_kind == mesh::CLOCK_SYNC_SAMPLE_SOURCE_SIGNED_ADVERT) return "advert"; + if (source_kind == mesh::CLOCK_SYNC_SAMPLE_SOURCE_PUBLIC_CHANNEL) return "public"; + return "unknown"; +} + +char sampleKindCode(uint8_t source_kind) { + if (source_kind == mesh::CLOCK_SYNC_SAMPLE_SOURCE_SIGNED_ADVERT) return 'A'; + if (source_kind == mesh::CLOCK_SYNC_SAMPLE_SOURCE_PUBLIC_CHANNEL) return 'P'; + return '?'; +} + +} // namespace + +namespace mesh { + +MeshClockSync::MeshClockSync(Radio& radio, MillisecondClock& millis, + RTCClock& rtc, ClientACL& acl, + SensorManager& sensors, + const float& tx_delay_factor, + MeshClockSyncCallbacks* callbacks) + : _radio(&radio), _millis(&millis), _rtc(&rtc), _acl(&acl), + _sensors(&sensors), _tx_delay_factor(&tx_delay_factor), + _callbacks(callbacks), _fs(nullptr) { + _last_millis = 0; + _uptime_millis = 0; + resetDefaults(); +} + +void MeshClockSync::resetDefaults() { + memset(_samples, 0, sizeof(_samples)); + _mesh_enabled = CLOCK_SYNC_MESH_DEFAULT_ENABLED != 0; + _mesh_edge_enabled = CLOCK_SYNC_MESH_EDGE_DEFAULT_ENABLED != 0; + _internet_enabled = false; + _complete = false; + _force_mesh_pending = false; + _suppressed_by = SUPPRESS_NONE; + _last_result = RESULT_WAITING; + _last_source_sample_count = 0; + _last_fresh_count = 0; + _last_required_count = REQUIRED_SAMPLES_DEFAULT; + _required_samples = REQUIRED_SAMPLES_DEFAULT; + _drift_seconds = DRIFT_DEFAULT_SECONDS; + _last_estimate = 0; + _last_abs_drift = 0; + _next_attempt_uptime = STARTUP_DELAY_MILLIS; +} + +void MeshClockSync::begin(FILESYSTEM* fs) { + _fs = fs; + _last_millis = _millis->getMillis(); + loadPrefs(); +} + +void MeshClockSync::loadPrefs() { + _mesh_enabled = CLOCK_SYNC_MESH_DEFAULT_ENABLED != 0; + _mesh_edge_enabled = CLOCK_SYNC_MESH_EDGE_DEFAULT_ENABLED != 0; + _internet_enabled = false; + _drift_seconds = DRIFT_DEFAULT_SECONDS; + _required_samples = REQUIRED_SAMPLES_DEFAULT; + + if (_fs != nullptr && _fs->exists(PREFS_FILE)) { + File file = openRead(_fs, PREFS_FILE); + if (file) { + uint8_t magic[4]; + uint8_t mesh_enabled = 0; + uint8_t edge_enabled = CLOCK_SYNC_MESH_EDGE_DEFAULT_ENABLED != 0 ? 1 : 0; + uint8_t internet_enabled = 0; + uint8_t required_samples = REQUIRED_SAMPLES_DEFAULT; + uint32_t drift_seconds = 0; + bool valid = file.read(magic, sizeof(magic)) == sizeof(magic); + bool version3 = valid && memcmp(magic, "CTS3", sizeof(magic)) == 0; + bool version4 = valid && memcmp(magic, "CTS4", sizeof(magic)) == 0; + valid = valid && (version3 || version4) + && file.read(&mesh_enabled, sizeof(mesh_enabled)) == sizeof(mesh_enabled) + && file.read(&internet_enabled, sizeof(internet_enabled)) == sizeof(internet_enabled) + && file.read((uint8_t*)&drift_seconds, sizeof(drift_seconds)) == sizeof(drift_seconds) + && file.read(&required_samples, sizeof(required_samples)) == sizeof(required_samples); + if (valid && version4) { + valid = file.read(&edge_enabled, sizeof(edge_enabled)) == sizeof(edge_enabled); + } + valid = valid && mesh_enabled <= 1 && edge_enabled <= 1 + && internet_enabled <= 1 + && drift_seconds >= DRIFT_MIN_SECONDS + && drift_seconds <= DRIFT_MAX_SECONDS + && required_samples >= REQUIRED_SAMPLES_MIN + && required_samples <= REQUIRED_SAMPLES_MAX; + file.close(); + if (valid) { + _mesh_enabled = mesh_enabled != 0; + _mesh_edge_enabled = edge_enabled != 0; + _internet_enabled = internet_enabled != 0; + _drift_seconds = drift_seconds; + _required_samples = required_samples; + } + } + } + resetAttempt(); +} + +bool MeshClockSync::savePrefs() { + if (_fs == nullptr) return false; + File file = openWrite(_fs, PREFS_FILE); + if (!file) return false; + const uint8_t magic[4] = {'C', 'T', 'S', '4'}; + const uint8_t mesh_enabled = _mesh_enabled ? 1 : 0; + const uint8_t edge_enabled = _mesh_edge_enabled ? 1 : 0; + const uint8_t internet_enabled = _internet_enabled ? 1 : 0; + bool success = file.write(magic, sizeof(magic)) == sizeof(magic) + && file.write(&mesh_enabled, sizeof(mesh_enabled)) == sizeof(mesh_enabled) + && file.write(&internet_enabled, sizeof(internet_enabled)) == sizeof(internet_enabled) + && file.write((const uint8_t*)&_drift_seconds, sizeof(_drift_seconds)) == sizeof(_drift_seconds) + && file.write(&_required_samples, sizeof(_required_samples)) == sizeof(_required_samples) + && file.write(&edge_enabled, sizeof(edge_enabled)) == sizeof(edge_enabled); + file.close(); + return success; +} + +void MeshClockSync::resetAttempt() { + _complete = false; + _force_mesh_pending = false; + _last_result = RESULT_WAITING; + _last_source_sample_count = 0; + _last_fresh_count = 0; + _last_required_count = _required_samples; + _last_estimate = 0; + _last_abs_drift = 0; + _next_attempt_uptime = _uptime_millis < STARTUP_DELAY_MILLIS + ? STARTUP_DELAY_MILLIS : _uptime_millis; +} + +const char* MeshClockSync::suppressionName(uint8_t source) { + switch (source) { + case MeshClockSync::SUPPRESS_CLI: return "cli"; + case MeshClockSync::SUPPRESS_GPS: return "gps"; + case MeshClockSync::SUPPRESS_INTERNET: return "internet"; + default: return "none"; + } +} + +void MeshClockSync::suppressForBoot(uint8_t source) { + if (source == SUPPRESS_NONE || _suppressed_by != SUPPRESS_NONE) return; + _suppressed_by = source; + _force_mesh_pending = false; + memset(_samples, 0, sizeof(_samples)); + MESH_DEBUG_PRINTLN("Clock sync: LoRa estimate suppressed by %s until reboot", + suppressionName(source)); +} + +void MeshClockSync::onManualClockSet() { + suppressForBoot(SUPPRESS_CLI); +} + +void MeshClockSync::checkGpsOverride() { + LocationProvider* location = _sensors->getLocationProvider(); + if (location != nullptr && location->consumeTimeSyncApplied()) { + suppressForBoot(SUPPRESS_GPS); + } +} + +bool MeshClockSync::collectionActive() const { + if (!_mesh_enabled || _suppressed_by != SUPPRESS_NONE) return false; + if (!_complete) return true; + if (_next_attempt_uptime == 0) return false; + if (_uptime_millis >= _next_attempt_uptime) return true; + return _next_attempt_uptime - _uptime_millis + <= (uint64_t)SAMPLE_MAX_AGE_MILLIS; +} + +uint32_t MeshClockSync::estimateTransitMillis(const Packet* packet) const { + if (packet == nullptr || _radio == nullptr) return 0; + uint8_t hops = packet->getPathHashCount(); + uint8_t hash_size = packet->getPathHashSize(); + int base_length = packet->getRawLength() - packet->getPathByteLen(); + if (base_length < 2) return 0; + + uint64_t total = _radio->getEstAirtimeFor(base_length); + const uint64_t maximum = (uint64_t)CONSENSUS_WINDOW_SECONDS * 1000ULL; + for (uint8_t relay = 1; relay <= hops; relay++) { + uint32_t airtime = _radio->getEstAirtimeFor(base_length + relay * hash_size); + if (*_tx_delay_factor > 0.0f) { + float expected_delay = (float)airtime * *_tx_delay_factor * 2.5f; + if (expected_delay > 0.0f) total += (uint32_t)(expected_delay + 0.5f); + } + total += airtime; + if (total >= maximum) return (uint32_t)maximum; + } + return (uint32_t)total; +} + +void MeshClockSync::recordSample(uint8_t source_kind, + const uint8_t source_id[4], uint32_t epoch, + const Packet* packet) { + if (!collectionActive() || source_id == nullptr || packet == nullptr + || !packet->isRouteFlood() || !validEpoch(epoch)) return; + + uint32_t transit_seconds = (estimateTransitMillis(packet) + 500UL) / 1000UL; + uint32_t maximum = maximumValidEpoch(); + if (epoch > maximum - transit_seconds) return; + epoch += transit_seconds; + + uint8_t path_id[PATH_ID_SIZE]; + derivePathId(packet, path_id); + uint32_t now = _millis->getMillis(); + uint32_t received_millis = _radio->getLastRecvMillis(); + if (received_millis == 0 || now - received_millis > 60000UL) received_millis = now; + + int source_slot = -1; + for (int i = 0; i < SAMPLE_SLOTS; i++) { + const Sample& sample = _samples[i]; + if (sample.active && sample.source_kind == source_kind + && memcmp(sample.source_id, source_id, sizeof(sample.source_id)) == 0) { + source_slot = i; + break; + } + } + if (source_slot >= 0) { + const Sample& prior = _samples[source_slot]; + if (prior.epoch == epoch + && memcmp(prior.path_id, path_id, sizeof(prior.path_id)) == 0) return; + } + + if (clockSyncRequiresUniquePath(_mesh_edge_enabled)) { + for (int i = 0; i < SAMPLE_SLOTS; i++) { + const Sample& sample = _samples[i]; + if (i == source_slot || !sample.active + || now - sample.received_millis > SAMPLE_MAX_AGE_MILLIS) continue; + if (memcmp(sample.path_id, path_id, sizeof(sample.path_id)) == 0) return; + } + } + + int slot = source_slot; + int reusable = -1; + int oldest = 0; + uint32_t oldest_age = 0; + for (int i = 0; slot < 0 && i < SAMPLE_SLOTS; i++) { + const Sample& sample = _samples[i]; + uint32_t age = sample.active ? now - sample.received_millis : 0; + if ((!sample.active || age > SAMPLE_MAX_AGE_MILLIS) && reusable < 0) reusable = i; + if (sample.active && age >= oldest_age) { + oldest_age = age; + oldest = i; + } + } + if (slot < 0) slot = reusable >= 0 ? reusable : oldest; + + Sample& sample = _samples[slot]; + sample.active = true; + sample.source_kind = source_kind; + memcpy(sample.source_id, source_id, sizeof(sample.source_id)); + memcpy(sample.path_id, path_id, sizeof(sample.path_id)); + sample.epoch = epoch; + sample.received_millis = received_millis; + + if (!_complete) { + uint32_t estimate_epoch = 0; + uint8_t fresh = 0; + uint8_t agreeing = 0; + uint8_t required = _required_samples; + estimate(estimate_epoch, fresh, agreeing, required); + if (fresh >= _required_samples) { + _last_fresh_count = fresh; + _last_source_sample_count = agreeing; + _last_required_count = required; + _next_attempt_uptime = _uptime_millis; + } + } +} + +void MeshClockSync::recordPublicChannel(const Packet* packet) { + if (!collectionActive()) return; + uint32_t timestamp = 0; + char sender[32]; + if (!decodePublicPlainText(packet, timestamp, sender, sizeof(sender))) return; + for (char* p = sender; *p; p++) { + if (*p >= 'A' && *p <= 'Z') *p = (char)(*p - 'A' + 'a'); + } + uint8_t source_id[4]; + Utils::sha256(source_id, sizeof(source_id), + (const uint8_t*)sender, strlen(sender)); + recordSample(CLOCK_SYNC_SAMPLE_SOURCE_PUBLIC_CHANNEL, + source_id, timestamp, packet); +} + +void MeshClockSync::observeVerifiedAdvert(const Packet* packet, + const Identity& id, + uint32_t timestamp) { + if (!_mesh_edge_enabled || !collectionActive()) return; + uint8_t source_id[4]; + Utils::sha256(source_id, sizeof(source_id), id.pub_key, PUB_KEY_SIZE); + recordSample(CLOCK_SYNC_SAMPLE_SOURCE_SIGNED_ADVERT, + source_id, timestamp, packet); +} + +void MeshClockSync::observeGroupPacket(const Packet* packet) { + if (_mesh_edge_enabled) recordPublicChannel(packet); +} + +void MeshClockSync::observeAcceptedFlood(const Packet* packet) { + if (_mesh_edge_enabled || !collectionActive() || packet == nullptr) return; + if (packet->getPayloadType() == PAYLOAD_TYPE_GRP_TXT) { + recordPublicChannel(packet); + } else if (packet->getPayloadType() == PAYLOAD_TYPE_ADVERT) { + const size_t minimum = PUB_KEY_SIZE + sizeof(uint32_t) + SIGNATURE_SIZE; + if (packet->payload_len < minimum) return; + uint8_t source_id[4]; + Utils::sha256(source_id, sizeof(source_id), packet->payload, PUB_KEY_SIZE); + uint32_t timestamp = 0; + memcpy(×tamp, &packet->payload[PUB_KEY_SIZE], sizeof(timestamp)); + recordSample(CLOCK_SYNC_SAMPLE_SOURCE_SIGNED_ADVERT, + source_id, timestamp, packet); + } +} + +bool MeshClockSync::estimate(uint32_t& epoch, uint8_t& fresh_count, + uint8_t& agreeing_count, + uint8_t& required_count) const { + uint32_t values[SAMPLE_SLOTS]; + uint8_t count = 0; + uint32_t now = _millis->getMillis(); + uint32_t maximum = maximumValidEpoch(); + for (int i = 0; i < SAMPLE_SLOTS; i++) { + const Sample& sample = _samples[i]; + if (!sample.active) continue; + uint32_t age_millis = now - sample.received_millis; + if (age_millis > SAMPLE_MAX_AGE_MILLIS) continue; + uint32_t age_seconds = age_millis / 1000UL; + if (sample.epoch > maximum - age_seconds) continue; + values[count++] = sample.epoch + age_seconds; + } + ClockSyncConsensusResult result = evaluateClockSyncConsensus( + values, count, _required_samples, CONSENSUS_WINDOW_SECONDS); + fresh_count = result.fresh_count; + agreeing_count = result.agreeing_count; + required_count = result.required_count; + epoch = result.estimate; + return result.consensus; +} + +bool MeshClockSync::applyEstimate(uint32_t epoch, uint8_t sample_count) { + if (!validEpoch(epoch)) return false; + if (_callbacks != nullptr && _callbacks->hasAuthoritativeClock()) { + suppressForBoot(SUPPRESS_INTERNET); + return true; + } + + uint32_t old_now = _rtc->getCurrentTime(); + int64_t delta = (int64_t)epoch - (int64_t)old_now; + uint64_t magnitude = delta < 0 ? (uint64_t)(-delta) : (uint64_t)delta; + _last_source_sample_count = sample_count; + _last_estimate = epoch; + _last_abs_drift = magnitude > UINT32_MAX ? UINT32_MAX : (uint32_t)magnitude; + _complete = true; + _next_attempt_uptime = _uptime_millis + RESYNC_INTERVAL_MILLIS; + + if (magnitude <= _drift_seconds) { + _last_result = RESULT_WITHIN_DRIFT; + MESH_DEBUG_PRINTLN("Clock sync: within drift (%lu seconds, source=mesh)", + (unsigned long)_last_abs_drift); + return true; + } + + _rtc->setCurrentTime(epoch); + _rtc->resetUniqueTime(epoch); + _last_result = delta > 0 ? RESULT_CORRECTED_FORWARD : RESULT_CORRECTED_BACKWARD; + for (int i = 0; i < _acl->getNumClients(); i++) { + ClientInfo* client = _acl->getClientByIdx(i); + client->last_activity = rebaseTimestamp(client->last_activity, old_now, epoch); + } + if (_callbacks != nullptr) _callbacks->onMeshClockAdjusted(old_now, epoch); + + MESH_DEBUG_PRINTLN("Clock sync: corrected %s by %lu seconds (source=mesh samples=%u)", + delta > 0 ? "forward" : "backward", + (unsigned long)_last_abs_drift, + (unsigned int)sample_count); + return true; +} + +void MeshClockSync::checkClock() { + if (_callbacks != nullptr && _callbacks->hasAuthoritativeClock()) { + suppressForBoot(SUPPRESS_INTERNET); + } + bool mesh_available = _mesh_enabled && _suppressed_by == SUPPRESS_NONE; + bool force_mesh = _force_mesh_pending && mesh_available; + if ((!mesh_available && !_internet_enabled) + || (!force_mesh && _uptime_millis < _next_attempt_uptime)) return; + if (force_mesh) _force_mesh_pending = false; + if (_complete) _complete = false; + + if (_internet_enabled && !force_mesh) { + _last_result = RESULT_INTERNET_UNAVAILABLE; + } + if (mesh_available) { + uint32_t estimate_epoch = 0; + uint8_t fresh = 0; + uint8_t agreeing = 0; + uint8_t required = _required_samples; + bool consensus = estimate(estimate_epoch, fresh, agreeing, required); + _last_fresh_count = fresh; + _last_source_sample_count = agreeing; + _last_required_count = required; + if (consensus && applyEstimate(estimate_epoch, agreeing)) return; + _last_result = fresh < _required_samples + ? RESULT_COLLECTING : RESULT_NO_CONSENSUS; + } + _next_attempt_uptime = _uptime_millis + RETRY_INTERVAL_MILLIS; +} + +void MeshClockSync::loop() { + uint32_t now = _millis->getMillis(); + _uptime_millis += now - _last_millis; + _last_millis = now; + checkGpsOverride(); + checkClock(); +} + +void MeshClockSync::formatSample(int index, char* reply, size_t reply_len) const { + if (index < 0 || index >= SAMPLE_SLOTS) { + snprintf(reply, reply_len, "Err - clock sample slot must be 1-%u", + (unsigned int)SAMPLE_SLOTS); + return; + } + const Sample& sample = _samples[index]; + if (!sample.active) { + snprintf(reply, reply_len, "> %d empty", index + 1); + return; + } + + uint32_t age_seconds = (_millis->getMillis() - sample.received_millis) / 1000UL; + bool fresh = age_seconds <= SAMPLE_MAX_AGE_MILLIS / 1000UL; + uint32_t current_epoch = sample.epoch; + if (age_seconds <= UINT32_MAX - current_epoch) current_epoch += age_seconds; + uint32_t local_epoch = _rtc->getCurrentTime(); + char delta_sign = current_epoch >= local_epoch ? '+' : '-'; + uint32_t delta = current_epoch >= local_epoch + ? current_epoch - local_epoch : local_epoch - current_epoch; + char source_id[sizeof(sample.source_id) * 2 + 1]; + char path_id[sizeof(sample.path_id) * 2 + 1]; + Utils::toHex(source_id, sample.source_id, sizeof(sample.source_id)); + Utils::toHex(path_id, sample.path_id, sizeof(sample.path_id)); + snprintf(reply, reply_len, + "> %d %s id=%s path=%s age=%lus epoch=%lu delta=%c%lus fresh=%s", + index + 1, sampleKindName(sample.source_kind), source_id, path_id, + (unsigned long)age_seconds, (unsigned long)current_epoch, + delta_sign, (unsigned long)delta, fresh ? "yes" : "no"); +} + +void MeshClockSync::formatTable(char* reply, size_t reply_len) const { + uint32_t now = _millis->getMillis(); + uint8_t fresh = 0; + uint8_t active = 0; + for (int i = 0; i < SAMPLE_SLOTS; i++) { + if (!_samples[i].active) continue; + active++; + if (now - _samples[i].received_millis <= SAMPLE_MAX_AGE_MILLIS) fresh++; + } + + const char* mode = _mesh_edge_enabled ? "edge" : "paths"; + size_t used = (size_t)snprintf( + reply, reply_len, "> %s collect=%s fresh=%u/%u", mode, + collectionActive() ? "active" : "inactive", (unsigned int)fresh, + (unsigned int)_required_samples); + if (active == 0 || used >= reply_len) { + if (active == 0 && used + 5 < reply_len) { + StrHelper::strncpy(&reply[used], " none", reply_len - used); + } + return; + } + + for (int i = 0; i < SAMPLE_SLOTS && used + 1 < reply_len; i++) { + const Sample& sample = _samples[i]; + if (!sample.active) continue; + uint32_t age_seconds = (now - sample.received_millis) / 1000UL; + unsigned long age_value = age_seconds < 120UL + ? (unsigned long)age_seconds : (unsigned long)(age_seconds / 60UL); + char age_unit = age_seconds < 120UL ? 's' : 'm'; + bool sample_fresh = age_seconds <= SAMPLE_MAX_AGE_MILLIS / 1000UL; + char item[24]; + snprintf(item, sizeof(item), " %d:%c:%02X%02X:%lu%c%s", i + 1, + sampleKindCode(sample.source_kind), sample.source_id[0], + sample.source_id[1], age_value, age_unit, + sample_fresh ? "" : "!"); + size_t item_len = strlen(item); + if (used + item_len >= reply_len - 4) { + StrHelper::strncpy(&reply[used], " ...", reply_len - used); + return; + } + memcpy(&reply[used], item, item_len + 1); + used += item_len; + } +} + +void MeshClockSync::formatStatus(const char* args, char* reply, + size_t reply_len) const { + const char* selector = skipSpaces(args); + if (selector != nullptr && *selector == '.') selector = skipSpaces(selector + 1); + if (selector != nullptr && *selector != 0) { + if (strcmp(selector, "table") == 0) { + formatTable(reply, reply_len); + return; + } + int slot = 0; + if (parsePositiveSelector(selector, slot) && slot <= SAMPLE_SLOTS) { + formatSample(slot - 1, reply, reply_len); + return; + } + snprintf(reply, reply_len, + "Err - use get clock.sync.status[.table|.1-.%u]", + (unsigned int)SAMPLE_SLOTS); + return; + } + + uint8_t active = 0; + for (int i = 0; i < SAMPLE_SLOTS; i++) { + if (_samples[i].active) active++; + } + uint32_t live_estimate = 0; + uint8_t fresh = 0; + uint8_t agreeing = 0; + uint8_t required = _required_samples; + bool live_consensus = estimate(live_estimate, fresh, agreeing, required); + bool mesh_available = _mesh_enabled && _suppressed_by == SUPPRESS_NONE; + const char* mesh_state = !_mesh_enabled ? "off" + : (mesh_available ? "on" + : (_suppressed_by == SUPPRESS_CLI ? "suppressed-cli" + : (_suppressed_by == SUPPRESS_GPS ? "suppressed-gps" + : "suppressed-internet"))); + const char* mode = _mesh_edge_enabled ? "edge" : "paths"; + const char* evidence = _mesh_edge_enabled ? "sources" : "paths"; + uint64_t remaining_ms = _next_attempt_uptime > _uptime_millis + ? _next_attempt_uptime - _uptime_millis : 0; + unsigned long next_seconds = (unsigned long)((remaining_ms + 999ULL) / 1000ULL); + if (!mesh_available && !_internet_enabled) { + const char* reason = _mesh_enabled ? mesh_state : "mesh-off"; + snprintf(reply, reply_len, + "> not-set reason=%s collect=inactive mode=%s %s=%u/%u table=%u", + reason, mode, evidence, (unsigned int)fresh, + (unsigned int)_required_samples, (unsigned int)active); + return; + } + + const char* result = "waiting"; + switch (_last_result) { + case RESULT_COLLECTING: result = "collecting"; break; + case RESULT_INTERNET_UNAVAILABLE: result = "internet-unavailable"; break; + case RESULT_NO_CONSENSUS: result = "no-consensus"; break; + case RESULT_WITHIN_DRIFT: result = "within-drift"; break; + case RESULT_CORRECTED_FORWARD: result = "corrected-forward"; break; + case RESULT_CORRECTED_BACKWARD: result = "corrected-backward"; break; + default: break; + } + if (_complete) { + bool clock_was_set = _last_result == RESULT_CORRECTED_FORWARD + || _last_result == RESULT_CORRECTED_BACKWARD; + snprintf(reply, reply_len, + "> %s reason=%s difference=%lus threshold=%lus via=mesh collect=%s mode=%s table=%u next=%lus", + clock_was_set ? "set" : "not-set", result, + (unsigned long)_last_abs_drift, (unsigned long)_drift_seconds, + collectionActive() ? "active" : "inactive", mode, + (unsigned int)active, next_seconds); + } else if (_last_result == RESULT_NO_CONSENSUS + || (fresh >= _required_samples && !live_consensus)) { + snprintf(reply, reply_len, + "> not-set reason=no-consensus collect=%s mode=%s %s=%u agree=%u/%u table=%u next=%lus", + collectionActive() ? "active" : "inactive", mode, evidence, + (unsigned int)fresh, (unsigned int)agreeing, + (unsigned int)required, (unsigned int)active, next_seconds); + } else if (live_consensus) { + snprintf(reply, reply_len, + "> not-set reason=ready collect=%s mode=%s %s=%u agree=%u/%u table=%u next=%lus", + collectionActive() ? "active" : "inactive", mode, evidence, + (unsigned int)fresh, (unsigned int)agreeing, + (unsigned int)required, (unsigned int)active, next_seconds); + } else { + const char* reason = result; + if (_last_result == RESULT_WAITING) reason = "waiting-deadline"; + else if (_last_result == RESULT_COLLECTING) { + reason = _mesh_edge_enabled ? "need-more-sources" : "need-more-paths"; + } + snprintf(reply, reply_len, + "> not-set reason=%s collect=%s mesh=%s mode=%s %s=%u/%u table=%u next=%lus", + reason, collectionActive() ? "active" : "inactive", mesh_state, + mode, evidence, (unsigned int)fresh, + (unsigned int)_required_samples, (unsigned int)active, + next_seconds); + } +} + +bool MeshClockSync::handleCommand(const char* command, char* reply, + size_t reply_len) { + if (command == nullptr || reply == nullptr || reply_len == 0) return false; + if (strncmp(command, "get clock.sync.status", 21) == 0 + && (command[21] == 0 || command[21] == '.')) { + formatStatus(command + 21, reply, reply_len); + } else if (strcmp(command, "get clock.sync") == 0) { + formatStatus("", reply, reply_len); + } else if (strcmp(command, "get clock.sync.mesh") == 0) { + if (_mesh_enabled && _suppressed_by != SUPPRESS_NONE) { + snprintf(reply, reply_len, "> on (suppressed by %s until reboot)", + suppressionName(_suppressed_by)); + } else { + snprintf(reply, reply_len, "> %s", _mesh_enabled ? "on" : "off"); + } + } else if (strcmp(command, "get clock.sync.mesh.edge") == 0) { + snprintf(reply, reply_len, "> %s", _mesh_edge_enabled ? "on" : "off"); + } else if (strcmp(command, "get clock.sync.internet") == 0) { + snprintf(reply, reply_len, "> %s (unavailable on this build)", + _internet_enabled ? "on" : "off"); + } else if (strcmp(command, "get clock.sync.drift") == 0) { + snprintf(reply, reply_len, "> %lu", (unsigned long)_drift_seconds); + } else if (strcmp(command, "get clock.sync.samples") == 0) { + snprintf(reply, reply_len, "> %u", (unsigned int)_required_samples); + } else if (strcmp(command, "clock.sync.mesh now") == 0) { + if (!_mesh_enabled) { + StrHelper::strncpy(reply, "Err - mesh clock sync is off", reply_len); + } else if (_suppressed_by != SUPPRESS_NONE) { + snprintf(reply, reply_len, "Err - mesh sync suppressed by %s until reboot", + suppressionName(_suppressed_by)); + } else { + resetAttempt(); + _force_mesh_pending = true; + _next_attempt_uptime = _uptime_millis; + StrHelper::strncpy(reply, "OK - mesh clock sync queued", reply_len); + } + } else if (strncmp(command, "set clock.sync.mesh ", 20) == 0) { + const char* value = command + 20; + bool enabled; + if (strcmp(value, "on") == 0) enabled = true; + else if (strcmp(value, "off") == 0) enabled = false; + else { + StrHelper::strncpy(reply, "Err - usage: set clock.sync.mesh ", reply_len); + return true; + } + bool previous = _mesh_enabled; + _mesh_enabled = enabled; + if (!savePrefs()) { + _mesh_enabled = previous; + StrHelper::strncpy(reply, "Err - unable to save clock sync settings", reply_len); + } else { + if (enabled && !previous) memset(_samples, 0, sizeof(_samples)); + resetAttempt(); + if (enabled && _suppressed_by != SUPPRESS_NONE) { + snprintf(reply, reply_len, "OK - enabled; suppressed by %s until reboot", + suppressionName(_suppressed_by)); + } else { + StrHelper::strncpy(reply, enabled ? "OK - mesh clock sync enabled" + : "OK - mesh clock sync disabled", + reply_len); + } + } + } else if (strncmp(command, "set clock.sync.mesh.edge ", 25) == 0) { + const char* value = command + 25; + bool enabled; + if (strcmp(value, "on") == 0) enabled = true; + else if (strcmp(value, "off") == 0) enabled = false; + else { + StrHelper::strncpy(reply, "Err - usage: set clock.sync.mesh.edge ", reply_len); + return true; + } + bool previous = _mesh_edge_enabled; + _mesh_edge_enabled = enabled; + if (!savePrefs()) { + _mesh_edge_enabled = previous; + StrHelper::strncpy(reply, "Err - unable to save clock sync settings", reply_len); + } else { + if (enabled != previous) memset(_samples, 0, sizeof(_samples)); + resetAttempt(); + StrHelper::strncpy(reply, enabled ? "OK - edge clock sync enabled" + : "OK - edge clock sync disabled", + reply_len); + } + } else if (strncmp(command, "set clock.sync.internet ", 24) == 0) { + const char* value = command + 24; + bool enabled; + if (strcmp(value, "on") == 0) enabled = true; + else if (strcmp(value, "off") == 0) enabled = false; + else { + StrHelper::strncpy(reply, "Err - usage: set clock.sync.internet ", reply_len); + return true; + } + bool previous = _internet_enabled; + _internet_enabled = enabled; + if (!savePrefs()) { + _internet_enabled = previous; + StrHelper::strncpy(reply, "Err - unable to save clock sync settings", reply_len); + } else { + resetAttempt(); + StrHelper::strncpy(reply, + enabled ? "OK - enabled (internet unavailable on this build)" + : "OK - internet clock sync disabled", + reply_len); + } + } else if (strncmp(command, "set clock.sync.drift ", 21) == 0) { + uint32_t drift = 0; + if (!parseUnsigned(command + 21, DRIFT_MAX_SECONDS, drift) + || drift < DRIFT_MIN_SECONDS) { + StrHelper::strncpy(reply, "Err - drift must be 30-86400 seconds", reply_len); + } else { + uint32_t previous = _drift_seconds; + _drift_seconds = drift; + if (!savePrefs()) { + _drift_seconds = previous; + StrHelper::strncpy(reply, "Err - unable to save clock sync settings", reply_len); + } else { + resetAttempt(); + snprintf(reply, reply_len, "OK - clock drift threshold %lu seconds", + (unsigned long)drift); + } + } + } else if (strncmp(command, "set clock.sync.samples ", 23) == 0) { + uint32_t required = 0; + if (!parseUnsigned(command + 23, REQUIRED_SAMPLES_MAX, required) + || required < REQUIRED_SAMPLES_MIN) { + StrHelper::strncpy(reply, "Err - samples must be 3-16", reply_len); + } else { + uint8_t previous = _required_samples; + _required_samples = (uint8_t)required; + if (!savePrefs()) { + _required_samples = previous; + StrHelper::strncpy(reply, "Err - unable to save clock sync settings", reply_len); + } else { + resetAttempt(); + snprintf(reply, reply_len, "OK - clock sync requires %u samples", + (unsigned int)required); + } + } + } else { + return false; + } + return true; +} + +} // namespace mesh diff --git a/src/helpers/MeshClockSync.h b/src/helpers/MeshClockSync.h new file mode 100644 index 00000000..26509307 --- /dev/null +++ b/src/helpers/MeshClockSync.h @@ -0,0 +1,130 @@ +#pragma once + +#include +#include +#include +#include + +namespace mesh { + +class MeshClockSyncCallbacks { +public: + virtual ~MeshClockSyncCallbacks() = default; + + // WiFi/NTP, or another authoritative source, owns the clock for this boot. + virtual bool hasAuthoritativeClock() const { return false; } + + // Let a role repair any timestamp-based caches after the RTC moves. + virtual void onMeshClockAdjusted(uint32_t old_epoch, uint32_t new_epoch) { + (void)old_epoch; + (void)new_epoch; + } +}; + +// Autonomous LoRa clock consensus shared by sensor and room-server roles. +// Repeater firmware retains its extended implementation, which additionally +// supports delayed internet/NTP estimates and repeater-specific state repair. +class MeshClockSync { +public: + MeshClockSync(Radio& radio, MillisecondClock& millis, RTCClock& rtc, + ClientACL& acl, SensorManager& sensors, + const float& tx_delay_factor, + MeshClockSyncCallbacks* callbacks = nullptr); + + void begin(FILESYSTEM* fs); + void loop(); + + void onManualClockSet(); + + // Edge mode observes valid packets even when this node does not forward + // them. Normal/path mode observes only packets accepted for forwarding. + void observeVerifiedAdvert(const Packet* packet, const Identity& id, + uint32_t timestamp); + void observeGroupPacket(const Packet* packet); + void observeAcceptedFlood(const Packet* packet); + + bool handleCommand(const char* command, char* reply, size_t reply_len = 160); + bool edgeMode() const { return _mesh_edge_enabled; } + bool collectionActive() const; + +private: + static constexpr uint8_t SAMPLE_SLOTS = 16; + static constexpr uint8_t PATH_ID_SIZE = 8; + + struct Sample { + bool active; + uint8_t source_kind; + uint8_t source_id[4]; + uint8_t path_id[PATH_ID_SIZE]; + uint32_t epoch; + uint32_t received_millis; + }; + + enum Suppression : uint8_t { + SUPPRESS_NONE = 0, + SUPPRESS_CLI = 1, + SUPPRESS_GPS = 2, + SUPPRESS_INTERNET = 3, + }; + + enum Result : uint8_t { + RESULT_WAITING = 0, + RESULT_COLLECTING = 1, + RESULT_INTERNET_UNAVAILABLE = 2, + RESULT_NO_CONSENSUS = 3, + RESULT_WITHIN_DRIFT = 4, + RESULT_CORRECTED_FORWARD = 5, + RESULT_CORRECTED_BACKWARD = 6, + }; + + Radio* _radio; + MillisecondClock* _millis; + RTCClock* _rtc; + ClientACL* _acl; + SensorManager* _sensors; + const float* _tx_delay_factor; + MeshClockSyncCallbacks* _callbacks; + FILESYSTEM* _fs; + + Sample _samples[SAMPLE_SLOTS]; + bool _mesh_enabled; + bool _mesh_edge_enabled; + bool _internet_enabled; + bool _complete; + bool _force_mesh_pending; + uint8_t _suppressed_by; + uint8_t _last_result; + uint8_t _last_source_sample_count; + uint8_t _last_fresh_count; + uint8_t _last_required_count; + uint8_t _required_samples; + uint32_t _drift_seconds; + uint32_t _last_estimate; + uint32_t _last_abs_drift; + uint32_t _last_millis; + uint64_t _uptime_millis; + uint64_t _next_attempt_uptime; + + void resetDefaults(); + void loadPrefs(); + bool savePrefs(); + void resetAttempt(); + void suppressForBoot(uint8_t source); + static const char* suppressionName(uint8_t source); + void checkGpsOverride(); + void checkClock(); + + uint32_t estimateTransitMillis(const Packet* packet) const; + void recordSample(uint8_t source_kind, const uint8_t source_id[4], + uint32_t epoch, const Packet* packet); + void recordPublicChannel(const Packet* packet); + bool estimate(uint32_t& epoch, uint8_t& fresh_count, + uint8_t& agreeing_count, uint8_t& required_count) const; + bool applyEstimate(uint32_t epoch, uint8_t sample_count); + + void formatStatus(const char* args, char* reply, size_t reply_len) const; + void formatTable(char* reply, size_t reply_len) const; + void formatSample(int index, char* reply, size_t reply_len) const; +}; + +} // namespace mesh