Recover uploads abandoned by preparation exceptions

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<HttpCallback>(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<unsigned>(ctx->recordIdsAndTenantIds.size()), ctx->latency, latencyToStr(ctx->latency), static_cast<unsigned>(ctx->packageIds.size()),
             ctx->httpRequest->GetId().c_str(), static_cast<unsigned>(ctx->httpRequest->GetSizeEstimate()));
 
+        {
+            LOCKGUARD(registry->mutex);
+            registry->callbacks.push_back(callback);
+            ctx->httpRequestRegistered = true;
+            pendingCallback.release();
+        }
+
 #if HAVE_EXCEPTIONS
         try
         {

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,17 @@ 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();
             if (lastOfflineReadCount)
             {
                 m_lastReadCount += lastOfflineReadCount;
-                m_readFromMemory = false;
+            }
+            else
+            {
+                m_readFromMemory = previousReadFromMemory;
             }
         }
 

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
-

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
-

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;

lib/tpm/TransmissionPolicyManager.cpp

@@ -7,6 +7,7 @@
 #include "TransmitProfiles.hpp"
 #include "utils/Utils.hpp"
 
+#include <exception>
 #include <limits>
 
 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;
         }
 

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<TransmissionPolicyManager, IncomingEventContextPtr const&> eventArrived{ this, &TransmissionPolicyManager::handleEventArrived };
 
         RouteSource<EventsUploadContextPtr const&>                           initiateUpload;
+        RouteSource<EventsUploadContextPtr const&>                           releaseFailedUpload;
         RouteSink<TransmissionPolicyManager, EventsUploadContextPtr const&>  nothingToUpload{ this, &TransmissionPolicyManager::handleNothingToUpload };
         RouteSink<TransmissionPolicyManager, EventsUploadContextPtr const&>  packagingFailed{ this, &TransmissionPolicyManager::handlePackagingFailed };
         RouteSink<TransmissionPolicyManager, EventsUploadContextPtr const&>  eventsUploadSuccessful{ this, &TransmissionPolicyManager::handleEventsUploadSuccessful };

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()