From c9d48267de86e2713cc044910878ff258bf8f634 Mon Sep 17 00:00:00 2001 From: Bhagirath Mehta Date: Fri, 6 Nov 2026 13:24:12 -0600 Subject: [PATCH] Prevent preparation exceptions from stranding active uploads Port the SDK recovery from bmehta001/onnxruntime#11 so failed upload preparation cannot leave teardown waiting for a callback that was never registered. Release reserved records before retiring an upload, contain cleanup exceptions, and preserve HTTP ownership after registration. Protect pending callback allocation with RAII and evaluate request logging before handing ownership to the registry. Adapt all 14 component cases and six native fault modes to the SDK harness, plus a request-log failure test. Files changed: - lib/http/HttpClientManager.cpp - lib/system/Contexts.hpp - lib/system/TelemetrySystem.cpp - lib/tpm/TransmissionPolicyManager.cpp - lib/tpm/TransmissionPolicyManager.hpp - tests/functests/CMakeLists.txt - tests/functests/upload-exception-tests.cpp - tests/unittests/HttpClientManagerTests.cpp - tests/unittests/TransmissionPolicyManagerTests.cpp Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 6982f030-2ce2-47a3-b52b-9d8a1cdb05d9 --- lib/http/HttpClientManager.cpp | 14 +- lib/system/Contexts.hpp | 2 +- lib/system/TelemetrySystem.cpp | 1 + lib/tpm/TransmissionPolicyManager.cpp | 66 ++++- lib/tpm/TransmissionPolicyManager.hpp | 3 + tests/functests/CMakeLists.txt | 17 ++ tests/functests/upload-exception-tests.cpp | 250 ++++++++++++++++++ tests/unittests/HttpClientManagerTests.cpp | 38 +++ .../TransmissionPolicyManagerTests.cpp | 108 ++++++++ 9 files changed, 490 insertions(+), 9 deletions(-) create mode 100644 tests/functests/upload-exception-tests.cpp diff --git a/lib/http/HttpClientManager.cpp b/lib/http/HttpClientManager.cpp index de29a84c3..ac591ddd2 100644 --- a/lib/http/HttpClientManager.cpp +++ b/lib/http/HttpClientManager.cpp @@ -367,17 +367,21 @@ namespace MAT_NS_BEGIN { void HttpClientManager::handleSendRequest(EventsUploadContextPtr const& ctx) { auto registry = m_callbackRegistry; - HttpCallback *callback = new HttpCallback(ctx, registry); + auto pendingCallback = std::unique_ptr(new HttpCallback(ctx, registry)); + HttpCallback* callback = pendingCallback.get(); auto completion = callback->m_completion; - { - LOCKGUARD(registry->mutex); - registry->callbacks.push_back(callback); - } LOG_INFO("Uploading %u event(s) of priority %d (%s) for %u tenant(s) in HTTP request %s (approx. %u bytes)...", static_cast(ctx->recordIdsAndTenantIds.size()), ctx->latency, latencyToStr(ctx->latency), static_cast(ctx->packageIds.size()), ctx->httpRequest->GetId().c_str(), static_cast(ctx->httpRequest->GetSizeEstimate())); + { + LOCKGUARD(registry->mutex); + registry->callbacks.push_back(callback); + ctx->httpRequestRegistered = true; + pendingCallback.release(); + } + #if HAVE_EXCEPTIONS try { diff --git a/lib/system/Contexts.hpp b/lib/system/Contexts.hpp index fa4529e78..19bc81ab7 100644 --- a/lib/system/Contexts.hpp +++ b/lib/system/Contexts.hpp @@ -111,6 +111,7 @@ namespace MAT_NS_BEGIN { // Sending IHttpRequest* httpRequest = nullptr; std::string httpRequestId; + bool httpRequestRegistered = false; // Receiving IHttpResponse* httpResponse = nullptr; @@ -151,4 +152,3 @@ namespace MAT_NS_BEGIN { } MAT_NS_END - diff --git a/lib/system/TelemetrySystem.cpp b/lib/system/TelemetrySystem.cpp index 31780e506..64d666c81 100644 --- a/lib/system/TelemetrySystem.cpp +++ b/lib/system/TelemetrySystem.cpp @@ -174,6 +174,7 @@ namespace MAT_NS_BEGIN { storage.storeRecordFailed >> stats.onIncomingEventFailed; tpm.initiateUpload >> storage.retrieveEvents; + tpm.releaseFailedUpload >> storage.releaseRecords; storage.retrievedEvent >> packager.addEventToPackage; storage.retrievalFinished >> packager.finalizePackage; diff --git a/lib/tpm/TransmissionPolicyManager.cpp b/lib/tpm/TransmissionPolicyManager.cpp index c8ba7ee30..b03566c71 100644 --- a/lib/tpm/TransmissionPolicyManager.cpp +++ b/lib/tpm/TransmissionPolicyManager.cpp @@ -7,6 +7,7 @@ #include "TransmitProfiles.hpp" #include "utils/Utils.hpp" +#include #include namespace MAT_NS_BEGIN { @@ -256,8 +257,68 @@ namespace MAT_NS_BEGIN { auto ctx = m_system.createEventsUploadContext(); ctx->requestedMinLatency = requestedLatency; + startUpload(ctx); + } + + void TransmissionPolicyManager::startUpload(EventsUploadContextPtr const& ctx) + { addUpload(ctx); - initiateUpload(ctx); +#if HAVE_EXCEPTIONS + try + { +#endif + initiateUpload(ctx); +#if HAVE_EXCEPTIONS + } + catch (const std::exception& ex) + { + (void)ex; + LOG_ERROR("Upload preparation failed: %s", ex.what()); + abandonUpload(ctx); + } + catch (...) + { + LOG_ERROR("Upload preparation failed with a non-standard exception"); + abandonUpload(ctx); + } +#endif + } + + void TransmissionPolicyManager::abandonUpload(EventsUploadContextPtr const& ctx) + { + // Once registered, the HTTP callback owns terminal completion, even if + // SendRequestAsync completes synchronously or throws after completion. + if (ctx->httpRequestRegistered) + { + return; + } + { + LOCKGUARD(m_activeUploads_lock); + if (m_activeUploads.find(ctx) == m_activeUploads.end()) + { + return; + } + } +#if HAVE_EXCEPTIONS + try + { +#endif + releaseFailedUpload(ctx); +#if HAVE_EXCEPTIONS + } + catch (const std::exception& ex) + { + (void)ex; + LOG_ERROR("Failed to release records after upload preparation failed: %s", ex.what()); + } + catch (...) + { + LOG_ERROR("Failed to release records after upload preparation failed with a non-standard exception"); + } +#endif + // Cleanup is best effort, but retiring the upload must not depend on it. + // Keep it active until cleanup finishes so teardown cannot race storage. + removeUpload(ctx); } void TransmissionPolicyManager::finishUpload(EventsUploadContextPtr const& ctx, const std::chrono::milliseconds& nextUpload) @@ -382,8 +443,7 @@ namespace MAT_NS_BEGIN { if (event->record.latency > EventLatency_RealTime) { auto ctx = m_system.createEventsUploadContext(); ctx->requestedMinLatency = event->record.latency; - addUpload(ctx); - initiateUpload(ctx); + startUpload(ctx); return; } diff --git a/lib/tpm/TransmissionPolicyManager.hpp b/lib/tpm/TransmissionPolicyManager.hpp index 840dac107..f81d7e9a6 100644 --- a/lib/tpm/TransmissionPolicyManager.hpp +++ b/lib/tpm/TransmissionPolicyManager.hpp @@ -83,6 +83,8 @@ constexpr const char* const DefaultBackoffConfig = "E,3000,300000,2,1"; std::chrono::milliseconds increaseBackoff(); void uploadAsync(EventLatency priority); + void startUpload(EventsUploadContextPtr const& ctx); + void abandonUpload(EventsUploadContextPtr const& ctx); void finishUpload(EventsUploadContextPtr const& ctx, const std::chrono::milliseconds& nextUpload); bool updateTimersIfNecessary(); @@ -179,6 +181,7 @@ constexpr const char* const DefaultBackoffConfig = "E,3000,300000,2,1"; RouteSink eventArrived{ this, &TransmissionPolicyManager::handleEventArrived }; RouteSource initiateUpload; + RouteSource releaseFailedUpload; RouteSink nothingToUpload{ this, &TransmissionPolicyManager::handleNothingToUpload }; RouteSink packagingFailed{ this, &TransmissionPolicyManager::handlePackagingFailed }; RouteSink eventsUploadSuccessful{ this, &TransmissionPolicyManager::handleEventsUploadSuccessful }; diff --git a/tests/functests/CMakeLists.txt b/tests/functests/CMakeLists.txt index 058a5a1e4..defc99ce6 100644 --- a/tests/functests/CMakeLists.txt +++ b/tests/functests/CMakeLists.txt @@ -79,3 +79,20 @@ endif() add_test(NAME FuncTests COMMAND FuncTests "--gtest_output=xml:${PROJECT_BINARY_DIR}/test-reports/FuncTests.xml") set_tests_properties(FuncTests PROPERTIES WORKING_DIRECTORY "${PROJECT_BINARY_DIR}") + +if(NOT MATSDK_DISABLE_EXCEPTIONS AND NOT MATSDK_PLATFORM_IOS + AND (WIN32 OR MATSDK_NEEDS_CURL OR MATSDK_BUILD_APPLE_HTTP)) + add_executable(UploadExceptionTests upload-exception-tests.cpp ${TESTS_COMMON_SRCS}) + target_link_libraries(UploadExceptionTests PRIVATE matsdk_test_config mat ZLIB::ZLIB gtest gmock) + if(NOT MATSDK_SQLITE_PROVIDER_RESOLVED STREQUAL "NONE") + target_link_libraries(UploadExceptionTests PRIVATE SQLite3::SQLite3) + endif() + if(TARGET nlohmann_json::nlohmann_json) + target_link_libraries(UploadExceptionTests PRIVATE nlohmann_json::nlohmann_json) + endif() + foreach(mode IN ITEMS exception exception-nonstandard exception-exit + exception-exit-nonstandard send-exception complete-then-throw) + add_test(NAME upload-${mode} COMMAND UploadExceptionTests ${mode}) + set_tests_properties(upload-${mode} PROPERTIES TIMEOUT 30) + endforeach() +endif() diff --git a/tests/functests/upload-exception-tests.cpp b/tests/functests/upload-exception-tests.cpp new file mode 100644 index 000000000..f09f885cf --- /dev/null +++ b/tests/functests/upload-exception-tests.cpp @@ -0,0 +1,250 @@ +// +// Copyright (c) Microsoft Corporation. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 +// + +#ifndef WIN32_LEAN_AND_MEAN +#define WIN32_LEAN_AND_MEAN +#endif + +#include "LogManagerProvider.hpp" +#include "common/HttpServer.hpp" +#include "http/HttpClientFactory.hpp" +#include "pal/TaskDispatcher.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace MAT; + +namespace +{ + void Check(bool condition, const char* message) + { + if (!condition) + { + std::fprintf(stderr, "FAIL: %s\n", message); + std::_Exit(1); + } + } + + class Collector : public testing::HttpServer::Callback + { + public: + Collector() + { + const int port = server.addListeningPort(0); + Check(port > 0, "could not bind collector"); + url = "http://127.0.0.1:" + std::to_string(port) + "/collect"; + server.setKeepalive(false); + server.addHandler("/collect", *this); + server.start(); + } + + ~Collector() + { + server.stop(); + } + + int onHttpRequest(testing::HttpServer::Request const& request, + testing::HttpServer::Response& response) override + { + { + std::lock_guard lock(mutex); + bodies.push_back(request.content); + } + changed.notify_all(); + response.content = "{ \"status\": \"0\" }"; + return 200; + } + + bool WaitFor(const std::string& event) + { + std::unique_lock lock(mutex); + return changed.wait_for(lock, std::chrono::seconds(10), [&]() + { return std::any_of(bodies.begin(), bodies.end(), [&](const std::string& body) + { return body.find(event) != std::string::npos; }); }); + } + + std::string url; + + private: + testing::HttpServer server; + std::mutex mutex; + std::condition_variable changed; + std::vector bodies; + }; + + class ThrowingListener : public DebugEventListener + { + public: + explicit ThrowingListener(bool nonstandard) : + nonstandard(nonstandard) + { + } + + void OnDebugEvent(DebugEvent&) override + { + { + std::lock_guard lock(mutex); + if (injected) + { + return; + } + injected = true; + } + changed.notify_all(); + if (nonstandard) + { + throw 7; + } + throw std::runtime_error("injected upload preparation failure"); + } + + void WaitForFailure() + { + std::unique_lock lock(mutex); + Check(changed.wait_for(lock, std::chrono::seconds(10), [&]() + { return injected; }), + "SDK did not reach the throwing upload-start listener"); + } + + private: + const bool nonstandard; + std::mutex mutex; + std::condition_variable changed; + bool injected = false; + }; + + class FaultingHttpClient : public IHttpClient + { + public: + FaultingHttpClient(const std::string& mode, const std::string& url) : + mode(mode), url(url), client(HttpClientFactory::Create()) + { + Check(client != nullptr, "native HTTP transport unavailable"); + } + + IHttpRequest* CreateRequest() override + { + return client->CreateRequest(); + } + + void SendRequestAsync(IHttpRequest* request, IHttpResponseCallback* callback) override + { + if ((mode == "send-exception" || mode == "complete-then-throw") && !injected.exchange(true)) + { + if (mode == "complete-then-throw") + { + callback->OnHttpResponse(new SimpleHttpResponse(request->GetId())); + } + throw std::runtime_error("injected HTTP handoff failure"); + } + request->SetUrl(url); + ++forwarded; + client->SendRequestAsync(request, callback); + } + + void CancelRequestAsync(const std::string& id) override + { + client->CancelRequestAsync(id); + } + + void CancelAllRequests() override + { + client->CancelAllRequests(); + } + + void ApplySettings(ILogConfiguration& config) override + { + client->ApplySettings(config); + } + + std::atomic injected{false}; + std::atomic forwarded{0}; + + private: + const std::string mode; + const std::string url; + std::shared_ptr client; + }; +} + +int main(int argc, char** argv) +{ + Check(argc == 2, "expected fault mode"); + const std::string mode(argv[1]); + const bool preparation = mode.compare(0, 9, "exception") == 0; + const bool exitWithoutRetry = mode == "exception-exit" || mode == "exception-exit-nonstandard"; + Check(preparation || mode == "send-exception" || mode == "complete-then-throw", "unknown fault mode"); + + Collector collector; + ThrowingListener listener(mode == "exception-nonstandard" || mode == "exception-exit-nonstandard"); + ILogConfiguration config; + config[CFG_STR_COLLECTOR_URL] = collector.url; + config[CFG_STR_CACHE_FILE_PATH] = ":memory:"; + config[CFG_BOOL_ENABLE_NET_DETECT] = false; + config[CFG_INT_MAX_TEARDOWN_TIME] = 0; + config[CFG_MAP_HTTP][CFG_BOOL_HTTP_COMPRESSION] = false; + config[CFG_MAP_METASTATS_CONFIG]["enabled"] = false; + config[CFG_MAP_TPM][CFG_STR_TPM_BACKOFF] = "E,10,100,2,0"; + auto client = std::make_shared(mode, collector.url); + client->ApplySettings(config); + config.AddModule(CFG_MODULE_HTTP_CLIENT, client); + + status_t status; + ILogManager* manager = LogManagerProvider::CreateLogManager("UploadExceptionTests", true, config, status); + Check(manager != nullptr && status == STATUS_SUCCESS, "could not initialize SDK"); + manager->SetTransmitProfile(TransmitProfile_RealTime); + if (preparation) + { + manager->AddEventListener(EVT_SENDING, listener); + } + ILogger* logger = manager->GetLogger("00000000000000000000000000000000-00000000-0000-0000-0000-000000000000-0000"); + Check(logger != nullptr, "could not create logger"); + logger->LogEvent(EventProperties("originalReservedEvent")); + manager->UploadNow(); + + if (preparation) + { + listener.WaitForFailure(); + // Run behind the failed upload so record cleanup has finished before retry or teardown. + auto drained = std::make_shared>(); + auto done = drained->get_future(); + auto dispatcher = PAL::getDefaultTaskDispatcher(); + auto barrier = PAL::scheduleTask(dispatcher.get(), 0, [drained]() + { drained->set_value(); }); + Check(barrier.GetTask() != nullptr, "could not queue upload failure barrier"); + Check(done.wait_for(std::chrono::seconds(10)) == std::future_status::ready, "failed upload task did not finish"); + Check(client->forwarded == 0, "preparation failure reached HTTP handoff"); + if (!exitWithoutRetry) + { + manager->UploadNow(); + } + } + if (!exitWithoutRetry) + { + Check(collector.WaitFor("originalReservedEvent"), "original reserved event was not retried through native transport"); + logger->LogEvent(EventProperties("subsequentEvent")); + manager->UploadNow(); + Check(collector.WaitFor("subsequentEvent"), "subsequent event did not upload"); + if (!preparation) + { + Check(client->injected, "HTTP handoff failure was not exercised"); + } + } + manager->FlushAndTeardown(); + Check(LogManagerProvider::Release(config) == STATUS_SUCCESS, "could not release SDK"); + if (exitWithoutRetry) + { + Check(client->forwarded == 0, "teardown retried the failed upload"); + } + std::puts("PASS: native upload exception recovery and teardown"); +} diff --git a/tests/unittests/HttpClientManagerTests.cpp b/tests/unittests/HttpClientManagerTests.cpp index 22178f367..94850d521 100644 --- a/tests/unittests/HttpClientManagerTests.cpp +++ b/tests/unittests/HttpClientManagerTests.cpp @@ -344,6 +344,7 @@ TEST_F(HttpClientManagerTests, HandlesRequestFlow) .WillOnce(SaveArg<1>(&callback)); hcm.sendRequest(ctx); ASSERT_THAT(callback, NotNull()); + EXPECT_TRUE(ctx->httpRequestRegistered); PAL::sleep(200); @@ -360,6 +361,41 @@ TEST_F(HttpClientManagerTests, HandlesRequestFlow) EXPECT_THAT(ctx->durationMs, Gt(199)); } +#if HAVE_EXCEPTIONS && defined(HAVE_MAT_LOGGING) +TEST_F(HttpClientManagerTests, ThrowingRequestLogFormattingLeavesNoCallback) +{ + class ThrowingRequest : public SimpleHttpRequest + { + public: + ThrowingRequest() : + SimpleHttpRequest("throwing-request-log") + { + } + + size_t GetSizeEstimate() const override + { + throw std::runtime_error("request log formatting failed"); + } + }; + + auto ctx = std::make_shared(); + ctx->httpRequest = new ThrowingRequest(); + ctx->httpRequestId = "throwing-request-log"; + std::weak_ptr context = ctx; + EXPECT_CALL(httpClientMock, SendRequestAsync(_, _)).Times(0); + + const auto previousLogLevel = PAL::detail::g_logLevel; + MATSDK_SET_LOG_LEVEL_(PAL::Info); + EXPECT_THROW(hcm.sendRequest(ctx), std::runtime_error); + MATSDK_SET_LOG_LEVEL_(previousLogLevel); + + EXPECT_FALSE(ctx->httpRequestRegistered); + EXPECT_THAT(hcm.requestCount(), 0u); + ctx.reset(); + EXPECT_TRUE(context.expired()); +} +#endif + TEST_F(HttpClientManagerTests, ThrowingRequestDoneSettlesFailureAndDrainsCallback) { auto ctx = std::make_shared(); @@ -432,6 +468,7 @@ TEST_F(HttpClientManagerTests, ThrowingSendProducesOneTerminalFailure) })); EXPECT_NO_THROW(hcm.sendRequest(ctx)); + EXPECT_TRUE(ctx->httpRequestRegistered); EXPECT_THAT(hcm.requestCount(), 0u); } @@ -459,6 +496,7 @@ TEST_F(HttpClientManagerTests, CallbackThenThrowDoesNotCompleteTwice) })); EXPECT_NO_THROW(hcm.sendRequest(ctx)); + EXPECT_TRUE(ctx->httpRequestRegistered); EXPECT_THAT(hcm.requestCount(), 0u); } diff --git a/tests/unittests/TransmissionPolicyManagerTests.cpp b/tests/unittests/TransmissionPolicyManagerTests.cpp index 0e868db9e..907ba3c3e 100644 --- a/tests/unittests/TransmissionPolicyManagerTests.cpp +++ b/tests/unittests/TransmissionPolicyManagerTests.cpp @@ -12,6 +12,8 @@ #include #include #include +#include +#include #include using namespace testing; @@ -453,6 +455,112 @@ TEST_F(TransmissionPolicyManagerTests, ImmediateIncomingEventStartsUploadImmedia EXPECT_THAT(upload->requestedMinLatency, EventLatency_Max); } +#if HAVE_EXCEPTIONS +enum class UploadFailure +{ + Standard, + Nonstandard, + CleanupStandard, + CleanupNonstandard, + Registered, + Completed, + CompletedThenThrow +}; + +class UploadExceptionTests : public TransmissionPolicyManagerTests, + public WithParamInterface> +{ + protected: + RouteSink release{ + this, &UploadExceptionTests::releaseRecords}; + + MOCK_METHOD1(releaseRecords, void(EventsUploadContextPtr const&)); +}; + +TEST_P(UploadExceptionTests, ReleasesOnlyUnregisteredUploadsAndAllowsStop) +{ + const bool immediate = std::get<0>(GetParam()); + const UploadFailure failure = std::get<1>(GetParam()); + const bool httpOwned = failure == UploadFailure::Registered || + failure == UploadFailure::Completed || failure == UploadFailure::CompletedThenThrow; + std::weak_ptr context; + EventsUploadContextPtr pending; + + tpm.paused(false); + tpm.releaseFailedUpload >> release; + EXPECT_CALL(*this, resultInitiateUpload(_)) + .WillOnce(Invoke([&](EventsUploadContextPtr const& ctx) + { + context = ctx; + if (httpOwned) + { + ctx->httpRequestRegistered = true; + if (failure == UploadFailure::Registered) + { + pending = ctx; + } + else + { + tpm.eventsUploadAborted(ctx); + if (failure == UploadFailure::Completed) + { + return; + } + } + } + if (failure == UploadFailure::Nonstandard) + { + throw 7; + } + throw std::runtime_error("injected upload preparation failure"); })); + if (httpOwned) + { + EXPECT_CALL(*this, releaseRecords(_)).Times(0); + } + else + { + EXPECT_CALL(*this, releaseRecords(_)) + .WillOnce(Invoke([&](EventsUploadContextPtr const& ctx) + { + EXPECT_THAT(tpm.activeUploads(), Contains(ctx)); + if (failure == UploadFailure::CleanupStandard) + { + throw std::runtime_error("injected record cleanup failure"); + } + if (failure == UploadFailure::CleanupNonstandard) + { + throw 9; + } })); + } + + if (immediate) + { + IncomingEventContext event; + event.record.latency = EventLatency_Max; + EXPECT_NO_THROW(tpm.eventArrived(&event)); + } + else + { + EXPECT_NO_THROW(tpm.uploadAsyncParent(EventLatency_Normal)); + } + + if (failure == UploadFailure::Registered) + { + EXPECT_THAT(tpm.activeUploads(), Contains(pending)); + EXPECT_FALSE(context.expired()); + tpm.eventsUploadAborted(pending); + tpm.eventsUploadAborted(pending); + pending.reset(); + } + ASSERT_THAT(tpm.activeUploads(), IsEmpty()); + EXPECT_TRUE(context.expired()); + EXPECT_CALL(*this, resultAllUploadsFinished()); + EXPECT_TRUE(tpm.stop()); +} + +INSTANTIATE_TEST_SUITE_P(ScheduledAndImmediate, UploadExceptionTests, Combine(Bool(), Values(UploadFailure::Standard, UploadFailure::Nonstandard, UploadFailure::CleanupStandard, UploadFailure::CleanupNonstandard, UploadFailure::Registered, UploadFailure::Completed, UploadFailure::CompletedThenThrow))); +#endif + TEST_F(TransmissionPolicyManagerTests, UploadDoesNothingWhenPaused) { tpm.uploadScheduled(true);