diff --git a/lib/ble_interface/BLEInterface.cpp b/lib/ble_interface/BLEInterface.cpp index c7589bd1..eefc295a 100644 --- a/lib/ble_interface/BLEInterface.cpp +++ b/lib/ble_interface/BLEInterface.cpp @@ -140,6 +140,15 @@ void BLEInterface::stop() { } _pending_handshake_count = 0; _pending_data_count = 0; + { + std::lock_guard lock(_mutex); + for (size_t index = 0; index < MAX_PENDING_PACKETS; ++index) { + _pending_packet_pool[index].clear(); + } + _pending_packet_read = 0; + _pending_packet_write = 0; + _pending_packet_count = 0; + } _online = false; INFO("BLEInterface: Stopped"); @@ -889,9 +898,37 @@ void BLEInterface::onMacRotation(const Bytes& old_mac, const Bytes& new_mac, con //============================================================================= void BLEInterface::onPacketReassembled(const Bytes& peer_identity, const Bytes& packet) { - // Packet reassembly complete - pass to transport + // Packet reassembly complete. Queue for the single Transport owner task; + // never enter Transport from the dedicated BLE task. _peer_manager.recordPacketReceived(peer_identity); - handle_incoming(packet); + std::lock_guard lock(_mutex); + if (_pending_packet_count >= MAX_PENDING_PACKETS) { + WARNING("BLEInterface: inbound packet queue full; dropping new packet"); + return; + } + _pending_packet_pool[_pending_packet_write] = packet; + _pending_packet_write = + (_pending_packet_write + 1) % MAX_PENDING_PACKETS; + ++_pending_packet_count; +} + +size_t BLEInterface::drain_inbound(size_t maximum_packets) { + size_t drained = 0; + while (drained < maximum_packets) { + Bytes packet; + { + std::lock_guard lock(_mutex); + if (_pending_packet_count == 0) break; + packet = _pending_packet_pool[_pending_packet_read]; + _pending_packet_pool[_pending_packet_read].clear(); + _pending_packet_read = + (_pending_packet_read + 1) % MAX_PENDING_PACKETS; + --_pending_packet_count; + } + handle_incoming(packet); + ++drained; + } + return drained; } void BLEInterface::onReassemblyTimeout(const Bytes& peer_identity, const std::string& reason) { diff --git a/lib/ble_interface/BLEInterface.h b/lib/ble_interface/BLEInterface.h index 421251b3..7c21c5c4 100644 --- a/lib/ble_interface/BLEInterface.h +++ b/lib/ble_interface/BLEInterface.h @@ -165,6 +165,10 @@ public: */ bool is_task_running() const { return _task_handle != nullptr; } + // Drain fully reassembled packets on the Reticulum/router owner task. + // BLE callbacks only enqueue, preventing concurrent Transport mutation. + size_t drain_inbound(size_t maximum_packets = 4); + protected: virtual bool send_outgoing(const RNS::Bytes& data) override; @@ -281,6 +285,12 @@ private: PendingData _pending_data_pool[MAX_PENDING_DATA]; size_t _pending_data_count = 0; + static constexpr size_t MAX_PENDING_PACKETS = 8; + RNS::Bytes _pending_packet_pool[MAX_PENDING_PACKETS]; + size_t _pending_packet_read = 0; + size_t _pending_packet_write = 0; + size_t _pending_packet_count = 0; + // Diagnostic counters — included in the periodic BLE heartbeat // log so we can see "did fragments actually flow over a peer // connection" without per-event logs flooding USB CDC. Cumulative diff --git a/lib/tdeck_ui/Telemetry/LocationPersistence.cpp b/lib/tdeck_ui/Telemetry/LocationPersistence.cpp index 8a4c67db..df991e6b 100644 --- a/lib/tdeck_ui/Telemetry/LocationPersistence.cpp +++ b/lib/tdeck_ui/Telemetry/LocationPersistence.cpp @@ -83,7 +83,10 @@ LocationPersistenceResult TransactionalLocationPersistence::save( const CandidateResult promoted = readCandidate(LocationPersistenceSlot::LIVE, candidate_size); if (promoted == CandidateResult::IO_ERROR) { - return LocationPersistenceResult::IO_ERROR; + // TEMP was fully read-back validated before the atomic promotion. + // Once rename succeeds, the new generation is committed even if an + // immediate second observation is unavailable. + return LocationPersistenceResult::SAVED; } if (promoted != CandidateResult::VALID) { return LocationPersistenceResult::INVALID_STATE; diff --git a/lib/tdeck_ui/Telemetry/LocationPersistenceController.cpp b/lib/tdeck_ui/Telemetry/LocationPersistenceController.cpp index 6cbcc4ba..e676d450 100644 --- a/lib/tdeck_ui/Telemetry/LocationPersistenceController.cpp +++ b/lib/tdeck_ui/Telemetry/LocationPersistenceController.cpp @@ -96,9 +96,14 @@ LocationControllerState LocationPersistenceController::service( !observeMonotonic(monotonic_now_millis)) { return LocationControllerState::BLOCKED; } + if (wall_now_millis < TRUSTED_WALL_CLOCK_MIN_MILLIS) { + if (state_ == LocationControllerState::READY) { + state_ = LocationControllerState::BLOCKED; + } + return state_; + } if (state_ == LocationControllerState::WAITING_FOR_CLOCK) { - if (wall_now_millis < TRUSTED_WALL_CLOCK_MIN_MILLIS) return state_; if (monotonic_now_millis < next_restore_attempt_monotonic_millis_) { return state_; } @@ -192,25 +197,12 @@ LocationConsentResult LocationPersistenceController::mapSessionResult( return LocationConsentResult::INVALID_ARGUMENT; } -void LocationPersistenceController::captureRollback(const PeerId& peer) { - const std::size_t index = scheduler_.find(peer); - rollback_existed_ = index != MAX_SHARE_SESSIONS; - rollback_session_ = rollback_existed_ - ? scheduler_.slots_[index].session - : ShareSession{}; - rollback_revision_ = scheduler_.revision_; +void LocationPersistenceController::captureRollback() { + rollback_scheduler_ = scheduler_; } -void LocationPersistenceController::restoreRollback(const PeerId& peer) { - const std::size_t index = scheduler_.find(peer); - if (rollback_existed_) { - if (index != MAX_SHARE_SESSIONS) { - scheduler_.slots_[index].session = rollback_session_; - } - } else if (index != MAX_SHARE_SESSIONS) { - scheduler_.clear(index); - } - scheduler_.revision_ = rollback_revision_; +void LocationPersistenceController::restoreRollback() { + scheduler_ = rollback_scheduler_; } LocationConsentResult LocationPersistenceController::startSharing( @@ -222,14 +214,14 @@ LocationConsentResult LocationPersistenceController::startSharing( !observeMonotonic(monotonic_now_millis)) { return LocationConsentResult::NOT_READY; } - captureRollback(peer); + captureRollback(); const ShareSessionResult result = scheduler_.start(peer, options, wall_now_millis); if (result != ShareSessionResult::STARTED && result != ShareSessionResult::UPDATED) { return mapSessionResult(result); } if (urgentSave(monotonic_now_millis) != LocationControllerSaveResult::SAVED) { - restoreRollback(peer); + restoreRollback(); return LocationConsentResult::STORAGE_FAILURE; } return mapSessionResult(result); @@ -243,11 +235,11 @@ LocationConsentResult LocationPersistenceController::stopSharing( !observeMonotonic(monotonic_now_millis)) { return LocationConsentResult::NOT_READY; } - captureRollback(peer); + captureRollback(); const ShareSessionResult result = scheduler_.stop(peer, wall_now_millis); if (result != ShareSessionResult::STOPPING) return mapSessionResult(result); if (urgentSave(monotonic_now_millis) != LocationControllerSaveResult::SAVED) { - restoreRollback(peer); + restoreRollback(); return LocationConsentResult::STORAGE_FAILURE; } return LocationConsentResult::STOPPING; diff --git a/lib/tdeck_ui/Telemetry/LocationPersistenceController.h b/lib/tdeck_ui/Telemetry/LocationPersistenceController.h index 5a9c72d9..51a3721f 100644 --- a/lib/tdeck_ui/Telemetry/LocationPersistenceController.h +++ b/lib/tdeck_ui/Telemetry/LocationPersistenceController.h @@ -66,8 +66,8 @@ private: void observeDirty(uint64_t monotonic_now_millis); void markSaved(); static LocationConsentResult mapSessionResult(ShareSessionResult result); - void captureRollback(const PeerId& peer); - void restoreRollback(const PeerId& peer); + void captureRollback(); + void restoreRollback(); LocationShareScheduler& scheduler_; PeerLocationStore& peers_; @@ -84,9 +84,7 @@ private: bool has_monotonic_observation_ = false; bool dirty_ = false; - ShareSession rollback_session_{}; - uint64_t rollback_revision_ = 0; - bool rollback_existed_ = false; + LocationShareScheduler rollback_scheduler_{}; }; } // namespace Telemetry diff --git a/lib/tdeck_ui/UI/LXMF/UIManager.cpp b/lib/tdeck_ui/UI/LXMF/UIManager.cpp index 44c2940f..526b76de 100644 --- a/lib/tdeck_ui/UI/LXMF/UIManager.cpp +++ b/lib/tdeck_ui/UI/LXMF/UIManager.cpp @@ -407,6 +407,11 @@ bool UIManager::init() { _announce_list_screen->set_send_announce_callback( [this]() { INFO("Sending LXMF announce..."); + RouterLock router_lock(0); + if (!router_lock.acquired()) { + WARNING("Router busy; announce deferred by user retry"); + return; + } try { _router.announce(); announce_lxst(); @@ -1116,6 +1121,11 @@ void UIManager::on_back_from_propagation_nodes() { } void UIManager::on_propagation_node_selected(const Bytes& node_hash) { + RouterLock router_lock(0); + if (!router_lock.acquired()) { + WARNING("Router busy; propagation selection not applied"); + return; + } std::string hash_hex = node_hash.toHex().substr(0, 16); std::string msg = "Propagation node selected: " + hash_hex + "..."; INFO(msg.c_str()); @@ -1150,6 +1160,11 @@ void UIManager::on_propagation_node_selected(const Bytes& node_hash) { } void UIManager::on_propagation_auto_select_changed(bool enabled) { + RouterLock router_lock(0); + if (!router_lock.acquired()) { + WARNING("Router busy; propagation mode not applied"); + return; + } std::string msg = "Propagation auto-select changed: "; msg += enabled ? "enabled" : "disabled"; INFO(msg.c_str()); @@ -1172,6 +1187,11 @@ void UIManager::on_propagation_auto_select_changed(bool enabled) { } void UIManager::on_propagation_sync() { + RouterLock router_lock(0); + if (!router_lock.acquired()) { + WARNING("Router busy; sync request not queued"); + return; + } INFO("Requesting messages from propagation node"); _router.request_messages_from_propagation_node(); } @@ -1254,9 +1274,9 @@ bool UIManager::send_message(const Bytes& dest_hash, const String& content) { // Queue for sending (pack already called, will use cached packed data). // Router queues are shared with loopTask; serialize the ownership copy. { - RouterLock router_lock; + RouterLock router_lock(0); if (!router_lock.acquired()) { - ERROR("Router lock unavailable; message not queued"); + WARNING("Router busy; message saved but user retry is required"); return false; } _router.handle_outbound(message); @@ -1324,19 +1344,29 @@ void UIManager::on_message_received(::LXMF::LXMessage& message) { WARNING("Malformed inbound location field ignored"); } if (location_decision.apply_location) { - const Telemetry::PeerLocationResult location_result = - _peer_locations.apply( - location_decision.authenticated_sender, - location_decision.location, - location_decision.meta, - location_decision.received_at_millis); - if (_location_persistence_controller && - location_result != Telemetry::PeerLocationResult::STALE && - location_result != Telemetry::PeerLocationResult::NOT_FOUND && - location_result != Telemetry::PeerLocationResult::INVALID_ARGUMENT) { - _location_persistence_controller->service( - static_cast(RNS::Utilities::OS::ltime()), - monotonicMillis()); + const uint64_t location_wall_now = + static_cast(RNS::Utilities::OS::ltime()); + const uint64_t location_monotonic_now = monotonicMillis(); + const Telemetry::LocationControllerState location_state = + _location_persistence_controller + ? _location_persistence_controller->service( + location_wall_now, location_monotonic_now) + : Telemetry::LocationControllerState::BLOCKED; + if (location_state == Telemetry::LocationControllerState::READY) { + const Telemetry::PeerLocationResult location_result = + _peer_locations.apply( + location_decision.authenticated_sender, + location_decision.location, + location_decision.meta, + location_decision.received_at_millis); + if (location_result != Telemetry::PeerLocationResult::STALE && + location_result != Telemetry::PeerLocationResult::NOT_FOUND && + location_result != Telemetry::PeerLocationResult::INVALID_ARGUMENT) { + _location_persistence_controller->service( + location_wall_now, location_monotonic_now); + } + } else { + WARNING("Location state not restored; inbound update ignored"); } } if (!location_decision.persist) { diff --git a/src/main.cpp b/src/main.cpp index 4d711702..4303b249 100644 --- a/src/main.cpp +++ b/src/main.cpp @@ -1415,6 +1415,11 @@ void setup_ui_manager() { // Set save callback (update app_settings and apply) settings->set_save_callback([](const UI::LXMF::AppSettings& new_settings) { + UI::LXMF::RouterLock router_lock(0); + if (!router_lock.acquired()) { + WARNING("Router busy; settings application deferred by user retry"); + return; + } // Check what changed bool wifi_settings_changed = (new_settings.wifi_ssid != app_settings.wifi_ssid) || (new_settings.wifi_password != app_settings.wifi_password); @@ -1462,7 +1467,12 @@ void setup_ui_manager() { // Update router display name if (router && !new_settings.display_name.isEmpty()) { - router->set_display_name(new_settings.display_name.c_str()); + UI::LXMF::RouterLock router_lock(0); + if (router_lock.acquired()) { + router->set_display_name(new_settings.display_name.c_str()); + } else { + WARNING("Router busy; display-name update not applied"); + } } // Handle TCP interface changes at runtime @@ -1594,8 +1604,13 @@ void setup_ui_manager() { // Apply propagation settings to router if (router) { - router->set_fallback_to_propagation(new_settings.prop_fallback_enabled); - router->set_propagation_only(new_settings.prop_only); + UI::LXMF::RouterLock router_lock(0); + if (router_lock.acquired()) { + router->set_fallback_to_propagation(new_settings.prop_fallback_enabled); + router->set_propagation_only(new_settings.prop_only); + } else { + WARNING("Router busy; propagation settings not applied"); + } // When auto-select is enabled, save the current effective node for next boot if (new_settings.prop_auto_select && propagation_manager) { @@ -2824,6 +2839,7 @@ void loop() { { UI::LXMF::RouterLock router_lock; if (router_lock.acquired() && router) { + if (ble_interface_impl) ble_interface_impl->drain_inbound(4); router->process_outbound(); router->process_inbound(); router->process_sync(); @@ -2844,8 +2860,11 @@ void loop() { (lora_interface && lora_interface->online()) || (ble_interface && ble_interface->online()); if (router && has_online_interface) { - announce_reachable_destinations(); - INFO("Periodic announce sent (interval: " + std::to_string(app_settings.announce_interval) + "s)"); + UI::LXMF::RouterLock router_lock; + if (router_lock.acquired()) { + announce_reachable_destinations(); + INFO("Periodic announce sent (interval: " + std::to_string(app_settings.announce_interval) + "s)"); + } } } } @@ -2873,9 +2892,12 @@ void loop() { // Only sync if TCP is online (propagation nodes need network) bool tcp_online = tcp_interface && tcp_interface->online(); if (tcp_online) { - router->request_messages_from_propagation_node(); - last_sync = now; - INFO("Periodic propagation sync (interval: " + std::to_string(app_settings.sync_interval / 3600) + " hours)"); + UI::LXMF::RouterLock router_lock; + if (router_lock.acquired()) { + router->request_messages_from_propagation_node(); + last_sync = now; + INFO("Periodic propagation sync (interval: " + std::to_string(app_settings.sync_interval / 3600) + " hours)"); + } } } } @@ -2885,7 +2907,10 @@ void loop() { INFO("TCP interface reconnected - sending announce"); if (router) { delay(500); // Brief stabilization delay - announce_reachable_destinations(); + UI::LXMF::RouterLock router_lock; + if (router_lock.acquired()) { + announce_reachable_destinations(); + } } last_tcp_online = true; } diff --git a/tests/build_scripts/test_location_live_integration.py b/tests/build_scripts/test_location_live_integration.py index ce41e5e0..0f139cd2 100644 --- a/tests/build_scripts/test_location_live_integration.py +++ b/tests/build_scripts/test_location_live_integration.py @@ -55,8 +55,7 @@ def test_inbound_location_and_consent_controls_use_controller_durability(): cpp = CPP.read_text() inbound = cpp[cpp.index("void UIManager::on_message_received") : cpp.index("void UIManager::on_message_delivered")] start = cpp[cpp.index("UIManager::start_location_sharing") : cpp.index("UIManager::get_location_share_session")] - assert "_location_persistence_controller->service" in inbound - assert "_peer_locations.apply" in inbound + assert inbound.index("_location_persistence_controller->service") < inbound.index("_peer_locations.apply") assert "_location_persistence_controller->startSharing" in start assert "_location_persistence_controller->stopSharing" in start @@ -91,3 +90,30 @@ def test_live_and_chat_outbound_share_a_router_mutex(): main.index("// Update UI manager") ] assert "RouterLock" in network_pump + assert "RouterLock router_lock(0)" in send + + +def test_ble_ingress_and_ui_router_mutators_follow_nonblocking_lock_order(): + cpp = CPP.read_text() + main = (ROOT / "src/main.cpp").read_text() + ble = (ROOT / "lib/ble_interface/BLEInterface.cpp").read_text() + announce = cpp[ + cpp.index("set_send_announce_callback") : + cpp.index("// Set up callbacks for status screen") + ] + propagation = cpp[ + cpp.index("void UIManager::on_propagation_node_selected") : + cpp.index("void UIManager::set_rns_status") + ] + assert "RouterLock router_lock(0)" in announce + assert propagation.count("RouterLock router_lock(0)") >= 3 + assert main.count("RouterLock router_lock(0)") >= 2 + reassembled = ble[ + ble.index("void BLEInterface::onPacketReassembled") : + ble.index("size_t BLEInterface::drain_inbound") + ] + assert "handle_incoming" not in reassembled + assert "drain_inbound" in ble + assert "ble_interface_impl->drain_inbound" in main + stop = ble[ble.index("void BLEInterface::stop()") : ble.index("void BLEInterface::loop()")] + assert "_pending_packet_count = 0" in stop diff --git a/tests/native/test_location_persistence.cpp b/tests/native/test_location_persistence.cpp index 41c2af5e..e6225c24 100644 --- a/tests/native/test_location_persistence.cpp +++ b/tests/native/test_location_persistence.cpp @@ -216,6 +216,21 @@ void validatesTempBeforeReplacingLive() { CHECK(marker(output) == 30); } +void successfulPromotionIsCommittedWhenReadbackIoFails() { + FakeStorage storage; + put(storage, Telemetry::LocationPersistenceSlot::LIVE, state(10)); + Telemetry::TransactionalLocationPersistence persistence(storage); + storage.fail_at = 11; // promoted LIVE stat/read-back cannot be observed + CHECK(persistence.save(state(20)) == + Telemetry::LocationPersistenceResult::SAVED); + storage.fail_at = 0; + storage.operations = 0; + Telemetry::LocationStateSnapshot output{}; + CHECK(persistence.load(output) == + Telemetry::LocationPersistenceResult::LOADED_LIVE); + CHECK(marker(output) == 20); +} + void transientHigherPriorityIoNeverFallsBackOrRepairs() { FakeStorage storage; put(storage, Telemetry::LocationPersistenceSlot::LIVE, state(20)); @@ -256,6 +271,7 @@ int main() { failsClosedWhenUnavailableMissingOrCorrupt(); everyInterruptedSaveRetainsAValidGeneration(); validatesTempBeforeReplacingLive(); + successfulPromotionIsCommittedWhenReadbackIoFails(); transientHigherPriorityIoNeverFallsBackOrRepairs(); std::cout << "location persistence: " << passed << " passed, " << failures << " failed\n"; diff --git a/tests/native/test_location_persistence_controller.cpp b/tests/native/test_location_persistence_controller.cpp index 5184f4b2..c8076e22 100644 --- a/tests/native/test_location_persistence_controller.cpp +++ b/tests/native/test_location_persistence_controller.cpp @@ -120,6 +120,8 @@ void clockGateRestoreAndPrune() { Telemetry::PeerLocationRecord record{}; CHECK(peers.get(peer(3), record)); CHECK(record.has_approx_radius && record.approx_radius_meters == 0); + CHECK(controller.service(1000, 30) == + Telemetry::LocationControllerState::BLOCKED); } void unavailableCorruptAndIoRetryFailClosed() { @@ -207,6 +209,28 @@ void urgentConsentSaveRollsBackAndRollbackClockBlocks() { CHECK(shares.get(peer(5), session) && !session.cease_pending); CHECK(controller.service(WALL + 2, 102) == Telemetry::LocationControllerState::BLOCKED); } + +void failedConsentRestoresWholeScheduler() { + MemoryStorage storage; + Telemetry::PeerLocationStore peers; + Telemetry::LocationShareScheduler shares; + Telemetry::TransactionalLocationPersistence persistence(storage); + Telemetry::LocationPersistenceController controller(shares, peers, persistence); + CHECK(controller.service(WALL, 100) == Telemetry::LocationControllerState::READY); + Telemetry::ShareStartOptions options{}; + options.duration = Telemetry::ShareDuration::INDEFINITE; + CHECK(controller.startSharing(peer(6), options, WALL + 1000, 101) == + Telemetry::LocationConsentResult::STARTED); + Telemetry::ShareSession before{}; + CHECK(shares.get(peer(6), before)); + storage.fail_io = true; + CHECK(controller.startSharing(peer(5), options, WALL, 102) == + Telemetry::LocationConsentResult::STORAGE_FAILURE); + Telemetry::ShareSession after{}; + CHECK(shares.get(peer(6), after)); + CHECK(after.next_attempt_millis == before.next_attempt_millis); + CHECK(after.last_sent_millis == before.last_sent_millis); +} } // namespace int main() { @@ -214,6 +238,7 @@ int main() { unavailableCorruptAndIoRetryFailClosed(); dirtyCadenceIsNonSlidingAndRetries(); urgentConsentSaveRollsBackAndRollbackClockBlocks(); + failedConsentRestoresWholeScheduler(); std::cout << "location persistence controller: " << passed << " passed, " << failures << " failed\n"; return failures == 0 ? EXIT_SUCCESS : EXIT_FAILURE;