Offline storage: data-safety fixes + batched flush (empty-filter guard, store-failure propagation, event-loss fix, one-transaction flush)

lib/api/LogConfiguration.cpp

@@ -19,6 +19,7 @@ namespace MAT_NS_BEGIN {
         { CFG_BOOL_ENABLE_ANALYTICS,        false },
         { CFG_INT_CACHE_FILE_SIZE,          3145728 },
         { CFG_INT_RAM_QUEUE_SIZE,           524288 },
+        { CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH, true },
         { CFG_BOOL_ENABLE_MULTITENANT,      true },
         { CFG_BOOL_ENABLE_DB_DROP_IF_FULL,  false },
         { CFG_INT_MAX_TEARDOWN_TIME,        0 },
@@ -51,6 +52,7 @@ namespace MAT_NS_BEGIN {
             { CFG_BOOL_ENABLE_ANALYTICS,    src.enableLifecycleSession },
             { CFG_INT_CACHE_FILE_SIZE,      src.cacheFileSizeLimitInBytes },
             { CFG_INT_RAM_QUEUE_SIZE,       src.cacheMemorySizeLimitInBytes },
+            { CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH, true },
             { CFG_BOOL_ENABLE_MULTITENANT,  src.multiTenantEnabled },
             { CFG_INT_MAX_TEARDOWN_TIME,    src.maxTeardownUploadTimeInSec },
             { CFG_INT_MAX_PENDING_REQ,      src.maxPendingHTTPRequests },
@@ -128,4 +130,3 @@ namespace MAT_NS_BEGIN {
     }
 
 } MAT_NS_END
-

lib/config/RuntimeConfig_Default.hpp

@@ -16,6 +16,7 @@ namespace MAT_NS_BEGIN
         {CFG_BOOL_ENABLE_ANALYTICS, false},
         {CFG_INT_CACHE_FILE_SIZE, 3145728},
         {CFG_INT_RAM_QUEUE_SIZE, 524288},
+        {CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH, true},
         {CFG_BOOL_ENABLE_MULTITENANT, true},
         {CFG_BOOL_ENABLE_DB_DROP_IF_FULL, false},
         {CFG_INT_MAX_TEARDOWN_TIME, 1},
@@ -233,4 +234,3 @@ namespace MAT_NS_BEGIN
 
 }
 MAT_NS_END
-

lib/include/public/ILogConfiguration.hpp

@@ -154,6 +154,12 @@ namespace MAT_NS_BEGIN
     /// </summary>
     static constexpr const char* const CFG_INT_RAM_QUEUE_BUFFERS = "maxDBFlushQueues";
 
+    /// <summary>
+    /// Batch records when flushing the RAM queue to disk storage.
+    /// Set to false to use per-record disk stores during flush.
+    /// </summary>
+    static constexpr const char* const CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH = "enableBatchedStorageFlush";
+
     /// <summary>
     /// SQLite DB will be checkpointed when flushing.
     /// </summary>
@@ -481,4 +487,3 @@ namespace MAT_NS_BEGIN
 }
 MAT_NS_END
 #endif
-

lib/jni/PrivacyGuard_jni.cpp

@@ -62,16 +62,22 @@ Java_com_microsoft_applications_events_PrivacyGuard_nativeInitializePrivacyGuard
     InitializationConfiguration config(
             reinterpret_cast<ILogger*>(iLoggerNativePtr),
             CommonDataContext{});
+    // InitializationConfiguration holds const char* pointers, so the backing
+    // std::string storage must outlive the PrivacyGuard construction below.
+    std::string notificationEventName, semanticContextEventName, summaryEventName;
     if (NotificationEventName != nullptr) {
-        config.NotificationEventName = JStringToStdString(env, NotificationEventName).c_str();
+        notificationEventName = JStringToStdString(env, NotificationEventName);
+        config.NotificationEventName = notificationEventName.c_str();
     }
 
     if (SemanticContextEventName != nullptr) {
-        config.SemanticContextNotificationEventName = JStringToStdString(env, SemanticContextEventName).c_str();
+        semanticContextEventName = JStringToStdString(env, SemanticContextEventName);
+        config.SemanticContextNotificationEventName = semanticContextEventName.c_str();
     }
 
     if (SummaryEventName != nullptr) {
-        config.SummaryEventName = JStringToStdString(env, SummaryEventName).c_str();
+        summaryEventName = JStringToStdString(env, SummaryEventName);
+        config.SummaryEventName = summaryEventName.c_str();
     }
 
     config.UseEventFieldPrefix = static_cast<bool>(UseEventFieldPrefix);
