From c7e48f925718d3ad27c055b0c779a02ae4f0c793 Mon Sep 17 00:00:00 2001 From: EHotwagner Date: Thu, 1 Oct 2026 11:06:16 +0200 Subject: [PATCH 1/2] Add bounded stock queue native trace --- CMakeLists.txt | 17 +- src/circuit/grpc/CommandDispatch.cpp | 67 ++++-- src/circuit/grpc/CommandDispatch.h | 7 + src/circuit/grpc/StockQueueReader.cpp | 9 +- src/circuit/grpc/StockQueueReader.h | 1 + src/circuit/grpc/StockQueueTrace.cpp | 290 +++++++++++++++++++++++ src/circuit/grpc/StockQueueTrace.h | 72 ++++++ src/circuit/module/GrpcGatewayModule.cpp | 52 ++++ src/circuit/module/GrpcGatewayModule.h | 2 + tests/unit/stock_queue_trace_test.cc | 134 +++++++++++ 10 files changed, 630 insertions(+), 21 deletions(-) create mode 100644 src/circuit/grpc/StockQueueTrace.cpp create mode 100644 src/circuit/grpc/StockQueueTrace.h create mode 100644 tests/unit/stock_queue_trace_test.cc diff --git a/CMakeLists.txt b/CMakeLists.txt index ca6c504f..009d5301 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -52,7 +52,8 @@ if (CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR) "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/CommandQueue.cpp" "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/LiveControlState.cpp" "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/Counters.cpp" - "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/StockQueueReader.cpp") + "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/StockQueueReader.cpp" + "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/StockQueueTrace.cpp") # The tactical helper lands in the native implementation lane. Keep this # contract checkpoint independently buildable while registering the helper # as soon as the coherent join supplies it. @@ -119,6 +120,16 @@ if (CMAKE_SOURCE_DIR STREQUAL CMAKE_CURRENT_SOURCE_DIR) add_test(NAME stock_queue_reader_test COMMAND stock_queue_reader_test) endif () + if (EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/tests/unit/stock_queue_trace_test.cc") + add_executable(stock_queue_trace_test + "${CMAKE_CURRENT_SOURCE_DIR}/tests/unit/stock_queue_trace_test.cc") + target_include_directories(stock_queue_trace_test PRIVATE + "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit") + target_link_libraries(stock_queue_trace_test PRIVATE + highbar_command_queue_test_support GTest::gtest GTest::gtest_main) + add_test(NAME stock_queue_trace_test COMMAND stock_queue_trace_test) + endif () + if (EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/tests/unit/tactical_feature_capacity_test.cc") add_executable(tactical_feature_capacity_test "${CMAKE_CURRENT_SOURCE_DIR}/tests/unit/tactical_feature_capacity_test.cc") @@ -391,7 +402,8 @@ constexpr const char* kEngineSha256 = "@HIGHBAR_ENGINE_SHA256@"; "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/CommandQueue.cpp" "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/LiveControlState.cpp" "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/Counters.cpp" - "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/StockQueueReader.cpp") + "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/StockQueueReader.cpp" + "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/StockQueueTrace.cpp") if (EXISTS "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/TacticalNativeState.cpp") target_sources(highbar_command_queue_test_support PRIVATE "${CMAKE_CURRENT_SOURCE_DIR}/src/circuit/grpc/TacticalNativeState.cpp") @@ -484,6 +496,7 @@ constexpr const char* kEngineSha256 = "@HIGHBAR_ENGINE_SHA256@"; OR _test_name STREQUAL "live_control_state_test" OR _test_name STREQUAL "tactical_native_state_test" OR _test_name STREQUAL "stock_queue_reader_test" + OR _test_name STREQUAL "stock_queue_trace_test" OR _test_name STREQUAL "tactical_feature_capacity_test") target_link_libraries("${_test_name}" PRIVATE highbar_command_queue_test_support) diff --git a/src/circuit/grpc/CommandDispatch.cpp b/src/circuit/grpc/CommandDispatch.cpp index f838a3ce..03ad81db 100644 --- a/src/circuit/grpc/CommandDispatch.cpp +++ b/src/circuit/grpc/CommandDispatch.cpp @@ -6,6 +6,7 @@ #include "grpc/FactoryProductionPolicy.h" #include "grpc/GrpcLog.h" #include "grpc/StockQueueReader.h" +#include "grpc/StockQueueTrace.h" #include "grpc/TacticalNativeState.h" #include "CircuitAI.h" @@ -68,7 +69,13 @@ springai::AIFloat3 ToFloat3(const ::highbar::v1::Vector3& v) { return springai::AIFloat3(v.x(), v.y(), v.z()); } -std::optional> ReadCurrentTacticalQueue( +struct TacticalQueueRead { + std::vector entries; + std::optional stock; + bool complete = false; +}; + +std::optional ReadCurrentTacticalQueue( ::circuit::CCircuitAI* ai, springai::Unit* unit, ::highbar::v1::NativeQueueDomain domain) { if (ai == nullptr || unit == nullptr) return std::nullopt; @@ -84,33 +91,35 @@ std::optional> ReadCurrentTacticalQueue( [&](const char* data, std::size_t size) { return lua->CallRules(data, static_cast(size)); }); - if (read.status != StockQueueReadStatus::Complete) return std::nullopt; - std::vector entries; - entries.reserve(read.entries.size()); + TacticalQueueRead result; + result.stock = read; + if (read.status != StockQueueReadStatus::Complete) return result; + result.complete = true; + result.entries.reserve(read.entries.size()); for (const auto& entry : read.entries) { - entries.push_back({0, entry.command_id, entry.options, + result.entries.push_back({0, entry.command_id, entry.options, entry.tag, 0, entry.params}); } - return entries; + return result; } if (domain != ::highbar::v1::NATIVE_QUEUE_DOMAIN_ACTOR_ORDER) { return std::nullopt; } #endif - auto materialize = [](auto& commands) - -> std::optional> { + auto materialize = [](auto& commands) -> std::optional { if (commands.size() > kMaximumTacticalQueueEntries) { utils::free_clear(commands); return std::nullopt; } - std::vector entries; - entries.reserve(commands.size()); + TacticalQueueRead result; + result.complete = true; + result.entries.reserve(commands.size()); for (auto* command : commands) { if (command == nullptr) { utils::free_clear(commands); return std::nullopt; } - entries.push_back({command->GetType(), command->GetId(), + result.entries.push_back({command->GetType(), command->GetId(), static_cast(command->GetOptions()), command->GetTag(), #ifdef HIGHBAR_STOCK_RECOIL @@ -121,7 +130,7 @@ std::optional> ReadCurrentTacticalQueue( command->GetParams()}); } utils::free_clear(commands); - return entries; + return result; }; try { @@ -159,6 +168,11 @@ bool DispatchTacticalCommand(::circuit::CCircuitAI* ai, const ::highbar::v1::NativeTacticalCommand& command, const FeatureLifetimeLedger* feature_lifetimes, const StockQueueRevisionContext* stock_queue_context, + StockQueueTraceSink* stock_queue_trace, + const ::highbar::v1::NativeObservationBasis* live_basis, + const StockQueueDispatchIdentity* dispatch_identity, + const std::function& current_frame, + std::uint32_t perspective_team_id, const std::function& post_control_fence, TacticalDispatchRefusalReason* refusal_reason) { auto refuse = [&](TacticalDispatchRefusalReason reason) { @@ -173,7 +187,7 @@ bool DispatchTacticalCommand(::circuit::CCircuitAI* ai, } auto* native = unit->GetUnit(); const auto current_queue = ReadCurrentTacticalQueue(ai, native, command.queue_domain()); - if (!current_queue.has_value()) { + if (!current_queue.has_value() || !current_queue->complete) { return refuse(TacticalDispatchRefusalReason::kQueueUnavailable); } std::optional required_tag; @@ -204,7 +218,7 @@ bool DispatchTacticalCommand(::circuit::CCircuitAI* ai, return NativeQueueMatchesExpected(queue,command.expected_queue_revision(),required_tag); #endif }; - if (!matches_expected(*current_queue)) { + if (!matches_expected(current_queue->entries)) { return refuse(TacticalDispatchRefusalReason::kQueueChanged); } auto friendly = [&](const ::highbar::v1::NativeUnitReference& ref) { @@ -225,7 +239,26 @@ bool DispatchTacticalCommand(::circuit::CCircuitAI* ai, [&] { const auto final_queue = ReadCurrentTacticalQueue( ai, native, command.queue_domain()); - return final_queue.has_value() && matches_expected(*final_queue); + if (!final_queue.has_value()) return false; + const bool matched = final_queue->complete + && matches_expected(final_queue->entries); + #ifdef HIGHBAR_STOCK_RECOIL + if (stock_queue_trace != nullptr && stock_queue_context != nullptr + && live_basis != nullptr && dispatch_identity != nullptr + && final_queue->stock.has_value()) { + auto identity = *dispatch_identity; identity.matched = matched; + std::uint64_t revision = 0; + if (final_queue->stock->status == StockQueueReadStatus::Complete) { + revision = ComputeStockQueueRevision( + *stock_queue_context, final_queue->stock->entries); + } + stock_queue_trace->Record({true, + current_frame ? current_frame() : 0u, perspective_team_id, + *live_basis, *stock_queue_context, *final_queue->stock, + revision, identity}); + } + #endif + return matched; }, [&] { return !post_control_fence || post_control_fence(); }, std::forward(effect)); @@ -279,7 +312,7 @@ bool DispatchTacticalCommand(::circuit::CCircuitAI* ai, : body.queue_policy() == ::highbar::v1::NATIVE_QUEUE_POLICY_REJECT_IF_BUSY ? StockFactoryQueuePolicy::RejectIfBusy : StockFactoryQueuePolicy::Replace; - if (!StockFactoryProductionPolicyAllows(policy, current_queue->empty())) { + if (!StockFactoryProductionPolicyAllows(policy, current_queue->entries.empty())) { return false; } return dispatch_under_external_control([&] { @@ -288,7 +321,7 @@ bool DispatchTacticalCommand(::circuit::CCircuitAI* ai, }); #else return DispatchFactoryProductionIfAllowed( - body.queue_policy(), current_queue->empty(), [&] { + body.queue_policy(), current_queue->entries.empty(), [&] { return dispatch_under_external_control([&] { unit->CmdBuild(def, native->GetPos(), UNIT_NO_FACING, 0); }); diff --git a/src/circuit/grpc/CommandDispatch.h b/src/circuit/grpc/CommandDispatch.h index ea4001b2..85baa4b1 100644 --- a/src/circuit/grpc/CommandDispatch.h +++ b/src/circuit/grpc/CommandDispatch.h @@ -35,7 +35,9 @@ class CEnemyInfo; namespace circuit::grpc { class FeatureLifetimeLedger; +class StockQueueTraceSink; struct StockQueueRevisionContext; +struct StockQueueDispatchIdentity; constexpr int EngineFacingForNativeBuild( ::highbar::v1::NativeBuildFacing facing) { @@ -115,6 +117,11 @@ bool DispatchTacticalCommand(::circuit::CCircuitAI* ai, const ::highbar::v1::NativeTacticalCommand& command, const FeatureLifetimeLedger* feature_lifetimes, const StockQueueRevisionContext* stock_queue_context, + StockQueueTraceSink* stock_queue_trace, + const ::highbar::v1::NativeObservationBasis* live_basis, + const StockQueueDispatchIdentity* dispatch_identity, + const std::function& current_frame, + std::uint32_t perspective_team_id, const std::function& post_control_fence, TacticalDispatchRefusalReason* refusal_reason = nullptr); diff --git a/src/circuit/grpc/StockQueueReader.cpp b/src/circuit/grpc/StockQueueReader.cpp index 92a7a338..0a416d93 100644 --- a/src/circuit/grpc/StockQueueReader.cpp +++ b/src/circuit/grpc/StockQueueReader.cpp @@ -181,6 +181,7 @@ StockQueueReadResult ParseStockQueueResponse(const std::string& response, StockQueueReadResult result; result.domain = expected_domain; result.unit_id = expected_unit_id; + result.canonical_request = exact_request; if (response.empty() || response.size() > kStockQueueMaximumBytes || !SevenBitAsciiWithoutNul(response) || response.back() != '\n') return result; const auto lines = Lines(response); @@ -260,10 +261,14 @@ StockQueueReadResult ReadStockQueue(StockQueueDomain domain,std::int32_t unit_id const auto request = BuildStockQueueRequest(domain, unit_id); if (request.empty()) return {}; try { - return ParseStockQueueResponse(call_rules(request.data(), request.size()), + auto result = ParseStockQueueResponse(call_rules(request.data(), request.size()), request, domain, unit_id); + result.domain = domain; result.unit_id = unit_id; + result.canonical_request = request; + return result; } catch (...) { - return {}; + StockQueueReadResult result; result.domain = domain; result.unit_id = unit_id; + result.canonical_request = request; return result; } } diff --git a/src/circuit/grpc/StockQueueReader.h b/src/circuit/grpc/StockQueueReader.h index 9af5d9f7..761f327b 100644 --- a/src/circuit/grpc/StockQueueReader.h +++ b/src/circuit/grpc/StockQueueReader.h @@ -41,6 +41,7 @@ struct StockQueueReadResult { StockQueueDomain domain = StockQueueDomain::Production; std::int32_t unit_id = -1; std::string unavailable_reason; + std::string canonical_request; std::string canonical_response; std::vector entries; }; diff --git a/src/circuit/grpc/StockQueueTrace.cpp b/src/circuit/grpc/StockQueueTrace.cpp new file mode 100644 index 00000000..5ad718d4 --- /dev/null +++ b/src/circuit/grpc/StockQueueTrace.cpp @@ -0,0 +1,290 @@ +// SPDX-License-Identifier: GPL-2.0-only +#include "grpc/StockQueueTrace.h" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace circuit::grpc { +namespace { + +constexpr std::size_t kMaximumTraceLineBytes = 32768; +constexpr std::uint64_t kMaximumTraceRecords = 4096; +constexpr std::uint64_t kMaximumTraceBytes = 16 * 1024 * 1024; + +bool SafeRunId(std::string_view value) { + if (value.empty() || value.size() > 64) return false; + for (const unsigned char c : value) { + if (!(c >= 'A' && c <= 'Z') && !(c >= 'a' && c <= 'z') + && !(c >= '0' && c <= '9') && c != '.' && c != '_' && c != '-') return false; + } + return true; +} + +bool Utf8(const std::string& value) { + for (std::size_t i = 0; i < value.size();) { + const auto first = static_cast(value[i]); + if (first == 0) return false; + if (first < 0x80) { ++i; continue; } + std::size_t length = 0; std::uint32_t code = 0; + if ((first & 0xe0) == 0xc0) { length = 2; code = first & 0x1f; } + else if ((first & 0xf0) == 0xe0) { length = 3; code = first & 0x0f; } + else if ((first & 0xf8) == 0xf0) { length = 4; code = first & 0x07; } + else return false; + if (i + length > value.size()) return false; + for (std::size_t j = 1; j < length; ++j) { + const auto next = static_cast(value[i + j]); + if ((next & 0xc0) != 0x80) return false; + code = (code << 6) | (next & 0x3f); + } + if ((length == 2 && code < 0x80) || (length == 3 && code < 0x800) + || (length == 4 && code < 0x10000) || code > 0x10ffff + || (code >= 0xd800 && code <= 0xdfff)) return false; + i += length; + } + return true; +} + +bool BoundedText(const std::string& value) { + if (value.empty() || value.size() > 256) return false; + return Utf8(value); +} + +std::string Json(const std::string& value) { + std::ostringstream out; + out << '"'; + for (const unsigned char c : value) { + switch (c) { + case '"': out << "\\\""; break; + case '\\': out << "\\\\"; break; + case '\b': out << "\\b"; break; + case '\f': out << "\\f"; break; + case '\n': out << "\\n"; break; + case '\r': out << "\\r"; break; + case '\t': out << "\\t"; break; + default: + if (c < 0x20) out << "\\u00" << std::hex << std::setw(2) + << std::setfill('0') << static_cast(c) << std::dec; + else out << static_cast(c); + } + } + out << '"'; + return out.str(); +} + +std::string Hex(const std::string& bytes) { + static constexpr char kHex[] = "0123456789abcdef"; + std::string out; + out.reserve(bytes.size() * 2); + for (const unsigned char c : bytes) { out.push_back(kHex[c >> 4]); out.push_back(kHex[c & 15]); } + return out; +} + +std::string FloatBits(float value) { + std::ostringstream out; + out << std::hex << std::setw(8) << std::setfill('0') << std::nouppercase + << std::bit_cast(value); + return out.str(); +} + +std::string Base64(const std::string& bytes) { + static constexpr char kAlphabet[] = + "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"; + std::string out; + for (std::size_t i = 0; i < bytes.size(); i += 3) { + const std::uint32_t a = static_cast(bytes[i]); + const std::uint32_t b = i + 1 < bytes.size() ? static_cast(bytes[i + 1]) : 0; + const std::uint32_t c = i + 2 < bytes.size() ? static_cast(bytes[i + 2]) : 0; + const auto value = (a << 16) | (b << 8) | c; + out.push_back(kAlphabet[(value >> 18) & 63]); out.push_back(kAlphabet[(value >> 12) & 63]); + out.push_back(i + 1 < bytes.size() ? kAlphabet[(value >> 6) & 63] : '='); + out.push_back(i + 2 < bytes.size() ? kAlphabet[value & 63] : '='); + } + return out; +} + +const char* Status(StockQueueReadStatus status) { + return status == StockQueueReadStatus::Complete ? "complete" + : status == StockQueueReadStatus::Unavailable ? "unavailable" : "malformed"; +} + +bool SameEntries(const std::vector& first, + const std::vector& second) { + if (first.size() != second.size()) return false; + for (std::size_t i = 0; i < first.size(); ++i) { + if (first[i].command_id != second[i].command_id + || first[i].options != second[i].options || first[i].tag != second[i].tag + || first[i].params.size() != second[i].params.size()) return false; + for (std::size_t j = 0; j < first[i].params.size(); ++j) { + if (std::bit_cast(first[i].params[j]) + != std::bit_cast(second[i].params[j])) return false; + } + } + return true; +} + +bool Valid(const StockQueueTraceRecord& r) { + const auto& b = r.basis; const auto& c = r.context; + if (r.perspective_team_id > 255 || b.token().empty() || b.token().size() > 64 + || b.state_sequence() == 0 || b.match_incarnation().size() != 16 + || !BoundedText(b.process_incarnation()) || !BoundedText(b.state_channel_incarnation()) + || b.snapshot_send_monotonic_ns() == 0 || b.effective_cadence_frames() == 0) return false; + if (c.profile != kStockTacticalProfile || c.revision != kStockTacticalRevision + || c.evidence_scheme != QueueEvidenceScheme::StockLuaSupportedFieldsV1 + || c.catalogue_id.size() != 16 || c.catalogue_revision == 0 + || c.engine_version != "2025.06.19" || !BoundedText(c.game_name) + || !BoundedText(c.game_version) || c.game_content_sha256.size() != 32 + || c.actor_id > 31999 || c.actor_lifetime == 0 + || (c.domain != "production" && c.domain != "rally")) return false; + if (r.read.canonical_request.empty() || r.read.canonical_request.size() > kStockQueueMaximumBytes + || r.read.unit_id != static_cast(c.actor_id) + || StockQueueDomainName(r.read.domain) != c.domain) return false; + if (r.read.status == StockQueueReadStatus::Complete) { + if (r.computed_revision == 0 || r.read.canonical_response.empty() + || r.read.canonical_response.size() > kStockQueueMaximumBytes + || r.read.entries.size() > kStockQueueMaximumEntries) return false; + std::size_t parameters = 0; + for (const auto& entry : r.read.entries) { + if (entry.params.size() > kStockQueueMaximumParametersPerEntry) return false; + parameters += entry.params.size(); + if (parameters > kStockQueueMaximumParameters) return false; + for (float value : entry.params) if (!std::isfinite(value)) return false; + } + const auto reparsed = ParseStockQueueResponse(r.read.canonical_response, + r.read.canonical_request, r.read.domain, r.read.unit_id); + if (reparsed.status != StockQueueReadStatus::Complete + || !SameEntries(reparsed.entries, r.read.entries) + || ComputeStockQueueRevision(c, r.read.entries) != r.computed_revision) return false; + } else if (r.computed_revision != 0 || !r.read.entries.empty()) return false; + else if (r.read.status == StockQueueReadStatus::Unavailable) { + const auto reparsed = ParseStockQueueResponse(r.read.canonical_response, + r.read.canonical_request, r.read.domain, r.read.unit_id); + if (reparsed.status != StockQueueReadStatus::Unavailable) return false; + } else if (!r.read.canonical_response.empty()) return false; + if (r.final_read != r.dispatch.has_value()) return false; + if (r.dispatch && (r.dispatch->broker_session_id.size() != 16 + || !BoundedText(r.dispatch->command_channel_incarnation) + || r.dispatch->batch_sequence == 0 || r.dispatch->correlation_id == 0 + || r.dispatch->authority_epoch == 0 || r.dispatch->module_sha256.size() != 32 + || r.dispatch->module_generation == 0 || r.dispatch->expected_queue_revision == 0)) return false; + return true; +} + +std::string Serialize(const std::string& run_id, std::uint64_t sequence, + const StockQueueTraceRecord& r) { + if (!Valid(r)) return {}; + const auto& b = r.basis; const auto& c = r.context; + std::ostringstream out; + out << "{\"schema\":\"highbar.barc-stock-queue-trace/v1\",\"runId\":" << Json(run_id) + << ",\"sequence\":" << Json(std::to_string(sequence)) << ",\"phase\":\"" + << (r.final_read ? "final-read" : "sample") << "\",\"readFrame\":" << r.read_frame + << ",\"perspectiveTeamId\":" << r.perspective_team_id + << ",\"nativeBasis\":{\"token\":" << Json(Base64(b.token())) + << ",\"stateSequence\":" << Json(std::to_string(b.state_sequence())) + << ",\"frame\":" << b.frame() << ",\"matchIncarnation\":" << Json(Base64(b.match_incarnation())) + << ",\"processIncarnation\":" << Json(b.process_incarnation()) + << ",\"stateChannelIncarnation\":" << Json(b.state_channel_incarnation()) + << ",\"snapshotSendMonotonicNs\":" << Json(std::to_string(b.snapshot_send_monotonic_ns())) + << ",\"effectiveCadenceFrames\":" << b.effective_cadence_frames() << "}" + << ",\"context\":{\"profile\":" << Json(c.profile) << ",\"tacticalRevision\":" << c.revision + << ",\"queueEvidenceScheme\":" << static_cast(c.evidence_scheme) + << ",\"actor\":{\"id\":" << Json(std::to_string(c.actor_id)) << ",\"lifetime\":" << Json(std::to_string(c.actor_lifetime)) << "}" + << ",\"catalogueId\":" << Json(Base64(c.catalogue_id)) << ",\"catalogueRevision\":" << Json(std::to_string(c.catalogue_revision)) + << ",\"engineVersion\":" << Json(c.engine_version) << ",\"gameName\":" << Json(c.game_name) + << ",\"gameVersion\":" << Json(c.game_version) << ",\"contentSha256\":" << Json(Hex(c.game_content_sha256)) + << ",\"domain\":" << Json(c.domain) << "}" + << ",\"reader\":{\"kind\":\"CallRules\",\"status\":" << Json(Status(r.read.status)) + << ",\"request\":" << Json(r.read.canonical_request) << ",\"response\":" + << (r.read.status == StockQueueReadStatus::Complete ? Json(r.read.canonical_response) : "null") << "}"; + if (r.read.status != StockQueueReadStatus::Complete) out << ",\"queue\":null"; + else { + out << ",\"queue\":{\"revision\":" << Json(std::to_string(r.computed_revision)) << ",\"entries\":["; + for (std::size_t i = 0; i < r.read.entries.size(); ++i) { + if (i) out << ','; + const auto& e = r.read.entries[i]; + out << "{\"id\":" << e.command_id << ",\"codedOptions\":" << e.options << ",\"tag\":" << e.tag << ",\"float32params\":["; + for (std::size_t j = 0; j < e.params.size(); ++j) { if (j) out << ','; out << Json(FloatBits(e.params[j])); } + out << "]}"; + } + out << "]}"; + } + if (!r.dispatch) out << ",\"dispatch\":null"; + else { const auto& d = *r.dispatch; + out << ",\"dispatch\":{\"brokerSessionId\":" << Json(Base64(d.broker_session_id)) + << ",\"commandChannelIncarnation\":" << Json(d.command_channel_incarnation) + << ",\"batchSequence\":" << Json(std::to_string(d.batch_sequence)) + << ",\"correlationId\":" << Json(std::to_string(d.correlation_id)) + << ",\"commandIndex\":" << d.command_index << ",\"authorityEpoch\":" << Json(std::to_string(d.authority_epoch)) + << ",\"moduleSha256\":" << Json(Hex(d.module_sha256)) << ",\"moduleGeneration\":" << Json(std::to_string(d.module_generation)) + << ",\"expectedQueueRevision\":" << Json(std::to_string(d.expected_queue_revision)) + << ",\"matched\":" << (d.matched ? "true" : "false") << "}"; } + out << '}'; return out.str(); +} + +} // namespace + +StockQueueTraceSink::StockQueueTraceSink(int fd, std::string run_id, + bool owns_fd, WriteFunction writer) + : fd_(fd), run_id_(std::move(run_id)), owns_fd_(owns_fd), writer_(std::move(writer)) { + if (!writer_) writer_ = [](int fd, const void* data, std::size_t size) { return ::write(fd, data, size); }; +} + +StockQueueTraceSink::~StockQueueTraceSink() { if (owns_fd_ && fd_ >= 0) ::close(fd_); } + +std::unique_ptr StockQueueTraceSink::CreateForTest( + int fd, std::string run_id, WriteFunction writer) { + if (fd < 0 || !SafeRunId(run_id)) return nullptr; + return std::unique_ptr(new StockQueueTraceSink(fd, std::move(run_id), false, std::move(writer))); +} + +std::unique_ptr StockQueueTraceSink::CreateFromEnvironment() { + const char* path_value = std::getenv("HIGHBAR_STOCK_QUEUE_TRACE"); + const char* run_value = std::getenv("HIGHBAR_STOCK_QUEUE_TRACE_RUN_ID"); + if (path_value == nullptr && run_value == nullptr) return nullptr; + if (path_value == nullptr || run_value == nullptr || !SafeRunId(run_value)) return nullptr; + const std::string path(path_value); const auto slash = path.find_last_of('/'); + if (path.empty() || path.front() != '/' || slash == std::string::npos + || slash == 0 || slash + 1 == path.size()) return nullptr; + const std::string parent = path.substr(0, slash), name = path.substr(slash + 1); + if (name == "." || name == "..") return nullptr; + const int parent_fd = ::open(parent.c_str(), O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); + if (parent_fd < 0) return nullptr; + struct stat parent_stat{}; + if (::fstat(parent_fd, &parent_stat) != 0 || !S_ISDIR(parent_stat.st_mode) + || parent_stat.st_uid != ::geteuid() || (parent_stat.st_mode & 0777) != 0700) { ::close(parent_fd); return nullptr; } + const int fd = ::openat(parent_fd, name.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC | O_APPEND, 0600); + ::close(parent_fd); if (fd < 0) return nullptr; + struct stat file_stat{}; + if (::fstat(fd, &file_stat) != 0 || !S_ISREG(file_stat.st_mode) || file_stat.st_nlink != 1 + || file_stat.st_uid != ::geteuid() || (file_stat.st_mode & 0777) != 0600) { ::close(fd); return nullptr; } + return std::unique_ptr(new StockQueueTraceSink(fd, run_value, true, {})); +} + +bool StockQueueTraceSink::WriteLine(const std::string& line) { + std::size_t offset = 0; + while (offset < line.size()) { + const auto result = writer_(fd_, line.data() + offset, line.size() - offset); + if (result < 0 && errno == EINTR) continue; + if (result <= 0 || static_cast(result) > line.size() - offset) return false; + offset += static_cast(result); + } + return true; +} + +bool StockQueueTraceSink::Record(const StockQueueTraceRecord& record) { + if (failed_) return false; + const auto line = Serialize(run_id_, records_written_ + 1, record) + "\n"; + if (line.size() == 1 || line.size() > kMaximumTraceLineBytes + || records_written_ >= kMaximumTraceRecords || bytes_written_ + line.size() > kMaximumTraceBytes + || !WriteLine(line)) { failed_ = true; return false; } + ++records_written_; bytes_written_ += line.size(); return true; +} + +} // namespace circuit::grpc diff --git a/src/circuit/grpc/StockQueueTrace.h b/src/circuit/grpc/StockQueueTrace.h new file mode 100644 index 00000000..a372f816 --- /dev/null +++ b/src/circuit/grpc/StockQueueTrace.h @@ -0,0 +1,72 @@ +// SPDX-License-Identifier: GPL-2.0-only +#pragma once + +#include "grpc/StockQueueReader.h" +#include "grpc/TacticalNativeState.h" +#include "highbar/live_control.pb.h" + +#include +#include +#include +#include +#include +#include +#include + +namespace circuit::grpc { + +struct StockQueueDispatchIdentity { + std::string broker_session_id; + std::string command_channel_incarnation; + std::uint64_t batch_sequence = 0; + std::uint64_t correlation_id = 0; + std::uint32_t command_index = 0; + std::uint64_t authority_epoch = 0; + std::string module_sha256; + std::uint64_t module_generation = 0; + std::uint64_t expected_queue_revision = 0; + bool matched = false; +}; + +struct StockQueueTraceRecord { + bool final_read = false; + std::uint32_t read_frame = 0; + std::uint32_t perspective_team_id = 0; + ::highbar::v1::NativeObservationBasis basis; + StockQueueRevisionContext context; + StockQueueReadResult read; + std::uint64_t computed_revision = 0; + std::optional dispatch; +}; + +class StockQueueTraceSink { +public: + using WriteFunction = std::function; + ~StockQueueTraceSink(); + StockQueueTraceSink(const StockQueueTraceSink&) = delete; + StockQueueTraceSink& operator=(const StockQueueTraceSink&) = delete; + + static std::unique_ptr CreateFromEnvironment(); + static std::unique_ptr CreateForTest( + int fd, std::string run_id, WriteFunction writer = {}); + + bool Record(const StockQueueTraceRecord& record); + bool failed() const { return failed_; } + std::uint64_t records_written() const { return records_written_; } + std::uint64_t bytes_written() const { return bytes_written_; } + +private: + StockQueueTraceSink(int fd, std::string run_id, bool owns_fd, + WriteFunction writer); + bool WriteLine(const std::string& line); + + int fd_ = -1; + std::string run_id_; + bool owns_fd_ = false; + bool failed_ = false; + std::uint64_t records_written_ = 0; + std::uint64_t bytes_written_ = 0; + WriteFunction writer_; +}; + +} // namespace circuit::grpc diff --git a/src/circuit/module/GrpcGatewayModule.cpp b/src/circuit/module/GrpcGatewayModule.cpp index 3fda8b47..498090ba 100644 --- a/src/circuit/module/GrpcGatewayModule.cpp +++ b/src/circuit/module/GrpcGatewayModule.cpp @@ -19,6 +19,7 @@ #include "grpc/GrpcLog.h" #include "grpc/RingBuffer.h" #include "grpc/StockQueueReader.h" +#include "grpc/StockQueueTrace.h" #include "grpc/OrderStateTracker.h" #include "grpc/LiveControlState.h" #include "grpc/TacticalNativeState.h" @@ -416,6 +417,7 @@ CGrpcGatewayModule::CGrpcGatewayModule(CCircuitAI* ai) // US2 pieces (T055 + T057 command path). command_queue_ = std::make_unique(counters_.get()); + stock_queue_trace_ = grpc::StockQueueTraceSink::CreateFromEnvironment(); order_state_tracker_ = std::make_unique(); admin_controller_ = std::make_unique(); admin_service_ = std::make_unique( @@ -749,6 +751,7 @@ void CGrpcGatewayModule::BuildAndReportTacticalCatalogue() { void CGrpcGatewayModule::BuildAndReportTacticalSnapshot( const ::highbar::v1::NativeObservationBasis& basis) { + std::vector pending_stock_trace; if (!tactical_profile_available_ || !tactical_catalogue_complete_) { TraceTacticalSuppression({grpc::TacticalSuppressionReason::CatalogueIncomplete, grpc::TacticalSuppressionSource::None, 0, 1}); @@ -914,6 +917,10 @@ void CGrpcGatewayModule::BuildAndReportTacticalSnapshot( std::optional> production_entries; std::optional> rally_entries; #ifdef HIGHBAR_STOCK_RECOIL + std::optional production_read; + std::optional rally_read; + std::uint32_t production_read_frame = 0; + std::uint32_t rally_read_frame = 0; auto read_stock_bridge = [&](grpc::StockQueueDomain domain) -> std::optional> { auto* lua = circuit->GetLua(); @@ -923,6 +930,13 @@ void CGrpcGatewayModule::BuildAndReportTacticalSnapshot( [&](const char* data, std::size_t size) { return lua->CallRules(data, static_cast(size)); }); + if (domain == grpc::StockQueueDomain::Production) { + production_read = read; + production_read_frame = CurrentFrame(); + } else { + rally_read = read; + rally_read_frame = CurrentFrame(); + } if (read.status != grpc::StockQueueReadStatus::Complete) { return std::nullopt; } @@ -1095,6 +1109,27 @@ void CGrpcGatewayModule::BuildAndReportTacticalSnapshot( production_entries, true); append_queue(::highbar::v1::NATIVE_QUEUE_DOMAIN_FACTORY_RALLY, rally_entries, false); +#ifdef HIGHBAR_STOCK_RECOIL + auto retain_trace = [&](const std::optional& read, + grpc::StockQueueDomain domain, std::uint32_t read_frame) { + if (!read.has_value()) return; + grpc::StockQueueRevisionContext context; + context.profile=grpc::kStockTacticalProfile; context.revision=grpc::kStockTacticalRevision; + context.evidence_scheme=grpc::QueueEvidenceScheme::StockLuaSupportedFieldsV1; + context.catalogue_id=tactical_catalogue_id_; context.catalogue_revision=tactical_catalogue_revision_; + context.engine_version=grpc::kEngineReleaseId; context.game_name=tactical_game_name_; + context.game_version=tactical_game_version_; context.game_content_sha256=tactical_game_content_sha256_; + context.actor_id=static_cast(unit_id); context.actor_lifetime=actor_out->actor().lifetime(); + context.domain=grpc::StockQueueDomainName(domain); + const auto revision = read->status == grpc::StockQueueReadStatus::Complete + ? grpc::ComputeStockQueueRevision(context, read->entries) : 0u; + pending_stock_trace.push_back({false, read_frame, + static_cast(circuit->GetTeamId()), basis, + std::move(context), *read, revision, std::nullopt}); + }; + retain_trace(production_read, grpc::StockQueueDomain::Production, production_read_frame); + retain_trace(rally_read, grpc::StockQueueDomain::Rally, rally_read_frame); +#endif } else { append_queue(::highbar::v1::NATIVE_QUEUE_DOMAIN_ACTOR_ORDER, ordinary_entries, true); @@ -1102,6 +1137,9 @@ void CGrpcGatewayModule::BuildAndReportTacticalSnapshot( } live_control_state_->RecordTacticalSnapshot(snapshot); coordinator_client_->ReportTacticalSnapshot(snapshot); + if (stock_queue_trace_) for (const auto& record : pending_stock_trace) { + stock_queue_trace_->Record(record); + } TacticalSuppressionTraceThrottle().Reset(); } @@ -2306,10 +2344,24 @@ void CGrpcGatewayModule::DrainCommandQueue() { && locked_state.OwnedLifetime(entry.live_actor.id()) == entry.live_actor.lifetime(); }; + std::optional trace_dispatch; + if (entry.live_tactical_command) { + trace_dispatch = grpc::StockQueueDispatchIdentity{ + entry.live_binding.broker_session_id(), + entry.live_binding.command_channel_incarnation(), + entry.batch_seq, entry.client_command_id, entry.command_index, + entry.live_binding.authority_epoch(), entry.live_binding.module_sha256(), + entry.live_binding.module_generation(), + entry.live_tactical_command->expected_queue_revision(), false}; + } const bool applied = entry.live_tactical_command ? grpc::DispatchTacticalCommand(circuit, fresh_actor, *entry.live_tactical_command, tactical_feature_lifetimes_.get(), stock_context_pointer, + stock_queue_trace_.get(), &entry.live_basis, + trace_dispatch ? &*trace_dispatch : nullptr, + [this] { return CurrentFrame(); }, + static_cast(circuit->GetTeamId()), post_control_fence, &tactical_refusal) : grpc::DispatchCommand(circuit, fresh_actor, cmd, fresh_target); diff --git a/src/circuit/module/GrpcGatewayModule.h b/src/circuit/module/GrpcGatewayModule.h index c364493a..0b4981db 100644 --- a/src/circuit/module/GrpcGatewayModule.h +++ b/src/circuit/module/GrpcGatewayModule.h @@ -57,6 +57,7 @@ class OrderStateTracker; class CoordinatorClient; class LiveControlState; class FeatureLifetimeLedger; +class StockQueueTraceSink; class MixedLifecycleQualification; struct QueuedCommand; } // namespace circuit::grpc @@ -228,6 +229,7 @@ class CGrpcGatewayModule final : public IModule { std::unique_ptr order_state_tracker_; std::unique_ptr live_control_state_; std::unique_ptr tactical_feature_lifetimes_; + std::unique_ptr stock_queue_trace_; std::unique_ptr mixed_lifecycle_qualification_; std::unique_ptr mixed_lifecycle_held_command_; std::unique_ptr service_; diff --git a/tests/unit/stock_queue_trace_test.cc b/tests/unit/stock_queue_trace_test.cc new file mode 100644 index 00000000..a5c40a7d --- /dev/null +++ b/tests/unit/stock_queue_trace_test.cc @@ -0,0 +1,134 @@ +// SPDX-License-Identifier: GPL-2.0-only +#include "grpc/StockQueueTrace.h" + +#include + +#include +#include +#include +#include +#include +#include +#include + +namespace circuit::grpc { +namespace { + +std::string Response(StockQueueDomain domain, std::int32_t unit, bool unavailable) { + const auto request = BuildStockQueueRequest(domain, unit); + std::string response = "BARC_QUEUE_RESPONSE/1\nlength=00000000\nbridge=barc-stock-queue-reader-v1\nrequest-sha256=" + + Sha256Hex(request) + "\nstatus=" + (unavailable ? "unavailable" : "ok") + + "\ndomain=" + StockQueueDomainName(domain) + "\nunit=" + std::to_string(unit) + "\n"; + if (unavailable) response += "reason=wrong-team\ncount=0\n"; + else response += "count=1\nrow=" + std::string(StockQueueDomainName(domain)) + "|-1|32|9|3f800000,80000000\n"; + response += "end\n"; + const auto size = std::to_string(response.size()); + response.replace(response.find("length=") + 7, 8, std::string(8 - size.size(), '0') + size); + return response; +} + +StockQueueTraceRecord Record(bool final_read = false) { + StockQueueTraceRecord value; + value.final_read = final_read; value.read_frame = 41; value.perspective_team_id = 3; + value.basis.set_token("token"); value.basis.set_state_sequence(12); value.basis.set_frame(40); + value.basis.set_match_incarnation(std::string(16, 'm')); + value.basis.set_process_incarnation("process"); value.basis.set_state_channel_incarnation("state"); + value.basis.set_snapshot_send_monotonic_ns(99); value.basis.set_effective_cadence_frames(3); + value.context.profile = kStockTacticalProfile; value.context.revision = kStockTacticalRevision; + value.context.evidence_scheme = QueueEvidenceScheme::StockLuaSupportedFieldsV1; + value.context.catalogue_id = std::string(16, 'c'); value.context.catalogue_revision = 8; + value.context.engine_version = "2025.06.19"; value.context.game_name = "BAR"; + value.context.game_version = "test"; value.context.game_content_sha256 = std::string(32, 'h'); + value.context.actor_id = 0; value.context.actor_lifetime = 7; value.context.domain = "production"; + value.read.status = StockQueueReadStatus::Complete; value.read.domain = StockQueueDomain::Production; + value.read.unit_id = 0; value.read.canonical_request = BuildStockQueueRequest(StockQueueDomain::Production, 0); + value.read.canonical_response = Response(StockQueueDomain::Production, 0, false); + value.read.entries = {{-1, 32, 9, {1.0f, -0.0f}}}; + value.computed_revision = ComputeStockQueueRevision(value.context, value.read.entries); + if (final_read) { + StockQueueDispatchIdentity dispatch; dispatch.broker_session_id = std::string(16, 's'); + dispatch.command_channel_incarnation = "channel"; dispatch.batch_sequence = 2; + dispatch.correlation_id = 3; dispatch.command_index = 0; dispatch.authority_epoch = 4; + dispatch.module_sha256 = std::string(32, 'z'); dispatch.module_generation = 5; + dispatch.expected_queue_revision = value.computed_revision; dispatch.matched = true; + value.dispatch = dispatch; + } + return value; +} + +TEST(StockQueueTrace, WritesClosedSampleAndFinalIdentityWithSignedZeroBits) { + char path[] = "/tmp/highbar-stock-trace-test-XXXXXX"; const int fd = ::mkstemp(path); + ASSERT_GE(fd, 0); auto sink = StockQueueTraceSink::CreateForTest(fd, "run-1"); ASSERT_TRUE(sink); + EXPECT_TRUE(sink->Record(Record())); EXPECT_TRUE(sink->Record(Record(true))); + ::lseek(fd, 0, SEEK_SET); std::string bytes; char buffer[4096]; ssize_t count; + while ((count = ::read(fd, buffer, sizeof buffer)) > 0) bytes.append(buffer, count); + EXPECT_NE(bytes.find("\"sequence\":\"1\",\"phase\":\"sample\""), std::string::npos); + EXPECT_NE(bytes.find("\"float32params\":[\"3f800000\",\"80000000\"]"), std::string::npos); + EXPECT_NE(bytes.find("\"sequence\":\"2\",\"phase\":\"final-read\""), std::string::npos); + EXPECT_NE(bytes.find("\"brokerSessionId\":\"c3Nzc3Nzc3Nzc3Nzc3Nzcw==\""), std::string::npos); + ::close(fd); ::unlink(path); +} + +TEST(StockQueueTrace, PartialWritesCompleteButZeroWriteLatchesFailure) { + char path[] = "/tmp/highbar-stock-trace-write-XXXXXX"; const int fd = ::mkstemp(path); ASSERT_GE(fd, 0); + auto partial = StockQueueTraceSink::CreateForTest(fd, "run", [](int descriptor, const void* data, std::size_t size) { + return ::write(descriptor, data, std::min(size, 7)); + }); + EXPECT_TRUE(partial->Record(Record())); EXPECT_FALSE(partial->failed()); partial.reset(); + auto stopped = StockQueueTraceSink::CreateForTest(fd, "run", [](int, const void*, std::size_t) { return ssize_t{0}; }); + EXPECT_FALSE(stopped->Record(Record())); EXPECT_TRUE(stopped->failed()); EXPECT_FALSE(stopped->Record(Record())); + ::close(fd); ::unlink(path); +} + +TEST(StockQueueTrace, MissingBasisMalformedShapeAndOversizeFailClosed) { + char path[] = "/tmp/highbar-stock-trace-invalid-XXXXXX"; const int fd = ::mkstemp(path); ASSERT_GE(fd, 0); + auto missing = Record(); missing.basis.clear_token(); + auto sink = StockQueueTraceSink::CreateForTest(fd, "run"); ASSERT_TRUE(sink); + EXPECT_FALSE(sink->Record(missing)); EXPECT_TRUE(sink->failed()); + ::close(fd); ::unlink(path); +} + +TEST(StockQueueTrace, UnavailableAndMalformedNeverRetainResponseOrMintQueue) { + char path[] = "/tmp/highbar-stock-trace-status-XXXXXX"; const int fd = ::mkstemp(path); ASSERT_GE(fd, 0); + auto sink = StockQueueTraceSink::CreateForTest(fd, "run"); ASSERT_TRUE(sink); + auto unavailable = Record(); unavailable.read.status = StockQueueReadStatus::Unavailable; + unavailable.read.canonical_response = Response(StockQueueDomain::Production, 0, true); + unavailable.read.entries.clear(); unavailable.computed_revision = 0; + EXPECT_TRUE(sink->Record(unavailable)); + auto malformed = unavailable; malformed.read.status = StockQueueReadStatus::Malformed; + malformed.read.canonical_response.clear(); + EXPECT_TRUE(sink->Record(malformed)); + ::lseek(fd, 0, SEEK_SET); std::string bytes; char buffer[4096]; ssize_t count; + while ((count = ::read(fd, buffer, sizeof buffer)) > 0) bytes.append(buffer, count); + EXPECT_EQ(bytes.find("reason=wrong-team"), std::string::npos); + EXPECT_NE(bytes.find("\"status\":\"unavailable\",\"request\":"), std::string::npos); + EXPECT_NE(bytes.find("\"status\":\"malformed\",\"request\":"), std::string::npos); + EXPECT_NE(bytes.find("\"response\":null},\"queue\":null"), std::string::npos); + ::close(fd); ::unlink(path); +} + +TEST(StockQueueTrace, RecordCeilingLatchesWithoutWritingPastBound) { + auto sink = StockQueueTraceSink::CreateForTest(1, "run", + [](int, const void*, std::size_t size) { return static_cast(size); }); + ASSERT_TRUE(sink); + for (int index = 0; index < 4096; ++index) ASSERT_TRUE(sink->Record(Record())) << index; + EXPECT_FALSE(sink->Record(Record())); EXPECT_TRUE(sink->failed()); EXPECT_EQ(sink->records_written(), 4096u); +} + +TEST(StockQueueTrace, EnvironmentRequiresOwnedModeAndCreateExclusive) { + ::unsetenv("HIGHBAR_STOCK_QUEUE_TRACE"); + ::unsetenv("HIGHBAR_STOCK_QUEUE_TRACE_RUN_ID"); + EXPECT_FALSE(StockQueueTraceSink::CreateFromEnvironment()); + const auto root = std::filesystem::temp_directory_path() / ("highbar-stock-trace-parent-" + std::to_string(::getpid())); + std::filesystem::create_directory(root); ::chmod(root.c_str(), 0755); + const auto path = root / "trace.jsonl"; + ::setenv("HIGHBAR_STOCK_QUEUE_TRACE", path.c_str(), 1); ::setenv("HIGHBAR_STOCK_QUEUE_TRACE_RUN_ID", "native.run", 1); + EXPECT_FALSE(StockQueueTraceSink::CreateFromEnvironment()); + ::chmod(root.c_str(), 0700); auto sink = StockQueueTraceSink::CreateFromEnvironment(); ASSERT_TRUE(sink); + struct stat info{}; ASSERT_EQ(::stat(path.c_str(), &info), 0); EXPECT_EQ(info.st_mode & 0777, 0600); EXPECT_EQ(info.st_nlink, 1); + EXPECT_FALSE(StockQueueTraceSink::CreateFromEnvironment()); + sink.reset(); std::filesystem::remove_all(root); ::unsetenv("HIGHBAR_STOCK_QUEUE_TRACE"); ::unsetenv("HIGHBAR_STOCK_QUEUE_TRACE_RUN_ID"); +} + +} // namespace +} // namespace circuit::grpc From 233b11f541986ee65e17062e2eed8548e795fd94 Mon Sep 17 00:00:00 2001 From: EHotwagner Date: Thu, 1 Oct 2026 11:35:56 +0200 Subject: [PATCH 2/2] Latch stock queue trace failures safely --- src/circuit/grpc/StockQueueTrace.cpp | 113 +++++++++++++++++++++------ src/circuit/grpc/StockQueueTrace.h | 9 ++- tests/unit/stock_queue_trace_test.cc | 81 ++++++++++++++++++- 3 files changed, 174 insertions(+), 29 deletions(-) diff --git a/src/circuit/grpc/StockQueueTrace.cpp b/src/circuit/grpc/StockQueueTrace.cpp index 5ad718d4..c2f59a56 100644 --- a/src/circuit/grpc/StockQueueTrace.cpp +++ b/src/circuit/grpc/StockQueueTrace.cpp @@ -19,6 +19,18 @@ constexpr std::size_t kMaximumTraceLineBytes = 32768; constexpr std::uint64_t kMaximumTraceRecords = 4096; constexpr std::uint64_t kMaximumTraceBytes = 16 * 1024 * 1024; +class ScopedFd { +public: + explicit ScopedFd(int fd = -1) noexcept : fd_(fd) {} + ~ScopedFd() { if (fd_ >= 0) ::close(fd_); } + ScopedFd(const ScopedFd&) = delete; + ScopedFd& operator=(const ScopedFd&) = delete; + int get() const noexcept { return fd_; } + int release() noexcept { const int result = fd_; fd_ = -1; return result; } +private: + int fd_; +}; + bool SafeRunId(std::string_view value) { if (value.empty() || value.size() > 64) return false; for (const unsigned char c : value) { @@ -244,27 +256,54 @@ std::unique_ptr StockQueueTraceSink::CreateForTest( return std::unique_ptr(new StockQueueTraceSink(fd, std::move(run_id), false, std::move(writer))); } +std::unique_ptr StockQueueTraceSink::AdoptOwned( + int fd, std::string run_id, WriteFunction writer) noexcept { + ScopedFd owned(fd); + if (fd < 0) return nullptr; + try { + if (!SafeRunId(run_id)) return nullptr; + auto result = std::unique_ptr( + new StockQueueTraceSink(fd, std::move(run_id), true, std::move(writer))); + owned.release(); + return result; + } catch (...) { + return nullptr; + } +} + +std::unique_ptr StockQueueTraceSink::AdoptOwnedForTest( + int fd, std::string run_id, WriteFunction writer) { + return AdoptOwned(fd, std::move(run_id), std::move(writer)); +} + std::unique_ptr StockQueueTraceSink::CreateFromEnvironment() { - const char* path_value = std::getenv("HIGHBAR_STOCK_QUEUE_TRACE"); - const char* run_value = std::getenv("HIGHBAR_STOCK_QUEUE_TRACE_RUN_ID"); - if (path_value == nullptr && run_value == nullptr) return nullptr; - if (path_value == nullptr || run_value == nullptr || !SafeRunId(run_value)) return nullptr; - const std::string path(path_value); const auto slash = path.find_last_of('/'); - if (path.empty() || path.front() != '/' || slash == std::string::npos - || slash == 0 || slash + 1 == path.size()) return nullptr; - const std::string parent = path.substr(0, slash), name = path.substr(slash + 1); - if (name == "." || name == "..") return nullptr; - const int parent_fd = ::open(parent.c_str(), O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC); - if (parent_fd < 0) return nullptr; - struct stat parent_stat{}; - if (::fstat(parent_fd, &parent_stat) != 0 || !S_ISDIR(parent_stat.st_mode) - || parent_stat.st_uid != ::geteuid() || (parent_stat.st_mode & 0777) != 0700) { ::close(parent_fd); return nullptr; } - const int fd = ::openat(parent_fd, name.c_str(), O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC | O_APPEND, 0600); - ::close(parent_fd); if (fd < 0) return nullptr; - struct stat file_stat{}; - if (::fstat(fd, &file_stat) != 0 || !S_ISREG(file_stat.st_mode) || file_stat.st_nlink != 1 - || file_stat.st_uid != ::geteuid() || (file_stat.st_mode & 0777) != 0600) { ::close(fd); return nullptr; } - return std::unique_ptr(new StockQueueTraceSink(fd, run_value, true, {})); + try { + const char* path_value = std::getenv("HIGHBAR_STOCK_QUEUE_TRACE"); + const char* run_value = std::getenv("HIGHBAR_STOCK_QUEUE_TRACE_RUN_ID"); + if (path_value == nullptr && run_value == nullptr) return nullptr; + if (path_value == nullptr || run_value == nullptr || !SafeRunId(run_value)) return nullptr; + std::string run_id(run_value); + const std::string path(path_value); const auto slash = path.find_last_of('/'); + if (path.empty() || path.front() != '/' || slash == std::string::npos + || slash == 0 || slash + 1 == path.size()) return nullptr; + const std::string parent = path.substr(0, slash), name = path.substr(slash + 1); + if (name == "." || name == "..") return nullptr; + ScopedFd parent_fd(::open(parent.c_str(), O_RDONLY | O_DIRECTORY | O_NOFOLLOW | O_CLOEXEC)); + if (parent_fd.get() < 0) return nullptr; + struct stat parent_stat{}; + if (::fstat(parent_fd.get(), &parent_stat) != 0 || !S_ISDIR(parent_stat.st_mode) + || parent_stat.st_uid != ::geteuid() || (parent_stat.st_mode & 0777) != 0700) return nullptr; + const int fd = ::openat(parent_fd.get(), name.c_str(), + O_WRONLY | O_CREAT | O_EXCL | O_NOFOLLOW | O_CLOEXEC | O_APPEND, 0600); + if (fd < 0) return nullptr; + ScopedFd file_fd(fd); + struct stat file_stat{}; + if (::fstat(fd, &file_stat) != 0 || !S_ISREG(file_stat.st_mode) || file_stat.st_nlink != 1 + || file_stat.st_uid != ::geteuid() || (file_stat.st_mode & 0777) != 0600) return nullptr; + return AdoptOwned(file_fd.release(), std::move(run_id), {}); + } catch (...) { + return nullptr; + } } bool StockQueueTraceSink::WriteLine(const std::string& line) { @@ -278,13 +317,35 @@ bool StockQueueTraceSink::WriteLine(const std::string& line) { return true; } -bool StockQueueTraceSink::Record(const StockQueueTraceRecord& record) { +void StockQueueTraceSink::LatchFailure() noexcept { + failed_ = true; + const int descriptor = fd_; + fd_ = -1; + if (descriptor >= 0) ::close(descriptor); +} + +bool StockQueueTraceSink::Record(const StockQueueTraceRecord& record) noexcept { if (failed_) return false; - const auto line = Serialize(run_id_, records_written_ + 1, record) + "\n"; - if (line.size() == 1 || line.size() > kMaximumTraceLineBytes - || records_written_ >= kMaximumTraceRecords || bytes_written_ + line.size() > kMaximumTraceBytes - || !WriteLine(line)) { failed_ = true; return false; } - ++records_written_; bytes_written_ += line.size(); return true; + try { + auto line = Serialize(run_id_, records_written_ + 1, record); + if (line.empty() || line.size() >= kMaximumTraceLineBytes + || records_written_ >= kMaximumTraceRecords + || line.size() + 1 > kMaximumTraceBytes - bytes_written_) { + LatchFailure(); + return false; + } + line.push_back('\n'); + if (!WriteLine(line)) { + LatchFailure(); + return false; + } + ++records_written_; + bytes_written_ += line.size(); + return true; + } catch (...) { + LatchFailure(); + return false; + } } } // namespace circuit::grpc diff --git a/src/circuit/grpc/StockQueueTrace.h b/src/circuit/grpc/StockQueueTrace.h index a372f816..01182e05 100644 --- a/src/circuit/grpc/StockQueueTrace.h +++ b/src/circuit/grpc/StockQueueTrace.h @@ -49,8 +49,12 @@ class StockQueueTraceSink { static std::unique_ptr CreateFromEnvironment(); static std::unique_ptr CreateForTest( int fd, std::string run_id, WriteFunction writer = {}); + // Exercises the same exception-safe owned-descriptor adoption used by the + // environment factory. The descriptor is consumed on every return path. + static std::unique_ptr AdoptOwnedForTest( + int fd, std::string run_id, WriteFunction writer = {}); - bool Record(const StockQueueTraceRecord& record); + bool Record(const StockQueueTraceRecord& record) noexcept; bool failed() const { return failed_; } std::uint64_t records_written() const { return records_written_; } std::uint64_t bytes_written() const { return bytes_written_; } @@ -58,7 +62,10 @@ class StockQueueTraceSink { private: StockQueueTraceSink(int fd, std::string run_id, bool owns_fd, WriteFunction writer); + static std::unique_ptr AdoptOwned( + int fd, std::string run_id, WriteFunction writer) noexcept; bool WriteLine(const std::string& line); + void LatchFailure() noexcept; int fd_ = -1; std::string run_id_; diff --git a/tests/unit/stock_queue_trace_test.cc b/tests/unit/stock_queue_trace_test.cc index a5c40a7d..992882d9 100644 --- a/tests/unit/stock_queue_trace_test.cc +++ b/tests/unit/stock_queue_trace_test.cc @@ -6,11 +6,30 @@ #include #include #include +#include #include +#include +#include #include #include #include +namespace { +thread_local bool fail_next_allocation = false; +} + +void* operator new(std::size_t size) { + if (fail_next_allocation) { + fail_next_allocation = false; + throw std::bad_alloc(); + } + if (void* value = std::malloc(size)) return value; + throw std::bad_alloc(); +} + +void operator delete(void* value) noexcept { std::free(value); } +void operator delete(void* value, std::size_t) noexcept { std::free(value); } + namespace circuit::grpc { namespace { @@ -56,6 +75,16 @@ StockQueueTraceRecord Record(bool final_read = false) { return value; } +bool ProcessHasOpenDescriptorFor(const std::filesystem::path& path) { + std::error_code error; + for (const auto& entry : std::filesystem::directory_iterator("/proc/self/fd", error)) { + const auto target = std::filesystem::read_symlink(entry.path(), error); + if (!error && target == path) return true; + error.clear(); + } + return false; +} + TEST(StockQueueTrace, WritesClosedSampleAndFinalIdentityWithSignedZeroBits) { char path[] = "/tmp/highbar-stock-trace-test-XXXXXX"; const int fd = ::mkstemp(path); ASSERT_GE(fd, 0); auto sink = StockQueueTraceSink::CreateForTest(fd, "run-1"); ASSERT_TRUE(sink); @@ -77,14 +106,58 @@ TEST(StockQueueTrace, PartialWritesCompleteButZeroWriteLatchesFailure) { EXPECT_TRUE(partial->Record(Record())); EXPECT_FALSE(partial->failed()); partial.reset(); auto stopped = StockQueueTraceSink::CreateForTest(fd, "run", [](int, const void*, std::size_t) { return ssize_t{0}; }); EXPECT_FALSE(stopped->Record(Record())); EXPECT_TRUE(stopped->failed()); EXPECT_FALSE(stopped->Record(Record())); + EXPECT_EQ(::fcntl(fd, F_GETFD), -1); EXPECT_EQ(errno, EBADF); ::close(fd); ::unlink(path); } +TEST(StockQueueTrace, AllocationFailureClosesDescriptorAndPermanentlyRefuses) { + char path[] = "/tmp/highbar-stock-trace-alloc-XXXXXX"; const int fd = ::mkstemp(path); ASSERT_GE(fd, 0); + auto sink = StockQueueTraceSink::CreateForTest(fd, "run"); ASSERT_TRUE(sink); + const auto record = Record(); + fail_next_allocation = true; + EXPECT_FALSE(sink->Record(record)); + EXPECT_TRUE(sink->failed()); + EXPECT_EQ(::fcntl(fd, F_GETFD), -1); EXPECT_EQ(errno, EBADF); + EXPECT_FALSE(sink->Record(record)); + EXPECT_EQ(sink->records_written(), 0u); EXPECT_EQ(sink->bytes_written(), 0u); + ::unlink(path); +} + +TEST(StockQueueTrace, ThrowingWriterCannotEscapeOrChangeCallerControlFlow) { + char path[] = "/tmp/highbar-stock-trace-throw-XXXXXX"; const int fd = ::mkstemp(path); ASSERT_GE(fd, 0); + auto sink = StockQueueTraceSink::CreateForTest(fd, "run", + [](int, const void*, std::size_t) -> ssize_t { throw std::runtime_error("write failed"); }); + ASSERT_TRUE(sink); + bool synchronized_effect_reached = false; + const auto dispatch = [&] { + (void)sink->Record(Record(true)); + synchronized_effect_reached = true; + return true; + }; + EXPECT_TRUE(dispatch()); + EXPECT_TRUE(synchronized_effect_reached); + EXPECT_TRUE(sink->failed()); + EXPECT_EQ(::fcntl(fd, F_GETFD), -1); EXPECT_EQ(errno, EBADF); + EXPECT_FALSE(sink->Record(Record(true))); + ::unlink(path); +} + +TEST(StockQueueTrace, OwnedDescriptorClosesWhenSinkConstructionAllocationFails) { + char path[] = "/tmp/highbar-stock-trace-adopt-XXXXXX"; const int fd = ::mkstemp(path); ASSERT_GE(fd, 0); + std::string run = "run"; + fail_next_allocation = true; + auto sink = StockQueueTraceSink::AdoptOwnedForTest(fd, std::move(run)); + EXPECT_FALSE(sink); + EXPECT_EQ(::fcntl(fd, F_GETFD), -1); EXPECT_EQ(errno, EBADF); + ::unlink(path); +} + TEST(StockQueueTrace, MissingBasisMalformedShapeAndOversizeFailClosed) { char path[] = "/tmp/highbar-stock-trace-invalid-XXXXXX"; const int fd = ::mkstemp(path); ASSERT_GE(fd, 0); auto missing = Record(); missing.basis.clear_token(); auto sink = StockQueueTraceSink::CreateForTest(fd, "run"); ASSERT_TRUE(sink); EXPECT_FALSE(sink->Record(missing)); EXPECT_TRUE(sink->failed()); + EXPECT_EQ(::fcntl(fd, F_GETFD), -1); EXPECT_EQ(errno, EBADF); ::close(fd); ::unlink(path); } @@ -108,11 +181,13 @@ TEST(StockQueueTrace, UnavailableAndMalformedNeverRetainResponseOrMintQueue) { } TEST(StockQueueTrace, RecordCeilingLatchesWithoutWritingPastBound) { - auto sink = StockQueueTraceSink::CreateForTest(1, "run", + const int fd = ::open("/dev/null", O_WRONLY | O_CLOEXEC); ASSERT_GE(fd, 0); + auto sink = StockQueueTraceSink::CreateForTest(fd, "run", [](int, const void*, std::size_t size) { return static_cast(size); }); ASSERT_TRUE(sink); for (int index = 0; index < 4096; ++index) ASSERT_TRUE(sink->Record(Record())) << index; EXPECT_FALSE(sink->Record(Record())); EXPECT_TRUE(sink->failed()); EXPECT_EQ(sink->records_written(), 4096u); + EXPECT_EQ(::fcntl(fd, F_GETFD), -1); EXPECT_EQ(errno, EBADF); } TEST(StockQueueTrace, EnvironmentRequiresOwnedModeAndCreateExclusive) { @@ -126,8 +201,10 @@ TEST(StockQueueTrace, EnvironmentRequiresOwnedModeAndCreateExclusive) { EXPECT_FALSE(StockQueueTraceSink::CreateFromEnvironment()); ::chmod(root.c_str(), 0700); auto sink = StockQueueTraceSink::CreateFromEnvironment(); ASSERT_TRUE(sink); struct stat info{}; ASSERT_EQ(::stat(path.c_str(), &info), 0); EXPECT_EQ(info.st_mode & 0777, 0600); EXPECT_EQ(info.st_nlink, 1); + EXPECT_TRUE(ProcessHasOpenDescriptorFor(path)); EXPECT_FALSE(StockQueueTraceSink::CreateFromEnvironment()); - sink.reset(); std::filesystem::remove_all(root); ::unsetenv("HIGHBAR_STOCK_QUEUE_TRACE"); ::unsetenv("HIGHBAR_STOCK_QUEUE_TRACE_RUN_ID"); + sink.reset(); EXPECT_FALSE(ProcessHasOpenDescriptorFor(path)); + std::filesystem::remove_all(root); ::unsetenv("HIGHBAR_STOCK_QUEUE_TRACE"); ::unsetenv("HIGHBAR_STOCK_QUEUE_TRACE_RUN_ID"); } } // namespace