Immediate command forwarding via module-controlled forward_command_on_receive#63
Open
danprudky wants to merge 9 commits into
Open
Immediate command forwarding via module-controlled forward_command_on_receive#63danprudky wants to merge 9 commits into
danprudky wants to merge 9 commits into
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
WalkthroughThis PR integrates QUIC protocol support with MsQuic, introduces immediate device command forwarding via a shared queue, refactors module error aggregation under mutex protection, detects server-initiated disconnects, and simplifies the module async handler's buffer ownership model. The changes span communication channels, module management, configuration parsing, and build system setup. ChangesQUIC Protocol and Core Communication
Immediate Device Command Forwarding
Error Aggregation and Concurrency Protection
Server Disconnect Detection and Handling
Connect Sequence Queue Draining
Module Handler and Library Refactoring
Build System and Configuration
InternalServer Buffer Processing Fix
🎯 4 (Complex) | ⏱️ ~60 minutes Possibly Related PRs
Suggested Reviewers
✨ Finishing Touches🧪 Generate unit tests (beta)
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
source/bringauto/modules/StatusAggregator.cpp (1)
101-106:⚠️ Potential issue | 🟠 Major | ⚡ Quick winData race:
deviceTimeouts_[deviceId]++called without holdingdevicesMutex_.The
timeouted_force_aggregationlambda capturesthisand modifiesdeviceTimeouts_at line 104 before callingforce_aggregation_on_device(). This lambda runs on the timer thread (viaio_context), while other methods accessdeviceTimeouts_underdevicesMutex_(e.g.,getDeviceTimeoutCountat line 260, initialization at line 93). This creates a data race.Proposed fix: Lock before modifying deviceTimeouts_
const std::function<int(const structures::DeviceIdentification&)> timeouted_force_aggregation = [this]( const structures::DeviceIdentification& deviceId) { + std::lock_guard lock(devicesMutex_); timeoutedMessageReady_.store(true); deviceTimeouts_[deviceId]++; return force_aggregation_on_device(deviceId); };Note: Since
force_aggregation_on_devicealso locksdevicesMutex_, the recursive_mutex will handle the nested locking correctly.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/modules/StatusAggregator.cpp` around lines 101 - 106, The lambda timeouted_force_aggregation mutates deviceTimeouts_ from the timer thread without holding devicesMutex_, causing a data race; modify the lambda to acquire the existing devicesMutex_ (e.g., std::lock_guard or std::unique_lock on devicesMutex_) before incrementing deviceTimeouts_[deviceId], then release (or rely on recursive_mutex for nested locking) and call force_aggregation_on_device(deviceId); ensure you reference devicesMutex_, deviceTimeouts_, timeouted_force_aggregation, and force_aggregation_on_device when applying the change.
🧹 Nitpick comments (7)
CMLibStorage.cmake (1)
7-8: ⚡ Quick winConsider pinning the revision with
CACHE STRING "" FORCE.
STORAGE_LIST_DEP_REVISIONis set with a plainSET. To propagate this pin globally across submodules that don't declare their own revision, prefer caching it withFORCE.♻️ Proposed change
SET(STORAGE_LIST_DEP "https://github.com/bacpack-system/package-tracker.git") -SET(STORAGE_LIST_DEP_REVISION "v2.0.0") +SET(STORAGE_LIST_DEP_REVISION "v2.0.0" CACHE STRING "" FORCE)Be mindful that
FORCEoverwrites local values; confirm downstream modules are compatible withv2.0.0.Based on learnings: "Use CACHE STRING "" FORCE for revision variables (e.g., STORAGE_LIST_DEP_REVISION) in CMake helper files (like CMLibStorage.cmake) when you intend to propagate version pins globally across submodules."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@CMLibStorage.cmake` around lines 7 - 8, Replace the plain SET of STORAGE_LIST_DEP_REVISION with a cached variable so the revision pin propagates to submodules; specifically change the assignment for STORAGE_LIST_DEP_REVISION to use CMake's CACHE STRING "" FORCE (keeping the same variable name STORAGE_LIST_DEP_REVISION and the current value "v2.0.0") so downstream modules that don't declare their own revision will inherit this pin while being aware FORCE will overwrite local values.CLAUDE.md (1)
52-65: 💤 Low valueAdd a language to the fenced architecture diagram block.
markdownlint (MD040) flags this fence. Use
text(or similar) to satisfy the linter.📝 Proposed change
-``` +```text Internal Client ──► Internal Server ──► [fromInternalQueue] ──► Module Handler🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@CLAUDE.md` around lines 52 - 65, The fenced architecture diagram in CLAUDE.md triggers markdownlint MD040 because it lacks a language identifier; update the fenced code block that contains the ASCII diagram (the block starting with the triple backticks showing "Internal Client ──► Internal Server ... External Server (MQTT / QUIC / Dummy)") to include a language label such as text (e.g. change ``` to ```text) so the linter is satisfied while keeping the diagram content unchanged.include/bringauto/external_client/connection/communication/MqttCommunication.hpp (1)
78-78: ⚡ Quick winInclude
<atomic>explicitly.
std::atomic<bool> serverDisconnectPending_is declared here, but the header only includes<string>,<mqtt/async_client.h>, and the channel interface. Relying on a transitive include for<atomic>is fragile; add it directly.🛡️ Proposed change
`#include` <mqtt/async_client.h> `#include` <string> +#include <atomic>🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@include/bringauto/external_client/connection/communication/MqttCommunication.hpp` at line 78, The header MqttCommunication.hpp declares std::atomic<bool> serverDisconnectPending_ but doesn't include <atomic>, relying on transitive includes; add a direct include for <atomic> at the top of MqttCommunication.hpp so the declaration of serverDisconnectPending_ (and any future atomic use) is well-formed and not fragile.include/bringauto/settings/Constants.hpp (1)
217-218: 💤 Low value
AERON_CONNECTIONandSEPARATORduplicate existingAeronClientConstantsvalues.These literals (
"aeron:ipc",":::") already exist asAeronClientConstants::aeron_connection(Line 154) andAeronClientConstants::separator(Line 158). Consider referencing those to avoid divergence if either value changes.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@include/bringauto/settings/Constants.hpp` around lines 217 - 218, AERON_CONNECTION and SEPARATOR in Constants.hpp duplicate existing literals; replace their inline constexpr string_view definitions so they reference the canonical AeronClientConstants values (use AeronClientConstants::aeron_connection and AeronClientConstants::separator) instead of hard-coded literals, updating the AERON_CONNECTION and SEPARATOR symbols accordingly to prevent future divergence.source/bringauto/settings/QuicSettingsParser.cpp (1)
52-59: 💤 Low valueOptional: avoid parsing the JSON twice.
nlohmann::json::accept(raw)parses the string just to validate, thenparse(raw)parses it again. You can parse once and guard with atry/catch(orparse(raw, nullptr, false)and checkis_discarded()).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/settings/QuicSettingsParser.cpp` around lines 52 - 59, Replace the double-parse pattern around nlohmann::json::accept(raw) and nlohmann::json::parse(raw) by parsing only once and handling errors: either call nlohmann::json::parse(raw, nullptr, false) and check j.is_discarded() to log the same warning via settings::Logger::logWarning("[quic] QUIC setting '{}' is not valid JSON", key) and return std::nullopt, or wrap a single nlohmann::json::parse(raw) in a try/catch for nlohmann::json::parse_error and handle the warning/return there; update the code that assigns to auto j to use this single-parse approach and remove the nlohmann::json::accept(raw) usage.source/bringauto/modules/ModuleHandler.cpp (1)
17-24: 💤 Low value
destroy()drains queues without synchronization with running threads.If
destroy()is called whilehandleMessages()orhandleCommandForwards()threads are still running, there's a potential race on the queues. This appears safe only ifdestroy()is called afterioContext.stop(), which causes the loops to exit. Consider adding a comment documenting this precondition, or adding explicit thread join/stop coordination.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/modules/ModuleHandler.cpp` around lines 17 - 24, ModuleHandler::destroy() currently drains fromInternalQueue_ and commandForwardingQueue_ without synchronizing with threads running handleMessages() or handleCommandForwards(), risking races; either document the precondition that ioContext.stop() (or equivalent) must be called and those threads joined before destroy(), or implement explicit coordination: signal thread shutdown, join the threads that run handleMessages()/handleCommandForwards(), or protect the queues with a mutex/condition_variable and ensure no concurrent access before popping in destroy(); update the method comment to reference the required precondition (ioContext.stop() and thread joins) if you choose the documentation route.include/bringauto/structures/StatusAggregatorDeviceState.hpp (1)
107-107: 💤 Low value
std::unique_ptr<std::mutex>leaves moved-from objects unsafe.After a move, the source object's
externalCommandMutex_becomesnullptr. If any method (likeconsumeCommandoraddExternalCommand) is called on a moved-from instance, it will dereference a null pointer. This is acceptable only if moved-from objects are never accessed, but is fragile ifStatusAggregatorDeviceStateis ever stored in a container that moves elements (e.g.,std::vectorreallocation).Consider making the class non-movable or using a
std::shared_ptr<std::mutex>if shared ownership is acceptable, or documenting the moved-from invariant explicitly.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@include/bringauto/structures/StatusAggregatorDeviceState.hpp` at line 107, The class currently stores the mutex as std::unique_ptr<std::mutex> (externalCommandMutex_) which becomes nullptr after a move, making methods like consumeCommand and addExternalCommand unsafe on moved-from instances; fix by either (A) making StatusAggregatorDeviceState non-movable (delete move ctor and move assignment) so the pointer never gets moved, or (B) replace std::unique_ptr<std::mutex> externalCommandMutex_ with an owning, non-nullable mutex (e.g., std::mutex externalCommandMutex_) or a std::shared_ptr<std::mutex> if shared ownership is needed; update or document invariants accordingly and ensure consumeCommand/addExternalCommand and any other methods using externalCommandMutex_ assume it is always valid.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@resources/config/quic_example.json`:
- Around line 25-33: The JSON example has inconsistent indentation (a TAB before
the "alpn" key) which is the only remaining issue; open the example and replace
the leading tab on the "alpn" line with the same spaces used elsewhere so all
keys ("ca-file", "client-cert", "client-key", "alpn", "stream-mode",
"DisconnectTimeoutMs", "PeerUnidiStreamCount") are consistently indented; no key
renaming is needed because getProtocolSettingsString(), Constants.hpp (CA_FILE,
CLIENT_CERT, CLIENT_KEY, ALPN, STREAM_MODE) and QuicSettingsParser.cpp already
expect the shown key names.
In `@resources/config/README.md`:
- Around line 42-48: Update the quic-settings section in README.md to document
the additional keys used in quic_example.json: add entries for "stream-mode"
(describe allowed values e.g., "bidirectional" or "unidirectional" and default),
"DisconnectTimeoutMs" (describe it's an integer timeout in milliseconds and
default), and "PeerUnidiStreamCount" (describe it's an integer limiting
peer-initiated unidirectional streams and default); ensure each entry mirrors
the style of the existing keys (key name, purpose, type, allowed values, and
default) so the schema doc matches the shipped example.
In
`@source/bringauto/external_client/connection/communication/QuicCommunication.cpp`:
- Around line 261-273: Stream opened by quic_->StreamOpen is leaked if
message.SerializeToArray fails; ensure you close the opened stream before
returning on serialization failure. After quic_->StreamOpen(...) succeeds and
before returning on a failed SerializeToArray call, call the appropriate close
API (e.g. quic_->StreamClose(stream) or the library equivalent) to release the
stream handle (reference the local variable stream and the call sites
quic_->StreamOpen and message.SerializeToArray), or alternatively move
creation/serialization logic so the stream is only opened after successful
serialization.
- Around line 40-73: In QuicCommunication::initializeConnection, ensure
connectionState_ is rolled back to ConnectionState::NOT_CONNECTED if
quic_->ConnectionOpen or quic_->ConnectionStart fails; after a failed QUIC call
(QUIC_FAILED(status)) set connectionState_ back to NOT_CONNECTED (use the same
atomic semantics as the initial compare_exchange_strong or an atomic store) and
return, and remove the redundant connectionState_ = ConnectionState::CONNECTING
at the end since the state was already set by compare_exchange_strong when
entering the function; reference symbols: initializeConnection,
connectionState_, ConnectionOpen, ConnectionStart, compare_exchange_strong,
QUIC_FAILED.
- Around line 231-241: QuicCommunication::stop currently calls closeConnection()
(which initiates an async ConnectionShutdown) then immediately calls
closeMsQuic(), causing a use-after-free when
QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE runs in connectionCallback and uses
quic_. Add a shutdown-complete flag (e.g., shutdown_complete_) and a
condition_variable (e.g., shutdownCv_) plus mutex; set shutdown_complete_ and
notify in connectionCallback when handling
QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE, and have QuicCommunication::stop wait
(with a sensible timeout) on shutdownCv_ until shutdown_complete_ is true before
calling closeMsQuic(); also add defensive checks in connectionCallback to ensure
quic_ is valid before calling quic_->ConnectionClose(connection).
In `@source/bringauto/modules/ModuleManagerLibraryHandlerAsync.cpp`:
- Around line 70-72: Multiple methods (getModuleNumber, isDeviceTypeSupported,
sendStatusCondition, generateCommand, aggregateStatus, aggregateError,
generateFirstCommand, statusDataValid, commandDataValid) call
aeronClient.callFunc(...).value() which will throw std::bad_optional_access on
timeout/failure; update each to check the returned std::optional with
has_value() (or use operator bool) before accessing the value and handle the
empty case gracefully (e.g., return a sensible error code/optional/false, or
propagate an error) instead of calling .value(); follow the same polling/guard
pattern used in loadLibrary to avoid exceptions and ensure consistent error
handling for the fp_async::* calls.
In `@source/bringauto/settings/QuicSettingsParser.cpp`:
- Around line 26-36: The code currently assigns values from getOptionalUint
(std::optional<uint64_t>) directly into QUIC_SETTINGS fields
(quic.DisconnectTimeoutMs:uint32_t and quic.PeerUnidiStreamCount:uint16_t) which
can silently truncate; add explicit range checks before assignment: after
obtaining value in the if (auto value = getOptionalUint(...)) blocks, verify the
uint64_t fits into the target width (<= UINT32_MAX for DisconnectTimeoutMs, <=
UINT16_MAX for PeerUnidiStreamCount), log an error/warning via
settings::Logger::logDebug or logError if out of range, and skip the assignment
and quic.IsSet.* flag when out of range; only assign and set
quic.IsSet.DisconnectTimeoutMs / quic.IsSet.PeerUnidiStreamCount when the value
passes the range check.
---
Outside diff comments:
In `@source/bringauto/modules/StatusAggregator.cpp`:
- Around line 101-106: The lambda timeouted_force_aggregation mutates
deviceTimeouts_ from the timer thread without holding devicesMutex_, causing a
data race; modify the lambda to acquire the existing devicesMutex_ (e.g.,
std::lock_guard or std::unique_lock on devicesMutex_) before incrementing
deviceTimeouts_[deviceId], then release (or rely on recursive_mutex for nested
locking) and call force_aggregation_on_device(deviceId); ensure you reference
devicesMutex_, deviceTimeouts_, timeouted_force_aggregation, and
force_aggregation_on_device when applying the change.
---
Nitpick comments:
In `@CLAUDE.md`:
- Around line 52-65: The fenced architecture diagram in CLAUDE.md triggers
markdownlint MD040 because it lacks a language identifier; update the fenced
code block that contains the ASCII diagram (the block starting with the triple
backticks showing "Internal Client ──► Internal Server ... External Server (MQTT
/ QUIC / Dummy)") to include a language label such as text (e.g. change ``` to
```text) so the linter is satisfied while keeping the diagram content unchanged.
In `@CMLibStorage.cmake`:
- Around line 7-8: Replace the plain SET of STORAGE_LIST_DEP_REVISION with a
cached variable so the revision pin propagates to submodules; specifically
change the assignment for STORAGE_LIST_DEP_REVISION to use CMake's CACHE STRING
"" FORCE (keeping the same variable name STORAGE_LIST_DEP_REVISION and the
current value "v2.0.0") so downstream modules that don't declare their own
revision will inherit this pin while being aware FORCE will overwrite local
values.
In
`@include/bringauto/external_client/connection/communication/MqttCommunication.hpp`:
- Line 78: The header MqttCommunication.hpp declares std::atomic<bool>
serverDisconnectPending_ but doesn't include <atomic>, relying on transitive
includes; add a direct include for <atomic> at the top of MqttCommunication.hpp
so the declaration of serverDisconnectPending_ (and any future atomic use) is
well-formed and not fragile.
In `@include/bringauto/settings/Constants.hpp`:
- Around line 217-218: AERON_CONNECTION and SEPARATOR in Constants.hpp duplicate
existing literals; replace their inline constexpr string_view definitions so
they reference the canonical AeronClientConstants values (use
AeronClientConstants::aeron_connection and AeronClientConstants::separator)
instead of hard-coded literals, updating the AERON_CONNECTION and SEPARATOR
symbols accordingly to prevent future divergence.
In `@include/bringauto/structures/StatusAggregatorDeviceState.hpp`:
- Line 107: The class currently stores the mutex as std::unique_ptr<std::mutex>
(externalCommandMutex_) which becomes nullptr after a move, making methods like
consumeCommand and addExternalCommand unsafe on moved-from instances; fix by
either (A) making StatusAggregatorDeviceState non-movable (delete move ctor and
move assignment) so the pointer never gets moved, or (B) replace
std::unique_ptr<std::mutex> externalCommandMutex_ with an owning, non-nullable
mutex (e.g., std::mutex externalCommandMutex_) or a std::shared_ptr<std::mutex>
if shared ownership is needed; update or document invariants accordingly and
ensure consumeCommand/addExternalCommand and any other methods using
externalCommandMutex_ assume it is always valid.
In `@source/bringauto/modules/ModuleHandler.cpp`:
- Around line 17-24: ModuleHandler::destroy() currently drains
fromInternalQueue_ and commandForwardingQueue_ without synchronizing with
threads running handleMessages() or handleCommandForwards(), risking races;
either document the precondition that ioContext.stop() (or equivalent) must be
called and those threads joined before destroy(), or implement explicit
coordination: signal thread shutdown, join the threads that run
handleMessages()/handleCommandForwards(), or protect the queues with a
mutex/condition_variable and ensure no concurrent access before popping in
destroy(); update the method comment to reference the required precondition
(ioContext.stop() and thread joins) if you choose the documentation route.
In `@source/bringauto/settings/QuicSettingsParser.cpp`:
- Around line 52-59: Replace the double-parse pattern around
nlohmann::json::accept(raw) and nlohmann::json::parse(raw) by parsing only once
and handling errors: either call nlohmann::json::parse(raw, nullptr, false) and
check j.is_discarded() to log the same warning via
settings::Logger::logWarning("[quic] QUIC setting '{}' is not valid JSON", key)
and return std::nullopt, or wrap a single nlohmann::json::parse(raw) in a
try/catch for nlohmann::json::parse_error and handle the warning/return there;
update the code that assigns to auto j to use this single-parse approach and
remove the nlohmann::json::accept(raw) usage.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 4a8ac9b7-66a5-491e-951f-7a2d395d2f1f
📒 Files selected for processing (50)
CLAUDE.mdCMLibStorage.cmakeCMakeLists.txtcmake/Dependencies.cmakeinclude/bringauto/common_utils/EnumUtils.hppinclude/bringauto/external_client/ExternalClient.hppinclude/bringauto/external_client/connection/ConnectionState.hppinclude/bringauto/external_client/connection/ExternalConnection.hppinclude/bringauto/external_client/connection/communication/DummyCommunication.hppinclude/bringauto/external_client/connection/communication/ICommunicationChannel.hppinclude/bringauto/external_client/connection/communication/MqttCommunication.hppinclude/bringauto/external_client/connection/communication/QuicCommunication.hppinclude/bringauto/modules/IModuleManagerLibraryHandler.hppinclude/bringauto/modules/ModuleBinaryException.hppinclude/bringauto/modules/ModuleHandler.hppinclude/bringauto/modules/ModuleManagerLibraryHandlerAsync.hppinclude/bringauto/modules/ModuleManagerLibraryHandlerLocal.hppinclude/bringauto/modules/StatusAggregator.hppinclude/bringauto/modules/library_loader.hppinclude/bringauto/settings/Constants.hppinclude/bringauto/settings/QuicSettingsParser.hppinclude/bringauto/structures/ExternalConnectionSettings.hppinclude/bringauto/structures/InternalClientMessage.hppinclude/bringauto/structures/ModuleLibrary.hppinclude/bringauto/structures/StatusAggregatorDeviceState.hppmain.cppresources/config/README.mdresources/config/quic_example.jsonsource/bringauto/common_utils/EnumUtils.cppsource/bringauto/external_client/ExternalClient.cppsource/bringauto/external_client/connection/ExternalConnection.cppsource/bringauto/external_client/connection/communication/MqttCommunication.cppsource/bringauto/external_client/connection/communication/QuicCommunication.cppsource/bringauto/internal_server/InternalServer.cppsource/bringauto/modules/ModuleHandler.cppsource/bringauto/modules/ModuleManagerLibraryHandlerAsync.cppsource/bringauto/modules/ModuleManagerLibraryHandlerLocal.cppsource/bringauto/modules/StatusAggregator.cppsource/bringauto/modules/library_loader_production.cppsource/bringauto/settings/QuicSettingsParser.cppsource/bringauto/settings/SettingsParser.cppsource/bringauto/structures/InternalClientMessage.cppsource/bringauto/structures/ModuleLibrary.cppsource/bringauto/structures/StatusAggregatorDeviceState.cpptest/include/ExternalConnectionTests.hpptest/include/testing_utils/CommunicationMock.hpptest/include/testing_utils/ConfigMock.hpptest/lib/example-moduletest/source/StatusAggregatorTests.cpptest/source/library_loader_test.cpp
There was a problem hiding this comment.
Caution
Inline review comments failed to post. This is likely due to GitHub's internal server error or limits when posting large numbers of comments. If you are seeing this consistently it is likely a permissions issue. Please check "Moderation" -> "Code review limits" under your organization settings.
Actionable comments posted: 7
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
source/bringauto/modules/StatusAggregator.cpp (1)
101-106:⚠️ Potential issue | 🟠 Major | ⚡ Quick winData race:
deviceTimeouts_[deviceId]++called without holdingdevicesMutex_.The
timeouted_force_aggregationlambda capturesthisand modifiesdeviceTimeouts_at line 104 before callingforce_aggregation_on_device(). This lambda runs on the timer thread (viaio_context), while other methods accessdeviceTimeouts_underdevicesMutex_(e.g.,getDeviceTimeoutCountat line 260, initialization at line 93). This creates a data race.Proposed fix: Lock before modifying deviceTimeouts_
const std::function<int(const structures::DeviceIdentification&)> timeouted_force_aggregation = [this]( const structures::DeviceIdentification& deviceId) { + std::lock_guard lock(devicesMutex_); timeoutedMessageReady_.store(true); deviceTimeouts_[deviceId]++; return force_aggregation_on_device(deviceId); };Note: Since
force_aggregation_on_devicealso locksdevicesMutex_, the recursive_mutex will handle the nested locking correctly.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/modules/StatusAggregator.cpp` around lines 101 - 106, The lambda timeouted_force_aggregation mutates deviceTimeouts_ from the timer thread without holding devicesMutex_, causing a data race; modify the lambda to acquire the existing devicesMutex_ (e.g., std::lock_guard or std::unique_lock on devicesMutex_) before incrementing deviceTimeouts_[deviceId], then release (or rely on recursive_mutex for nested locking) and call force_aggregation_on_device(deviceId); ensure you reference devicesMutex_, deviceTimeouts_, timeouted_force_aggregation, and force_aggregation_on_device when applying the change.
🧹 Nitpick comments (7)
CMLibStorage.cmake (1)
7-8: ⚡ Quick winConsider pinning the revision with
CACHE STRING "" FORCE.
STORAGE_LIST_DEP_REVISIONis set with a plainSET. To propagate this pin globally across submodules that don't declare their own revision, prefer caching it withFORCE.♻️ Proposed change
SET(STORAGE_LIST_DEP "https://github.com/bacpack-system/package-tracker.git") -SET(STORAGE_LIST_DEP_REVISION "v2.0.0") +SET(STORAGE_LIST_DEP_REVISION "v2.0.0" CACHE STRING "" FORCE)Be mindful that
FORCEoverwrites local values; confirm downstream modules are compatible withv2.0.0.Based on learnings: "Use CACHE STRING "" FORCE for revision variables (e.g., STORAGE_LIST_DEP_REVISION) in CMake helper files (like CMLibStorage.cmake) when you intend to propagate version pins globally across submodules."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@CMLibStorage.cmake` around lines 7 - 8, Replace the plain SET of STORAGE_LIST_DEP_REVISION with a cached variable so the revision pin propagates to submodules; specifically change the assignment for STORAGE_LIST_DEP_REVISION to use CMake's CACHE STRING "" FORCE (keeping the same variable name STORAGE_LIST_DEP_REVISION and the current value "v2.0.0") so downstream modules that don't declare their own revision will inherit this pin while being aware FORCE will overwrite local values.CLAUDE.md (1)
52-65: 💤 Low valueAdd a language to the fenced architecture diagram block.
markdownlint (MD040) flags this fence. Use
text(or similar) to satisfy the linter.📝 Proposed change
-``` +```text Internal Client ──► Internal Server ──► [fromInternalQueue] ──► Module Handler🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@CLAUDE.md` around lines 52 - 65, The fenced architecture diagram in CLAUDE.md triggers markdownlint MD040 because it lacks a language identifier; update the fenced code block that contains the ASCII diagram (the block starting with the triple backticks showing "Internal Client ──► Internal Server ... External Server (MQTT / QUIC / Dummy)") to include a language label such as text (e.g. change ``` to ```text) so the linter is satisfied while keeping the diagram content unchanged.include/bringauto/external_client/connection/communication/MqttCommunication.hpp (1)
78-78: ⚡ Quick winInclude
<atomic>explicitly.
std::atomic<bool> serverDisconnectPending_is declared here, but the header only includes<string>,<mqtt/async_client.h>, and the channel interface. Relying on a transitive include for<atomic>is fragile; add it directly.🛡️ Proposed change
`#include` <mqtt/async_client.h> `#include` <string> +#include <atomic>🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@include/bringauto/external_client/connection/communication/MqttCommunication.hpp` at line 78, The header MqttCommunication.hpp declares std::atomic<bool> serverDisconnectPending_ but doesn't include <atomic>, relying on transitive includes; add a direct include for <atomic> at the top of MqttCommunication.hpp so the declaration of serverDisconnectPending_ (and any future atomic use) is well-formed and not fragile.include/bringauto/settings/Constants.hpp (1)
217-218: 💤 Low value
AERON_CONNECTIONandSEPARATORduplicate existingAeronClientConstantsvalues.These literals (
"aeron:ipc",":::") already exist asAeronClientConstants::aeron_connection(Line 154) andAeronClientConstants::separator(Line 158). Consider referencing those to avoid divergence if either value changes.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@include/bringauto/settings/Constants.hpp` around lines 217 - 218, AERON_CONNECTION and SEPARATOR in Constants.hpp duplicate existing literals; replace their inline constexpr string_view definitions so they reference the canonical AeronClientConstants values (use AeronClientConstants::aeron_connection and AeronClientConstants::separator) instead of hard-coded literals, updating the AERON_CONNECTION and SEPARATOR symbols accordingly to prevent future divergence.source/bringauto/settings/QuicSettingsParser.cpp (1)
52-59: 💤 Low valueOptional: avoid parsing the JSON twice.
nlohmann::json::accept(raw)parses the string just to validate, thenparse(raw)parses it again. You can parse once and guard with atry/catch(orparse(raw, nullptr, false)and checkis_discarded()).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/settings/QuicSettingsParser.cpp` around lines 52 - 59, Replace the double-parse pattern around nlohmann::json::accept(raw) and nlohmann::json::parse(raw) by parsing only once and handling errors: either call nlohmann::json::parse(raw, nullptr, false) and check j.is_discarded() to log the same warning via settings::Logger::logWarning("[quic] QUIC setting '{}' is not valid JSON", key) and return std::nullopt, or wrap a single nlohmann::json::parse(raw) in a try/catch for nlohmann::json::parse_error and handle the warning/return there; update the code that assigns to auto j to use this single-parse approach and remove the nlohmann::json::accept(raw) usage.source/bringauto/modules/ModuleHandler.cpp (1)
17-24: 💤 Low value
destroy()drains queues without synchronization with running threads.If
destroy()is called whilehandleMessages()orhandleCommandForwards()threads are still running, there's a potential race on the queues. This appears safe only ifdestroy()is called afterioContext.stop(), which causes the loops to exit. Consider adding a comment documenting this precondition, or adding explicit thread join/stop coordination.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/modules/ModuleHandler.cpp` around lines 17 - 24, ModuleHandler::destroy() currently drains fromInternalQueue_ and commandForwardingQueue_ without synchronizing with threads running handleMessages() or handleCommandForwards(), risking races; either document the precondition that ioContext.stop() (or equivalent) must be called and those threads joined before destroy(), or implement explicit coordination: signal thread shutdown, join the threads that run handleMessages()/handleCommandForwards(), or protect the queues with a mutex/condition_variable and ensure no concurrent access before popping in destroy(); update the method comment to reference the required precondition (ioContext.stop() and thread joins) if you choose the documentation route.include/bringauto/structures/StatusAggregatorDeviceState.hpp (1)
107-107: 💤 Low value
std::unique_ptr<std::mutex>leaves moved-from objects unsafe.After a move, the source object's
externalCommandMutex_becomesnullptr. If any method (likeconsumeCommandoraddExternalCommand) is called on a moved-from instance, it will dereference a null pointer. This is acceptable only if moved-from objects are never accessed, but is fragile ifStatusAggregatorDeviceStateis ever stored in a container that moves elements (e.g.,std::vectorreallocation).Consider making the class non-movable or using a
std::shared_ptr<std::mutex>if shared ownership is acceptable, or documenting the moved-from invariant explicitly.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@include/bringauto/structures/StatusAggregatorDeviceState.hpp` at line 107, The class currently stores the mutex as std::unique_ptr<std::mutex> (externalCommandMutex_) which becomes nullptr after a move, making methods like consumeCommand and addExternalCommand unsafe on moved-from instances; fix by either (A) making StatusAggregatorDeviceState non-movable (delete move ctor and move assignment) so the pointer never gets moved, or (B) replace std::unique_ptr<std::mutex> externalCommandMutex_ with an owning, non-nullable mutex (e.g., std::mutex externalCommandMutex_) or a std::shared_ptr<std::mutex> if shared ownership is needed; update or document invariants accordingly and ensure consumeCommand/addExternalCommand and any other methods using externalCommandMutex_ assume it is always valid.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@resources/config/quic_example.json`:
- Around line 25-33: The JSON example has inconsistent indentation (a TAB before
the "alpn" key) which is the only remaining issue; open the example and replace
the leading tab on the "alpn" line with the same spaces used elsewhere so all
keys ("ca-file", "client-cert", "client-key", "alpn", "stream-mode",
"DisconnectTimeoutMs", "PeerUnidiStreamCount") are consistently indented; no key
renaming is needed because getProtocolSettingsString(), Constants.hpp (CA_FILE,
CLIENT_CERT, CLIENT_KEY, ALPN, STREAM_MODE) and QuicSettingsParser.cpp already
expect the shown key names.
In `@resources/config/README.md`:
- Around line 42-48: Update the quic-settings section in README.md to document
the additional keys used in quic_example.json: add entries for "stream-mode"
(describe allowed values e.g., "bidirectional" or "unidirectional" and default),
"DisconnectTimeoutMs" (describe it's an integer timeout in milliseconds and
default), and "PeerUnidiStreamCount" (describe it's an integer limiting
peer-initiated unidirectional streams and default); ensure each entry mirrors
the style of the existing keys (key name, purpose, type, allowed values, and
default) so the schema doc matches the shipped example.
In
`@source/bringauto/external_client/connection/communication/QuicCommunication.cpp`:
- Around line 261-273: Stream opened by quic_->StreamOpen is leaked if
message.SerializeToArray fails; ensure you close the opened stream before
returning on serialization failure. After quic_->StreamOpen(...) succeeds and
before returning on a failed SerializeToArray call, call the appropriate close
API (e.g. quic_->StreamClose(stream) or the library equivalent) to release the
stream handle (reference the local variable stream and the call sites
quic_->StreamOpen and message.SerializeToArray), or alternatively move
creation/serialization logic so the stream is only opened after successful
serialization.
- Around line 40-73: In QuicCommunication::initializeConnection, ensure
connectionState_ is rolled back to ConnectionState::NOT_CONNECTED if
quic_->ConnectionOpen or quic_->ConnectionStart fails; after a failed QUIC call
(QUIC_FAILED(status)) set connectionState_ back to NOT_CONNECTED (use the same
atomic semantics as the initial compare_exchange_strong or an atomic store) and
return, and remove the redundant connectionState_ = ConnectionState::CONNECTING
at the end since the state was already set by compare_exchange_strong when
entering the function; reference symbols: initializeConnection,
connectionState_, ConnectionOpen, ConnectionStart, compare_exchange_strong,
QUIC_FAILED.
- Around line 231-241: QuicCommunication::stop currently calls closeConnection()
(which initiates an async ConnectionShutdown) then immediately calls
closeMsQuic(), causing a use-after-free when
QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE runs in connectionCallback and uses
quic_. Add a shutdown-complete flag (e.g., shutdown_complete_) and a
condition_variable (e.g., shutdownCv_) plus mutex; set shutdown_complete_ and
notify in connectionCallback when handling
QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE, and have QuicCommunication::stop wait
(with a sensible timeout) on shutdownCv_ until shutdown_complete_ is true before
calling closeMsQuic(); also add defensive checks in connectionCallback to ensure
quic_ is valid before calling quic_->ConnectionClose(connection).
In `@source/bringauto/modules/ModuleManagerLibraryHandlerAsync.cpp`:
- Around line 70-72: Multiple methods (getModuleNumber, isDeviceTypeSupported,
sendStatusCondition, generateCommand, aggregateStatus, aggregateError,
generateFirstCommand, statusDataValid, commandDataValid) call
aeronClient.callFunc(...).value() which will throw std::bad_optional_access on
timeout/failure; update each to check the returned std::optional with
has_value() (or use operator bool) before accessing the value and handle the
empty case gracefully (e.g., return a sensible error code/optional/false, or
propagate an error) instead of calling .value(); follow the same polling/guard
pattern used in loadLibrary to avoid exceptions and ensure consistent error
handling for the fp_async::* calls.
In `@source/bringauto/settings/QuicSettingsParser.cpp`:
- Around line 26-36: The code currently assigns values from getOptionalUint
(std::optional<uint64_t>) directly into QUIC_SETTINGS fields
(quic.DisconnectTimeoutMs:uint32_t and quic.PeerUnidiStreamCount:uint16_t) which
can silently truncate; add explicit range checks before assignment: after
obtaining value in the if (auto value = getOptionalUint(...)) blocks, verify the
uint64_t fits into the target width (<= UINT32_MAX for DisconnectTimeoutMs, <=
UINT16_MAX for PeerUnidiStreamCount), log an error/warning via
settings::Logger::logDebug or logError if out of range, and skip the assignment
and quic.IsSet.* flag when out of range; only assign and set
quic.IsSet.DisconnectTimeoutMs / quic.IsSet.PeerUnidiStreamCount when the value
passes the range check.
---
Outside diff comments:
In `@source/bringauto/modules/StatusAggregator.cpp`:
- Around line 101-106: The lambda timeouted_force_aggregation mutates
deviceTimeouts_ from the timer thread without holding devicesMutex_, causing a
data race; modify the lambda to acquire the existing devicesMutex_ (e.g.,
std::lock_guard or std::unique_lock on devicesMutex_) before incrementing
deviceTimeouts_[deviceId], then release (or rely on recursive_mutex for nested
locking) and call force_aggregation_on_device(deviceId); ensure you reference
devicesMutex_, deviceTimeouts_, timeouted_force_aggregation, and
force_aggregation_on_device when applying the change.
---
Nitpick comments:
In `@CLAUDE.md`:
- Around line 52-65: The fenced architecture diagram in CLAUDE.md triggers
markdownlint MD040 because it lacks a language identifier; update the fenced
code block that contains the ASCII diagram (the block starting with the triple
backticks showing "Internal Client ──► Internal Server ... External Server (MQTT
/ QUIC / Dummy)") to include a language label such as text (e.g. change ``` to
```text) so the linter is satisfied while keeping the diagram content unchanged.
In `@CMLibStorage.cmake`:
- Around line 7-8: Replace the plain SET of STORAGE_LIST_DEP_REVISION with a
cached variable so the revision pin propagates to submodules; specifically
change the assignment for STORAGE_LIST_DEP_REVISION to use CMake's CACHE STRING
"" FORCE (keeping the same variable name STORAGE_LIST_DEP_REVISION and the
current value "v2.0.0") so downstream modules that don't declare their own
revision will inherit this pin while being aware FORCE will overwrite local
values.
In
`@include/bringauto/external_client/connection/communication/MqttCommunication.hpp`:
- Line 78: The header MqttCommunication.hpp declares std::atomic<bool>
serverDisconnectPending_ but doesn't include <atomic>, relying on transitive
includes; add a direct include for <atomic> at the top of MqttCommunication.hpp
so the declaration of serverDisconnectPending_ (and any future atomic use) is
well-formed and not fragile.
In `@include/bringauto/settings/Constants.hpp`:
- Around line 217-218: AERON_CONNECTION and SEPARATOR in Constants.hpp duplicate
existing literals; replace their inline constexpr string_view definitions so
they reference the canonical AeronClientConstants values (use
AeronClientConstants::aeron_connection and AeronClientConstants::separator)
instead of hard-coded literals, updating the AERON_CONNECTION and SEPARATOR
symbols accordingly to prevent future divergence.
In `@include/bringauto/structures/StatusAggregatorDeviceState.hpp`:
- Line 107: The class currently stores the mutex as std::unique_ptr<std::mutex>
(externalCommandMutex_) which becomes nullptr after a move, making methods like
consumeCommand and addExternalCommand unsafe on moved-from instances; fix by
either (A) making StatusAggregatorDeviceState non-movable (delete move ctor and
move assignment) so the pointer never gets moved, or (B) replace
std::unique_ptr<std::mutex> externalCommandMutex_ with an owning, non-nullable
mutex (e.g., std::mutex externalCommandMutex_) or a std::shared_ptr<std::mutex>
if shared ownership is needed; update or document invariants accordingly and
ensure consumeCommand/addExternalCommand and any other methods using
externalCommandMutex_ assume it is always valid.
In `@source/bringauto/modules/ModuleHandler.cpp`:
- Around line 17-24: ModuleHandler::destroy() currently drains
fromInternalQueue_ and commandForwardingQueue_ without synchronizing with
threads running handleMessages() or handleCommandForwards(), risking races;
either document the precondition that ioContext.stop() (or equivalent) must be
called and those threads joined before destroy(), or implement explicit
coordination: signal thread shutdown, join the threads that run
handleMessages()/handleCommandForwards(), or protect the queues with a
mutex/condition_variable and ensure no concurrent access before popping in
destroy(); update the method comment to reference the required precondition
(ioContext.stop() and thread joins) if you choose the documentation route.
In `@source/bringauto/settings/QuicSettingsParser.cpp`:
- Around line 52-59: Replace the double-parse pattern around
nlohmann::json::accept(raw) and nlohmann::json::parse(raw) by parsing only once
and handling errors: either call nlohmann::json::parse(raw, nullptr, false) and
check j.is_discarded() to log the same warning via
settings::Logger::logWarning("[quic] QUIC setting '{}' is not valid JSON", key)
and return std::nullopt, or wrap a single nlohmann::json::parse(raw) in a
try/catch for nlohmann::json::parse_error and handle the warning/return there;
update the code that assigns to auto j to use this single-parse approach and
remove the nlohmann::json::accept(raw) usage.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 4a8ac9b7-66a5-491e-951f-7a2d395d2f1f
📒 Files selected for processing (50)
CLAUDE.mdCMLibStorage.cmakeCMakeLists.txtcmake/Dependencies.cmakeinclude/bringauto/common_utils/EnumUtils.hppinclude/bringauto/external_client/ExternalClient.hppinclude/bringauto/external_client/connection/ConnectionState.hppinclude/bringauto/external_client/connection/ExternalConnection.hppinclude/bringauto/external_client/connection/communication/DummyCommunication.hppinclude/bringauto/external_client/connection/communication/ICommunicationChannel.hppinclude/bringauto/external_client/connection/communication/MqttCommunication.hppinclude/bringauto/external_client/connection/communication/QuicCommunication.hppinclude/bringauto/modules/IModuleManagerLibraryHandler.hppinclude/bringauto/modules/ModuleBinaryException.hppinclude/bringauto/modules/ModuleHandler.hppinclude/bringauto/modules/ModuleManagerLibraryHandlerAsync.hppinclude/bringauto/modules/ModuleManagerLibraryHandlerLocal.hppinclude/bringauto/modules/StatusAggregator.hppinclude/bringauto/modules/library_loader.hppinclude/bringauto/settings/Constants.hppinclude/bringauto/settings/QuicSettingsParser.hppinclude/bringauto/structures/ExternalConnectionSettings.hppinclude/bringauto/structures/InternalClientMessage.hppinclude/bringauto/structures/ModuleLibrary.hppinclude/bringauto/structures/StatusAggregatorDeviceState.hppmain.cppresources/config/README.mdresources/config/quic_example.jsonsource/bringauto/common_utils/EnumUtils.cppsource/bringauto/external_client/ExternalClient.cppsource/bringauto/external_client/connection/ExternalConnection.cppsource/bringauto/external_client/connection/communication/MqttCommunication.cppsource/bringauto/external_client/connection/communication/QuicCommunication.cppsource/bringauto/internal_server/InternalServer.cppsource/bringauto/modules/ModuleHandler.cppsource/bringauto/modules/ModuleManagerLibraryHandlerAsync.cppsource/bringauto/modules/ModuleManagerLibraryHandlerLocal.cppsource/bringauto/modules/StatusAggregator.cppsource/bringauto/modules/library_loader_production.cppsource/bringauto/settings/QuicSettingsParser.cppsource/bringauto/settings/SettingsParser.cppsource/bringauto/structures/InternalClientMessage.cppsource/bringauto/structures/ModuleLibrary.cppsource/bringauto/structures/StatusAggregatorDeviceState.cpptest/include/ExternalConnectionTests.hpptest/include/testing_utils/CommunicationMock.hpptest/include/testing_utils/ConfigMock.hpptest/lib/example-moduletest/source/StatusAggregatorTests.cpptest/source/library_loader_test.cpp
🛑 Comments failed to post (7)
resources/config/quic_example.json (1)
25-33:
⚠️ Potential issue | 🟡 Minor | ⚡ Quick win🧩 Analysis chain
🏁 Script executed:
#!/bin/bash # Find the literal QUIC setting keys the channel looks up and compare to the example. rg -nP '(getProtocolSettingsString|protocolSettings\.find|protocolSettings\[)' -C2 \ --iglob '*Quic*' # Also dump any QUIC-related key constants rg -nP '(ca-file|client-cert|client-key|alpn|stream-mode|DisconnectTimeoutMs|PeerUnidiStreamCount|STREAM_MODE|ALPN|DISCONNECT|UNIDI)' -C1 \ --iglob '*Quic*' --iglob '*Constants*'Repository: bringauto/module-gateway
Length of output: 12582
Rework QUIC setting key-name concern (keys match code; only indentation differs)
getProtocolSettingsString()does an exact lookup (settings.protocolSettings.find(std::string(key))), but the example JSON keys forca-file,client-cert,client-key,alpn, andstream-modematchinclude/bringauto/settings/Constants.hpp(CA_FILE,CLIENT_CERT,CLIENT_KEY,ALPN,STREAM_MODE) and are the same keysQuicCommunication.cpprequests.DisconnectTimeoutMsandPeerUnidiStreamCountare read directly bysource/bringauto/settings/QuicSettingsParser.cppusing the same exact strings, not viagetProtocolSettingsString.- The only remaining issue in the snippet is mixed indentation (the
"alpn"line has a leading tab).🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@resources/config/quic_example.json` around lines 25 - 33, The JSON example has inconsistent indentation (a TAB before the "alpn" key) which is the only remaining issue; open the example and replace the leading tab on the "alpn" line with the same spaces used elsewhere so all keys ("ca-file", "client-cert", "client-key", "alpn", "stream-mode", "DisconnectTimeoutMs", "PeerUnidiStreamCount") are consistently indented; no key renaming is needed because getProtocolSettingsString(), Constants.hpp (CA_FILE, CLIENT_CERT, CLIENT_KEY, ALPN, STREAM_MODE) and QuicSettingsParser.cpp already expect the shown key names.resources/config/README.md (1)
42-48:
⚠️ Potential issue | 🟡 Minor | ⚡ Quick winDocument the remaining
quic-settingskeys used in the example.
quic_example.jsonalso setsstream-mode,DisconnectTimeoutMs, andPeerUnidiStreamCount, but these are not described here. Please add them (with allowed values/defaults) so the schema doc matches the shipped example.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@resources/config/README.md` around lines 42 - 48, Update the quic-settings section in README.md to document the additional keys used in quic_example.json: add entries for "stream-mode" (describe allowed values e.g., "bidirectional" or "unidirectional" and default), "DisconnectTimeoutMs" (describe it's an integer timeout in milliseconds and default), and "PeerUnidiStreamCount" (describe it's an integer limiting peer-initiated unidirectional streams and default); ensure each entry mirrors the style of the existing keys (key name, purpose, type, allowed values, and default) so the schema doc matches the shipped example.source/bringauto/external_client/connection/communication/QuicCommunication.cpp (3)
40-73:
⚠️ Potential issue | 🟠 Major | ⚡ Quick winConnection state not rolled back on failure.
If
ConnectionOpenorConnectionStartfails,connectionState_remainsCONNECTINGinstead of being rolled back toNOT_CONNECTED. This prevents future connection attempts sincecompare_exchange_strongwill fail when expected state isn'tNOT_CONNECTED.Also, line 72 redundantly sets state to
CONNECTINGwhen it's already in that state from line 48.Proposed fix
QUIC_STATUS status = quic_->ConnectionOpen(registration_, connectionCallback, this, &connection_); if (QUIC_FAILED(status)) { settings::Logger::logError("ConnectionOpen failed (status=0x{:x})", status); + connectionState_.store(ConnectionState::NOT_CONNECTED); return; } status = quic_->ConnectionStart( connection_, config_, QUIC_ADDRESS_FAMILY_INET, settings_.serverIp.c_str(), settings_.port ); if (QUIC_FAILED(status)) { settings::Logger::logError("ConnectionStart failed (status=0x{:x})", status); + quic_->ConnectionClose(connection_); + connection_ = nullptr; + connectionState_.store(ConnectionState::NOT_CONNECTED); return; } - - connectionState_ = ConnectionState::CONNECTING; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/external_client/connection/communication/QuicCommunication.cpp` around lines 40 - 73, In QuicCommunication::initializeConnection, ensure connectionState_ is rolled back to ConnectionState::NOT_CONNECTED if quic_->ConnectionOpen or quic_->ConnectionStart fails; after a failed QUIC call (QUIC_FAILED(status)) set connectionState_ back to NOT_CONNECTED (use the same atomic semantics as the initial compare_exchange_strong or an atomic store) and return, and remove the redundant connectionState_ = ConnectionState::CONNECTING at the end since the state was already set by compare_exchange_strong when entering the function; reference symbols: initializeConnection, connectionState_, ConnectionOpen, ConnectionStart, compare_exchange_strong, QUIC_FAILED.
231-241:
⚠️ Potential issue | 🔴 Critical | ⚡ Quick winUse-after-free:
stop()closes MsQuic before async shutdown callback fires.
stop()callscloseConnection()(which triggers asyncConnectionShutdown), then immediately proceeds tocloseMsQuic(). WhenQUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETEfires later inconnectionCallback(line 351), it callsself->quic_->ConnectionClose(connection), butquic_has already been nullified bycloseMsQuic().Proposed fix - wait for shutdown completion before closing resources
Add a condition variable to wait for shutdown completion:
+// In header, add: +std::condition_variable shutdownCv_; +std::mutex shutdownMutex_; +std::atomic<bool> shutdownComplete_{false}; void QuicCommunication::stop() { - QuicCommunication::closeConnection(); + if (connection_) { + quic_->ConnectionShutdown(connection_, QUIC_CONNECTION_SHUTDOWN_FLAG_NONE, 0); + // Wait for SHUTDOWN_COMPLETE callback + std::unique_lock lock(shutdownMutex_); + shutdownCv_.wait_for(lock, std::chrono::seconds(5), [this] { + return shutdownComplete_.load(); + }); + } closeConfiguration(); closeRegistration(); closeMsQuic(); inboundCv_.notify_all(); outboundCv_.notify_all(); settings::Logger::logInfo("[quic] Connection stopped"); } // In connectionCallback SHUTDOWN_COMPLETE case: case QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE: { settings::Logger::logInfo("[quic] Connection shutdown complete"); self->connectionState_ = ConnectionState::NOT_CONNECTED; self->outboundCv_.notify_all(); if (self->senderThread_.joinable()) { self->senderThread_.request_stop(); } self->quic_->ConnectionClose(connection); self->connection_ = nullptr; + { + std::lock_guard lock(self->shutdownMutex_); + self->shutdownComplete_.store(true); + } + self->shutdownCv_.notify_all(); break; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/external_client/connection/communication/QuicCommunication.cpp` around lines 231 - 241, QuicCommunication::stop currently calls closeConnection() (which initiates an async ConnectionShutdown) then immediately calls closeMsQuic(), causing a use-after-free when QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE runs in connectionCallback and uses quic_. Add a shutdown-complete flag (e.g., shutdown_complete_) and a condition_variable (e.g., shutdownCv_) plus mutex; set shutdown_complete_ and notify in connectionCallback when handling QUIC_CONNECTION_EVENT_SHUTDOWN_COMPLETE, and have QuicCommunication::stop wait (with a sensible timeout) on shutdownCv_ until shutdown_complete_ is true before calling closeMsQuic(); also add defensive checks in connectionCallback to ensure quic_ is valid before calling quic_->ConnectionClose(connection).
261-273:
⚠️ Potential issue | 🟠 Major | ⚡ Quick winStream handle leaked on serialization failure.
If
StreamOpensucceeds butSerializeToArrayfails, the opened stream is never closed, leaking the QUIC stream resource.Proposed fix
if (!message.SerializeToArray(sendBuffer->storage.data(), static_cast<int>(size))) { settings::Logger::logError("[quic] Message serialization failed"); + quic_->StreamClose(stream); return; }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.if (QUIC_FAILED( quic_->StreamOpen(connection_, flags, streamCallback, this, &stream))) { settings::Logger::logError("[quic] StreamOpen failed"); return; } const size_t size = message.ByteSizeLong(); auto sendBuffer = std::make_unique<SendBuffer>(size); if (!message.SerializeToArray(sendBuffer->storage.data(), static_cast<int>(size))) { settings::Logger::logError("[quic] Message serialization failed"); quic_->StreamClose(stream); return; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/external_client/connection/communication/QuicCommunication.cpp` around lines 261 - 273, Stream opened by quic_->StreamOpen is leaked if message.SerializeToArray fails; ensure you close the opened stream before returning on serialization failure. After quic_->StreamOpen(...) succeeds and before returning on a failed SerializeToArray call, call the appropriate close API (e.g. quic_->StreamClose(stream) or the library equivalent) to release the stream handle (reference the local variable stream and the call sites quic_->StreamOpen and message.SerializeToArray), or alternatively move creation/serialization logic so the stream is only opened after successful serialization.source/bringauto/modules/ModuleManagerLibraryHandlerAsync.cpp (1)
70-72:
⚠️ Potential issue | 🟠 Major | ⚡ Quick winUnchecked
.value()may throw on timeout/failure.
callFuncreturnsstd::optional, and calling.value()without checkinghas_value()will throwstd::bad_optional_accessif the async call times out or fails. The same pattern appears inisDeviceTypeSupported,sendStatusCondition,generateCommand,aggregateStatus,aggregateError,generateFirstCommand,statusDataValid, andcommandDataValid.Consider returning an error code or handling the empty optional gracefully, similar to how
loadLibrarypolls withhas_value().Example guard pattern
int ModuleManagerLibraryHandlerAsync::getModuleNumber() const { - return aeronClient.callFunc(fp_async::getModuleNumberAsync).value(); + auto result = aeronClient.callFunc(fp_async::getModuleNumberAsync); + if (!result.has_value()) { + // Handle timeout/failure - return sentinel or throw a documented exception + throw std::runtime_error("Aeron call timed out for getModuleNumber"); + } + return result.value(); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/modules/ModuleManagerLibraryHandlerAsync.cpp` around lines 70 - 72, Multiple methods (getModuleNumber, isDeviceTypeSupported, sendStatusCondition, generateCommand, aggregateStatus, aggregateError, generateFirstCommand, statusDataValid, commandDataValid) call aeronClient.callFunc(...).value() which will throw std::bad_optional_access on timeout/failure; update each to check the returned std::optional with has_value() (or use operator bool) before accessing the value and handle the empty case gracefully (e.g., return a sensible error code/optional/false, or propagate an error) instead of calling .value(); follow the same polling/guard pattern used in loadLibrary to avoid exceptions and ensure consistent error handling for the fp_async::* calls.source/bringauto/settings/QuicSettingsParser.cpp (1)
26-36:
⚠️ Potential issue | 🟡 Minor | ⚡ Quick win🧩 Analysis chain
🌐 Web query:
msquic QUIC_SETTINGS struct field types DisconnectTimeoutMs PeerUnidiStreamCount IdleTimeoutMs widths💡 Result:
In the MsQuic library, the QUIC_SETTINGS structure is used to configure various connection and library parameters [1][2]. The field types and widths for the specified fields are as follows: - DisconnectTimeoutMs: This field is defined as a uint32_t [1][3]. - PeerUnidiStreamCount: This field is defined as a uint16_t [1][3]. - IdleTimeoutMs: This field is defined as a uint64_t [3]. The QUIC_SETTINGS structure utilizes a union of flags and a structure of individual fields to manage these settings. Each configurable parameter has a corresponding bit-flag within the structure to indicate if it has been explicitly set by the user [4][2][5].
Citations:
- 1: https://github.com/microsoft/msquic/blob/main/docs/Settings.md
- 2: https://github.com/microsoft/msquic/blob/main/docs/api/QUIC_SETTINGS.md
- 3: https://microsoft.github.io/msquic/msquicdocs/docs/Settings.html
- 4: https://microsoft.github.io/msquic/msquicdocs/docs/api/QUIC_SETTINGS.html
- 5: https://raw.githubusercontent.com/microsoft/msquic/main/src/inc/msquic.h
🏁 Script executed:
#!/bin/bash set -euo pipefail # Find the helper and where it's used rg -n "getOptionalUint\\b" -S . rg -n "DisconnectTimeoutMs\\b|PeerUnidiStreamCount\\b" -S source/bringauto/settings/QuicSettingsParser.cpp source/bringauto -S # Show the QuicSettingsParser.cpp area around the snippet sed -n '1,120p' source/bringauto/settings/QuicSettingsParser.cpp # Print the definition of getOptionalUint (first match only) file="$(rg -l "getOptionalUint\\b" -S . | head -n 1)" echo "=== getOptionalUint file: $file ===" nl -ba "$file" | sed -n '1,200p'Repository: bringauto/module-gateway
Length of output: 5016
🏁 Script executed:
#!/bin/bash set -euo pipefail # Find all mentions of the QUIC_SETTINGS fields in the repo rg -n "DisconnectTimeoutMs|PeerUnidiStreamCount|QUIC_SETTINGS" -S include source | head -n 200 # Locate where QUIC_SETTINGS is introduced/typedef'ed/included rg -n "msquic|QUIC_SETTINGS|QUIC_API|msquic\\.h|msquicdocs|inc/msquic\\.h" -S include source | head -n 200 # Print the header that declares QUIC_SETTINGS (if it's vendored) rg -n "typedef\\s+struct\\s+QUIC_SETTINGS|struct\\s+QUIC_SETTINGS" -S . | head -n 50 # Also inspect the QuicSettingsParser.hpp for QUIC_SETTINGS type origin sed -n '1,200p' include/bringauto/settings/QuicSettingsParser.hppRepository: bringauto/module-gateway
Length of output: 4418
🏁 Script executed:
#!/bin/bash set -euo pipefail # Inspect the header for any helpers/overloads sed -n '1,120p' include/bringauto/settings/QuicSettingsParser.hpp # Look for any range validation around these keys rg -n "DisconnectTimeoutMs|PeerUnidiStreamCount" -S source include # Look for any generic numeric clamping/validation helpers used by QuicSettingsParser rg -n "range|clamp|min|max|bounds|out of range" -S source/bringauto/settings # Inspect SettingsParser.cpp nearby to see if protocolSettings performs validation sed -n '150,340p' source/bringauto/settings/SettingsParser.cppRepository: bringauto/module-gateway
Length of output: 8668
Prevent silent truncation when parsing QUIC settings.
getOptionalUintreturnsstd::optional<uint64_t>and values are assigned directly intoQUIC_SETTINGS::DisconnectTimeoutMs(uint32_t) andQUIC_SETTINGS::PeerUnidiStreamCount(uint16_t); values exceeding those widths will truncate/wrap silently. Add explicit range checks (log + skip/reject) before assignment.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@source/bringauto/settings/QuicSettingsParser.cpp` around lines 26 - 36, The code currently assigns values from getOptionalUint (std::optional<uint64_t>) directly into QUIC_SETTINGS fields (quic.DisconnectTimeoutMs:uint32_t and quic.PeerUnidiStreamCount:uint16_t) which can silently truncate; add explicit range checks before assignment: after obtaining value in the if (auto value = getOptionalUint(...)) blocks, verify the uint64_t fits into the target width (<= UINT32_MAX for DisconnectTimeoutMs, <= UINT16_MAX for PeerUnidiStreamCount), log an error/warning via settings::Logger::logDebug or logError if out of range, and skip the assignment and quic.IsSet.* flag when out of range; only assign and set quic.IsSet.DisconnectTimeoutMs / quic.IsSet.PeerUnidiStreamCount when the value passes the range check.
danprudky
force-pushed
the
BAF-1204/Teleoperation-Control
branch
3 times, most recently
from
605381e to
a5eb233
Compare
…dQueue_ - getCommand and addExternalCommand can be called from different threads; the queue was unprotected - added externalCommandMutex_ as a unique_ptr<mutex> (keeps the type movable) and a lock_guard in both methods
- Add optional forward_command_on_receive(device_type) to IModuleManagerLibraryHandler; modules returning OK opt into immediate dispatch, others keep status-triggered behaviour - ModuleManagerLibraryHandlerLocal loads the symbol via checkOptionalFunction (no throw if absent); Async handler always returns NOT_OK - StatusAggregatorDeviceState caches the forwarding flag at device registration; update_command() returns new FORWARD_IMMEDIATELY code when the flag is set - ExternalClient detects FORWARD_IMMEDIATELY and pushes a makeCommandForward event to a dedicated commandForwardingQueue separate from fromInternalQueue - ModuleHandler spawns a second jthread (handleCommandForwards) that drains commandForwardingQueue independently of the status-driven handleMessages loop - StatusAggregator gains a recursive_mutex protecting devices and deviceTimeouts_ against concurrent access from both ModuleHandler threads and ExternalClient
…alClient - Remove FORWARD_IMMEDIATELY return code from StatusAggregator::update_command; the forwarding flag no longer leaks module policy into the aggregator layer - ExternalClient::handleCommand now calls forwardCommandOnReceive() directly after a successful update_command to decide whether to push to commandForwardingQueue, keeping the dispatch logic co-located with the queue it writes to
- Previous condition bytesTransferred < bufferOffset was logically inverted; replace with bufferOffset + bytesTransferred > buffer.size() to catch actual overflows - Fix recursive call offset: pass bufferOffset + (bytesTransferred - bytesLeft) instead of bytesTransferred - bytesLeft
…nd to consumeCommand - rename getCommand() → consumeCommand() to signal side-effectful dequeue semantics, distinguishing it from pure const getters like getStatus() - return std::nullopt from consumeCommand() for push-only devices with empty external command queue instead of falling through to default command - handle NO_MESSAGE_AVAILABLE in handleCommandForward: log at debug level when command was already consumed by the status path, avoiding a spurious warning - send empty DeviceCommand in handleStatus when push-only device has no pending command, satisfying the InternalProtocol 250 ms response requirement without forwarding to the vehicle - use *currentCommand instead of .value() after explicit has_value() guard in get_command - remove redundant #include <optional> from StatusAggregatorDeviceState.cpp
…ator - introduce private unlocked helpers (isDeviceValidUnlocked, clearDeviceUnlocked, forceAggregationOnDeviceUnlocked, getCommandUnlocked) that operate under an assumed held lock - update all callers that already hold devicesMutex_ to use the unlocked variants, eliminating recursive lock acquisition - replace mutable std::recursive_mutex devicesMutex_ with std::mutex - replace std::unique_ptr<std::mutex> externalCommandMutex_ in StatusAggregatorDeviceState with a plain std::mutex member - switch devices.emplace() to try_emplace() to construct StatusAggregatorDeviceState in-place, avoiding a move constructor that std::mutex would have deleted
- introduce FunctionPtr = void(*)() alias to express that checkFunction and checkOptionalFunction return function pointers, not data pointers
…_cast Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
danprudky
force-pushed
the
BAF-1204/Teleoperation-Control
branch
from
a5eb233 to
109c8c0
Compare
|
jiriskuta
reviewed
Jun 3, 2026
| @@ -98,10 +108,16 @@ int StatusAggregator::add_status_to_aggregator(const Buffer& status, | |||
| deviceTimeouts_[deviceId]++; | |||
Contributor
There was a problem hiding this comment.
Data race on deviceTimeouts_ — the new devicesMutex_ invariant is only half-applied (UB).
devicesMutex_ is documented as protecting deviceTimeouts_, and every accessor now locks it — e.g. getDeviceTimeoutCount() reads deviceTimeouts_.find() under the lock, and add_status_to_aggregator() writes deviceTimeouts_[device] = 0 under the lock.
But this increment runs without the lock. The timeouted_force_aggregation lambda is stored in each device's ThreadTimer and invoked from ThreadTimer::tick on the io_context threads (two threads run ioContext.run() in main.cpp), while getDeviceTimeoutCount() runs on the ModuleHandler thread. So deviceTimeouts_[deviceId]++ here is an unsynchronized read-modify-write racing the locked reads/writes on the same map element → data race / UB.
It is not holding the lock at this point either: the very next call into force_aggregation_on_device() re-locks the (non-recursive) devicesMutex_, which would self-deadlock on every tick if the lambda already held it — confirming the increment runs unlocked.
Note the atomic timeoutedMessageReady_ store does not rescue this: it is sequenced before the increment, so its release edge never covers the ++.
Fix: take devicesMutex_ around the increment (or fold it into the already-locked force_aggregation_on_device). A concurrent locked insert of a different device can also rehash the map under the lambda's unlocked access, so the lock needs to cover this path.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Adds an optional
forward_command_on_receive(device_type)API that modules canexport
to opt in to immediate command dispatch — bypassing the up-to-40 ms wait for the
next
device status. Required for teleoperation, where the rest of the pipeline is already
event-driven.
How it works
forward_command_on_receive.Absent symbol or
NOT_OKreturn = existing pull behaviour.OK= push mode.StatusAggregator::add_status_to_aggregatorcalls the functiononce
and caches the result in
StatusAggregatorDeviceState::forwardCommandImmediately_.InternalClientMessage::makeCommandForward(deviceId)event intocommandForwardingQueue_(a new queue shared with ModuleHandler).
handleCommandForwards()thread that drainscommandForwardingQueue_and callsget_command_for_forwarding()(cached statusas input),
then pushes the result straight to
toInternalQueue_.consumeCommand()(renamed fromgetCommand()) returnsstd::nulloptwhen a push-only device has no pending command.handleStatussends an
empty
DeviceCommandto satisfy the InternalProtocol 250 ms response requirement;the
bridge detects empty
commanddataand does not forward it to the vehicle.Thread safety
StatusAggregator::devicesMutex_(recursive_mutex) serialises all access todevicesand
deviceTimeouts_between the ModuleHandler main thread, the new forwardthread, and
ExternalClient.
StatusAggregatorDeviceState::externalCommandMutex_protectsexternalCommandQueue_against concurrent
addExternalCommand/consumeCommandcalls.Also included
InternalServer::processBufferDatabounds check: `bufferOffset +bytesTransferred
Summary by CodeRabbit
New Features
CLOSINGstateDocumentation
Chores
BRINGAUTO_SANITIZERSbuild option for flexible sanitizer configuration