@@ -119,16 +125,22 @@ Java_com_microsoft_applications_events_PrivacyGuard_nativeInitializePrivacyGuard
                                             machineIds,
                                             outOfScopeIdentifiers));
 
+    // InitializationConfiguration holds const char* pointers, so the backing
+    // std::string storage must outlive the PrivacyGuard construction below.
+    std::string notificationEventName, semanticContextEventName, summaryEventName;
     if (NotificationEventName != NULL) {
-        config.NotificationEventName = JStringToStdString(env, NotificationEventName).c_str();
+        notificationEventName = JStringToStdString(env, NotificationEventName);
+        config.NotificationEventName = notificationEventName.c_str();
     }
 
     if (SemanticContextEventName != NULL) {
-        config.SemanticContextNotificationEventName = JStringToStdString(env, SemanticContextEventName).c_str();
+        semanticContextEventName = JStringToStdString(env, SemanticContextEventName);
+        config.SemanticContextNotificationEventName = semanticContextEventName.c_str();
     }
 
     if (SummaryEventName != NULL) {
-        config.SummaryEventName = JStringToStdString(env, SummaryEventName).c_str();
+        summaryEventName = JStringToStdString(env, SummaryEventName);
+        config.SummaryEventName = summaryEventName.c_str();
     }
 
     config.UseEventFieldPrefix = static_cast<bool>(UseEventFieldPrefix);

lib/offline/MemoryStorage.cpp

