From c9d48267de86e2713cc044910878ff258bf8f634 Mon Sep 17 00:00:00 2001 From: Bhagirath Mehta Date: Fri, 6 Nov 2026 13:24:12 -0600 Subject: [PATCH 1/3] 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); From 606c7d8c0268dbd345e84ade318911cd9e151ac3 Mon Sep 17 00:00:00 2001 From: Bhagirath Mehta Date: Fri, 6 Nov 2026 13:54:53 -0600 Subject: [PATCH 2/3] Preserve reservation provenance when upload retrieval throws Address Copilot round 1 on microsoft/cpp_client_telemetry#1559. 4496757055: Mark the active memory/disk backend before retrieval can throw and copy that provenance to the upload context during unwinding, so partial RAM reservations return to RAM rather than being stranded. Verified at lib/offline/OfflineStorageHandler.cpp:654 and lib/offline/StorageObserver.cpp:66. Add four standard/nonstandard RAM/disk regressions in tests/unittests/OfflineStorageTests.cpp. 4496757163: Verify the zero-delay barrier through its future rather than a task pointer that is cleared on completion. Verified at lib/pal/TaskDispatcher.hpp:73; update tests/functests/upload-exception-tests.cpp. Files: lib/offline/OfflineStorageHandler.cpp, lib/offline/StorageObserver.cpp, tests/unittests/OfflineStorageTests.cpp, tests/functests/upload-exception-tests.cpp. Validation: Windows 195 and Linux 199 targeted unit tests; six native upload cases and no-exceptions smoke on both platforms; 20 repetitions of all four Windows preparation barriers. Changed-code clang-format and git diff --check passed; no selected tests were skipped or excluded. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: ffecbc0a-951e-4cae-9e94-49ddce94c050 --- lib/offline/OfflineStorageHandler.cpp | 4 +- lib/offline/StorageObserver.cpp | 23 ++++-- tests/functests/upload-exception-tests.cpp | 5 +- tests/unittests/OfflineStorageTests.cpp | 95 ++++++++++++++++++++++ 4 files changed, 117 insertions(+), 10 deletions(-) diff --git a/lib/offline/OfflineStorageHandler.cpp b/lib/offline/OfflineStorageHandler.cpp index 99e12ad0a..cc9fcfce2 100644 --- a/lib/offline/OfflineStorageHandler.cpp +++ b/lib/offline/OfflineStorageHandler.cpp @@ -651,11 +651,11 @@ namespace MAT_NS_BEGIN { if (m_offlineStorageMemory) { + m_readFromMemory = true; returnValue |= m_offlineStorageMemory->GetAndReserveRecords(consumer, leaseTimeMs, minLatency, maxCount); m_lastReadCount += m_offlineStorageMemory->LastReadRecordCount(); if (m_lastReadCount <= maxCount) maxCount -= m_lastReadCount; - m_readFromMemory = true; // Prefer to send all of in-memory first before going to disk. This also helps in case if in-ram queue // is larger than request size (2MB), we'd exit the function because the consumer no longer wants more // records. @@ -665,12 +665,12 @@ namespace MAT_NS_BEGIN { if (m_offlineStorageDisk) { + m_readFromMemory = false; returnValue |= m_offlineStorageDisk->GetAndReserveRecords(consumer, leaseTimeMs, minLatency, maxCount); auto lastOfflineReadCount = m_offlineStorageDisk->LastReadRecordCount(); if (lastOfflineReadCount) { m_lastReadCount += lastOfflineReadCount; - m_readFromMemory = false; } } diff --git a/lib/offline/StorageObserver.cpp b/lib/offline/StorageObserver.cpp index e0bcc341e..c99aa3c25 100644 --- a/lib/offline/StorageObserver.cpp +++ b/lib/offline/StorageObserver.cpp @@ -43,21 +43,35 @@ namespace MAT_NS_BEGIN { void StorageObserver::handleRetrieveEvents(EventsUploadContextPtr const& ctx) { - auto consumer = [&ctx, this](StorageRecord&& record) -> bool { + auto consumer = [&ctx, this](StorageRecord&& record) -> bool + { bool wantMore = true; retrievedEvent(ctx, std::move(record), wantMore); return wantMore; }; - // TODO: [MG] - expose 120000 as a configuration parameter - if (!m_offlineStorage.GetAndReserveRecords(consumer, 120000, ctx->requestedMinLatency, ctx->requestedMaxCount)) + bool retrieved; +#if HAVE_EXCEPTIONS + try + { +#endif + // TODO: [MG] - expose 120000 as a configuration parameter + retrieved = m_offlineStorage.GetAndReserveRecords(consumer, 120000, ctx->requestedMinLatency, ctx->requestedMaxCount); +#if HAVE_EXCEPTIONS + } + catch (...) { ctx->fromMemory = m_offlineStorage.IsLastReadFromMemory(); + throw; + } +#endif + ctx->fromMemory = m_offlineStorage.IsLastReadFromMemory(); + if (!retrieved) + { retrievalFailed(ctx); } else { - ctx->fromMemory = m_offlineStorage.IsLastReadFromMemory(); retrievalFinished(ctx); } } @@ -217,4 +231,3 @@ namespace MAT_NS_BEGIN { } } MAT_NS_END - diff --git a/tests/functests/upload-exception-tests.cpp b/tests/functests/upload-exception-tests.cpp index f09f885cf..5bfc2d23b 100644 --- a/tests/functests/upload-exception-tests.cpp +++ b/tests/functests/upload-exception-tests.cpp @@ -219,9 +219,8 @@ int main(int argc, char** argv) 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"); + PAL::scheduleTask(dispatcher.get(), 0, [drained]() + { drained->set_value(); }); 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) diff --git a/tests/unittests/OfflineStorageTests.cpp b/tests/unittests/OfflineStorageTests.cpp index 802c061ad..d92424299 100644 --- a/tests/unittests/OfflineStorageTests.cpp +++ b/tests/unittests/OfflineStorageTests.cpp @@ -7,12 +7,15 @@ #include "offline/OfflineStorageHandler.hpp" #include "offline/IOfflineStorageProvider.hpp" #include "offline/StorageObserver.hpp" +#include "offline/MemoryStorage.hpp" #include "NullObjects.hpp" #include #include #include #include +#include +#include #include using namespace testing; @@ -256,6 +259,98 @@ namespace MAT_NS_BEGIN { } MAT_NS_END +#if HAVE_EXCEPTIONS +class OfflineStorageRetrievalExceptionTests : public OfflineStorageTests, + public WithParamInterface> +{ +}; + +TEST_P(OfflineStorageRetrievalExceptionTests, PartialRetrievalReleasesRecordsToTheirOriginalStorage) +{ + const bool fromMemory = std::get<0>(GetParam()); + const bool nonstandard = std::get<1>(GetParam()); + NullLogManager logManager; + NiceMock config; + NoopTaskDispatcher dispatcher; + NiceMock observer; + config[CFG_INT_RAM_QUEUE_SIZE] = 1024 * 1024; + auto memory = std::make_shared(logManager, config); + auto disk = std::make_shared>(); + auto provider = std::make_shared(memory, disk); + OfflineStorageHandler handler(logManager, config, dispatcher, provider); + EXPECT_CALL(*disk, Initialize(Ref(handler))); + handler.Initialize(observer); + StorageObserver storage(testing::getSystem(), handler); + storage.retrievedEvent >> retrievedEvent; + storage.retrievalFinished >> retrievalFinished; + storage.retrievalFailed >> retrievalFailed; + + StorageRecord first("first", "tenant-token", EventLatency_Normal, + EventPersistence_Normal, 1, std::vector{'x'}); + StorageRecord second("second", "tenant-token", EventLatency_Normal, + EventPersistence_Normal, 1, std::vector{'y'}); + if (fromMemory) + { + ASSERT_TRUE(memory->StoreRecord(first)); + ASSERT_TRUE(memory->StoreRecord(second)); + } + else + { + EXPECT_CALL(*disk, GetAndReserveRecords(_, _, _, _)) + .WillOnce(Invoke([&](std::function const& consumer, unsigned, EventLatency, unsigned) + { + EXPECT_TRUE(consumer(std::move(first))); + return consumer(std::move(second)); })); + } + + auto ctx = std::make_shared(); + ctx->fromMemory = !fromMemory; + EXPECT_CALL(*this, resultRetrievedEvent(ctx, _, _)) + .WillOnce(Invoke([](EventsUploadContextPtr const& upload, StorageRecord const& record, bool&) + { upload->recordIdsAndTenantIds.emplace(record.id, record.tenantToken); })) + .WillOnce(Invoke([&](EventsUploadContextPtr const&, StorageRecord const&, bool&) + { + if (nonstandard) + { + throw 7; + } + throw std::runtime_error("mid-retrieval failure"); })); + if (nonstandard) + { + EXPECT_THROW(storage.retrieveEvents(ctx), int); + } + else + { + EXPECT_THROW(storage.retrieveEvents(ctx), std::runtime_error); + } + ASSERT_EQ(ctx->recordIdsAndTenantIds.size(), size_t{1}); + EXPECT_EQ(ctx->fromMemory, fromMemory); + EXPECT_EQ(handler.IsLastReadFromMemory(), fromMemory); + if (fromMemory) + { + ASSERT_EQ(memory->GetReservedCount(), size_t{1}); + } + else + { + EXPECT_CALL(*disk, ReleaseRecords(ElementsAre(ctx->recordIdsAndTenantIds.begin()->first), false, _, Eq(false))); + } + EXPECT_TRUE(storage.releaseRecords(ctx)); + if (fromMemory) + { + EXPECT_EQ(memory->GetReservedCount(), size_t{0}); + std::vector retriedIds; + EXPECT_TRUE(handler.GetAndReserveRecords([&](StorageRecord&& record) + { + retriedIds.push_back(record.id); + return true; }, 120000)); + EXPECT_THAT(retriedIds, UnorderedElementsAre("first", "second")); + EXPECT_TRUE(handler.IsLastReadFromMemory()); + } +} + +INSTANTIATE_TEST_SUITE_P(MemoryAndDisk, OfflineStorageRetrievalExceptionTests, Combine(Bool(), Bool())); +#endif + TEST(OfflineStorageHandlerFlushTests, FailedMemoryRequeueIsReportedAndDropped) { NullLogManager logManager; From dbb9ef2d08e2938d2346fffa30268ccab92039e7 Mon Sep 17 00:00:00 2001 From: Bhagirath Mehta Date: Fri, 6 Nov 2026 14:06:09 -0600 Subject: [PATCH 3/3] Keep RAM event provenance when disk retrieval consumes nothing Address Copilot round 2, comment 4496977358: preserve pre-call disk exception provenance, but restore the previous source when a successful disk read consumes no records. This prevents an oversized first RAM event from being mislabeled as disk-backed and uploaded repeatedly after successful completion. Verified at lib/offline/MemoryStorage.cpp:163-168 and lib/packager/Packager.cpp:31-41. Files changed: lib/offline/OfflineStorageHandler.cpp and tests/unittests/OfflineStorageTests.cpp. The new regression failed before the fix, leaving the acknowledged RAM event queued, and passes afterward. Windows and Linux: 25 storage tests, six native upload CTests, and no-exceptions smoke passed without skipped or excluded selected tests. Changed-code clang-format and git diff --check passed. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: ffecbc0a-951e-4cae-9e94-49ddce94c050 --- lib/offline/OfflineStorageHandler.cpp | 5 ++++ tests/unittests/OfflineStorageTests.cpp | 40 +++++++++++++++++++++++++ 2 files changed, 45 insertions(+) diff --git a/lib/offline/OfflineStorageHandler.cpp b/lib/offline/OfflineStorageHandler.cpp index cc9fcfce2..60f67fb64 100644 --- a/lib/offline/OfflineStorageHandler.cpp +++ b/lib/offline/OfflineStorageHandler.cpp @@ -665,6 +665,7 @@ namespace MAT_NS_BEGIN { if (m_offlineStorageDisk) { + const bool previousReadFromMemory = m_readFromMemory; m_readFromMemory = false; returnValue |= m_offlineStorageDisk->GetAndReserveRecords(consumer, leaseTimeMs, minLatency, maxCount); auto lastOfflineReadCount = m_offlineStorageDisk->LastReadRecordCount(); @@ -672,6 +673,10 @@ namespace MAT_NS_BEGIN { { m_lastReadCount += lastOfflineReadCount; } + else + { + m_readFromMemory = previousReadFromMemory; + } } if (m_config.IsClockSkewEnabled() && !m_clockSkewManager.GetResumeTransmissionAfterClockSkew() diff --git a/tests/unittests/OfflineStorageTests.cpp b/tests/unittests/OfflineStorageTests.cpp index d92424299..9f3f5fc35 100644 --- a/tests/unittests/OfflineStorageTests.cpp +++ b/tests/unittests/OfflineStorageTests.cpp @@ -259,6 +259,46 @@ namespace MAT_NS_BEGIN { } MAT_NS_END +TEST_F(OfflineStorageTests, EmptyDiskReadPreservesFirstRamEventProvenance) +{ + NullLogManager logManager; + NiceMock config; + NoopTaskDispatcher dispatcher; + NiceMock observer; + config[CFG_INT_RAM_QUEUE_SIZE] = 1024 * 1024; + auto memory = std::make_shared(logManager, config); + auto disk = std::make_shared>(); + auto provider = std::make_shared(memory, disk); + OfflineStorageHandler handler(logManager, config, dispatcher, provider); + EXPECT_CALL(*disk, Initialize(Ref(handler))); + handler.Initialize(observer); + StorageObserver storage(testing::getSystem(), handler); + storage.retrievedEvent >> retrievedEvent; + storage.retrievalFinished >> retrievalFinished; + + StorageRecord record("oversized", "tenant-token", EventLatency_Normal, + EventPersistence_Normal, 1, std::vector{'x'}); + ASSERT_TRUE(memory->StoreRecord(record)); + auto ctx = std::make_shared(); + EXPECT_CALL(*this, resultRetrievedEvent(ctx, _, _)) + .WillOnce(Invoke([](EventsUploadContextPtr const& upload, StorageRecord const& event, bool& wantMore) + { + // Packager includes an oversized first event but refuses subsequent events. + upload->recordIdsAndTenantIds.emplace(event.id, event.tenantToken); + wantMore = false; })); + EXPECT_CALL(*disk, GetAndReserveRecords(_, _, _, _)).WillOnce(Return(true)); + EXPECT_CALL(*disk, LastReadRecordCount()).WillOnce(Return(0)); + EXPECT_CALL(*this, resultRetrievalFinished(ctx)); + + storage.retrieveEvents(ctx); + EXPECT_TRUE(ctx->fromMemory); + EXPECT_TRUE(handler.IsLastReadFromMemory()); + EXPECT_EQ(memory->LastReadRecordCount(), 0u); + EXPECT_EQ(memory->GetRecordCount(), size_t{1}); + EXPECT_TRUE(storage.deleteRecords(ctx)); + EXPECT_EQ(memory->GetRecordCount(), size_t{0}); +} + #if HAVE_EXCEPTIONS class OfflineStorageRetrievalExceptionTests : public OfflineStorageTests, public WithParamInterface>