@@ -224,6 +224,16 @@ namespace MAT_NS_BEGIN {
 
     void MemoryStorage::DeleteRecords(const std::map<std::string, std::string> & whereFilter)
     {
+        // An empty filter matches every record. Never silently wipe the whole
+        // in-memory queue from a no-op predicate; callers must use
+        // DeleteAllRecords() for an intentional full clear. This mirrors the
+        // fail-closed behavior of OfflineStorage_SQLite::DeleteRecords.
+        if (whereFilter.empty())
+        {
+            LOG_WARN("DeleteRecords called with an empty filter; ignoring to avoid deleting all records.");
+            return;
+        }
+
         auto matcher = [&](const StorageRecord &r, const std::map<std::string, std::string> & whereFilter)
         {
             bool matched = true;

lib/offline/OfflineStorageHandler.cpp

@@ -9,6 +9,7 @@
 #include "offline/MemoryStorage.hpp"
 
 #include "ILogManager.hpp"
+#include "utils/Utils.hpp"
 #include <algorithm>
 #include <numeric>
 #include <set>
@@ -174,28 +175,42 @@ namespace MAT_NS_BEGIN {
         // than the handle gets replaced by nullptr in this DeferredCallbackHandle obj.
         m_flushHandle.Cancel();
 
-        size_t dbSizeBeforeFlush = m_offlineStorageMemory->GetSize();
+        size_t dbSizeBeforeFlush = (m_offlineStorageMemory != nullptr) ? m_offlineStorageMemory->GetSize() : 0;
         if ((m_offlineStorageMemory) && (dbSizeBeforeFlush > 0) && (m_offlineStorageDisk))
         {
-            // This will block on and then take a lock for the duration of this move, and
-            // StoreRecord() will then block until the move completes.
+            // Drain the in-memory queue into a local batch. Records are removed
+            // from memory here; any that fail to persist below are re-inserted, so
+            // a disk write failure does not silently lose events. Draining (rather
+            // than reserving) keeps only a single copy of each record in flight and
+            // avoids stamping a reservation lease that the Room backend would
+            // persist to disk.
             auto records = m_offlineStorageMemory->GetRecords(false, EventLatency_Unspecified);
-            std::vector<StorageRecordId> ids;
 
-            // TODO: [MG] - consider running the batch in transaction
-            //            if (sqlite)
-            //                sqlite->Execute("BEGIN");
-
-            size_t totalSaved = m_offlineStorageDisk->StoreRecords(records);
-
-            // TODO: [MG] - consider running the batch in transaction
-            //            if (sqlite)
-            //                sqlite->Execute("END");
-
-            // Delete records from reserved on flush
-            HttpHeaders dummy;
-            bool fromMemory = true;
-            m_offlineStorageMemory->DeleteRecords(ids, dummy, fromMemory);
+            size_t totalSaved = 0;
+            if (IsBatchedStorageFlushEnabled())
+            {
+                // Persist the drained batch to disk in a single transaction.
+                // StoreRecords() commits as many records as it durably can and
+                // returns that count. Records it can never store (e.g. ones failing
+                // validation, reported separately) are dropped from the batch rather
+                // than counted, so a return of 0 with records still queued means a
+                // transient failure committed nothing -- return those records to the
+                // in-memory queue for retry. No events are lost, and a rolled-back
+                // batch leaves nothing on disk, so re-queuing cannot create duplicates
+                // (the events table has no unique record_id constraint). A non-zero
+                // count means those records are durably stored; do not re-queue.
+                totalSaved = m_offlineStorageDisk->StoreRecords(records);
+                if (totalSaved == 0 && !records.empty())
+                {
+                    LOG_WARN("Flush: disk store failed for the batch of %zu records; returning to the queue for retry",
+                        records.size());
+                    ReturnRecordsToMemory(records);
+                }
+            }
+            else
+            {
+                totalSaved = StoreRecordsIndividually(records);
+            }
 
             // Notify event listener about the records cached
             OnStorageRecordsSaved(totalSaved);
@@ -210,7 +225,7 @@ namespace MAT_NS_BEGIN {
         }
 
         // Checkpoint DB
-        if (m_config.HasConfig(CFG_BOOL_CHECKPOINT_DB_ON_FLUSH) && m_config[CFG_BOOL_CHECKPOINT_DB_ON_FLUSH]) 
+        if (m_offlineStorageDisk && m_config.HasConfig(CFG_BOOL_CHECKPOINT_DB_ON_FLUSH) && m_config[CFG_BOOL_CHECKPOINT_DB_ON_FLUSH])
         {
             m_offlineStorageDisk->Flush();
         }
@@ -244,7 +259,12 @@ namespace MAT_NS_BEGIN {
                 // are selected and removed from the cache (but will
                 // not block for the subsequent handoff to persistent
                 // storage)
-                m_offlineStorageMemory->StoreRecord(record);
+                if (!m_offlineStorageMemory->StoreRecord(record))
+                {
+                    LOG_ERROR("Failed to store event %s:%s in memory queue",
+                        tenantTokenToId(record.tenantToken).c_str(), record.id.c_str());
+                    return false;
+                }
             }
 
             // Perform periodic flush to disk
@@ -269,14 +289,105 @@ namespace MAT_NS_BEGIN {
             {
                 if (record.persistence != EventPersistence::EventPersistence_DoNotStoreOnDisk)
                 {
-                    m_offlineStorageDisk->StoreRecord(record);
+                    // Propagate a synchronous disk write failure to the caller so a
+                    // failed store is not counted as successfully persisted.
+                    return m_offlineStorageDisk->StoreRecord(record);
                 }
             }
         }
 
         return true;
     }
 
+    bool OfflineStorageHandler::IsBatchedStorageFlushEnabled()
+    {
+        return !m_config.HasConfig(CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH) ||
+            m_config[CFG_BOOL_ENABLE_BATCHED_STORAGE_FLUSH];
+    }
+
+    bool OfflineStorageHandler::IsValidDiskRecord(StorageRecord const& record)
+    {
+        return !(record.id.empty() || record.tenantToken.empty() ||
+            static_cast<int>(record.latency) < 0 || record.timestamp <= 0);
+    }
+
+    void OfflineStorageHandler::ReportInvalidDiskRecord(StorageRecord const& record)
+    {
+        LOG_ERROR("Flush: dropping event %s:%s: Invalid parameters",
+            tenantTokenToId(record.tenantToken).c_str(), record.id.c_str());
+        OnStorageFailed("Invalid parameters");
+    }
+
+    size_t OfflineStorageHandler::StoreRecordsIndividually(std::vector<StorageRecord> const& records)
+    {
+        size_t totalSaved = 0;
+        std::vector<StorageRecord> recordsToRetry;
+
+        for (auto it = records.begin(); it != records.end(); ++it)
+        {
+            if (!IsValidDiskRecord(*it))
+            {
+                ReportInvalidDiskRecord(*it);
+                continue;
+            }
+
+            if (m_offlineStorageDisk->StoreRecord(*it))
+            {
+                ++totalSaved;
+                continue;
+            }
+
+            for (auto retryIt = it; retryIt != records.end(); ++retryIt)
+            {
+                if (IsValidDiskRecord(*retryIt))
+                {
+                    recordsToRetry.push_back(*retryIt);
+                }
+                else
+                {
+                    ReportInvalidDiskRecord(*retryIt);
+                }
+            }
+            break;
+        }
+
+        if (!recordsToRetry.empty())
+        {
+            LOG_WARN("Flush: per-record disk store failed after saving %zu of %zu records; returning %zu records to the queue for retry",
+                totalSaved, records.size(), recordsToRetry.size());
+            ReturnRecordsToMemory(recordsToRetry);
+        }
+
+        return totalSaved;
+    }
+
+    size_t OfflineStorageHandler::ReturnRecordsToMemory(std::vector<StorageRecord> const& records)
+    {
+        size_t returned = 0;
+        DroppedMap dropped;
+
+        for (auto const& record : records)
+        {
+            if (m_offlineStorageMemory && m_offlineStorageMemory->StoreRecord(record))
+            {
+                ++returned;
+            }
+            else
+            {
+                LOG_ERROR("Flush: failed to return event %s:%s to memory queue after disk store failure; dropping record",
+                    tenantTokenToId(record.tenantToken).c_str(), record.id.c_str());
+                dropped[record.tenantToken]++;
+            }
+        }
+
+        if (!dropped.empty())
+        {
+            OnStorageRecordsDropped(dropped);
+        }
+
+        return returned;
+    }
+
     size_t OfflineStorageHandler::StoreRecords(std::vector<StorageRecord>& records)
     {
         size_t stored = 0;

lib/offline/OfflineStorageHandler.hpp

@@ -25,6 +25,8 @@ namespace MAT_NS_BEGIN {
 
     class OfflineStorageHandler final : public IOfflineStorage, public IOfflineStorageObserver
     {
+        friend class OfflineStorageHandlerTestPeer;
+
     public:
         OfflineStorageHandler(ILogManager& logManager, IRuntimeConfig& runtimeConfig, ITaskDispatcher& taskDispatcher);
         virtual ~OfflineStorageHandler() override;
@@ -85,8 +87,8 @@ namespace MAT_NS_BEGIN {
         std::unique_ptr<IOfflineStorage>       m_offlineStorageMemory;
         std::shared_ptr<IOfflineStorage>       m_offlineStorageDisk;
 
-        bool                                   m_readFromMemory;
-        unsigned                               m_lastReadCount;
+        std::atomic<bool>                      m_readFromMemory;
+        std::atomic<unsigned>                  m_lastReadCount;
 
         bool                                   m_shutdownStarted;
         unsigned                               m_memoryDbSize;
@@ -99,6 +101,11 @@ namespace MAT_NS_BEGIN {
 
     private:
         void WaitForFlush();
+        bool IsBatchedStorageFlushEnabled();
+        bool IsValidDiskRecord(StorageRecord const& record);
+        void ReportInvalidDiskRecord(StorageRecord const& record);
+        size_t StoreRecordsIndividually(std::vector<StorageRecord> const& records);
+        size_t ReturnRecordsToMemory(std::vector<StorageRecord> const& records);
 
     };