Limit upload batches by event count

lib/android_build/app/src/androidTest/java/com/microsoft/applications/events/maesdktest/LogManagerDDVUnitTest.java

@@ -578,6 +578,7 @@ public void getDefaultConfig() {
     assertThat(tpm, is(notNullValue()));
     assertThat(tpm, isA(ILogConfiguration.class));
     assertThat(tpm.getLong(LogConfigurationKey.CFG_INT_TPM_MAX_BLOB_BYTES), is(2L * 1024 * 1024));
+    assertThat(tpm.getLong(LogConfigurationKey.CFG_INT_TPM_MAX_EVENTS_PER_UPLOAD), is(1500L));
     assertThat(tpm.getLong(LogConfigurationKey.CFG_INT_TPM_MAX_RETRY), is(5L));
     assertThat(tpm.getBoolean(LogConfigurationKey.CFG_BOOL_TPM_CLOCK_SKEW_ENABLED), is(true));
     ILogConfiguration compat = defaultConfig.getLogConfiguration(LogConfigurationKey.CFG_MAP_COMPAT);

lib/android_build/maesdk/src/main/java/com/microsoft/applications/events/LogConfigurationKey.java

@@ -166,6 +166,8 @@ public enum LogConfigurationKey {
 
   CFG_INT_TPM_MAX_BLOB_BYTES("maxBlobSize", Long.class),
 
+  CFG_INT_TPM_MAX_EVENTS_PER_UPLOAD("maxEventsPerUpload", Long.class),
+
   CFG_BOOL_SESSION_RESET_ENABLED("sessionResetEnabled", Boolean.class);
 
   private String key;
@@ -184,4 +186,3 @@ public Class getValueType() {
     return valueType;
   }
 }
-

lib/config/RuntimeConfig_Default.hpp

@@ -68,6 +68,7 @@ namespace MAT_NS_BEGIN
         {CFG_MAP_TPM,
          {
              {CFG_INT_TPM_MAX_BLOB_BYTES, 2097152},
+             {CFG_INT_TPM_MAX_EVENTS_PER_UPLOAD, 1500},
              {CFG_INT_TPM_MAX_RETRY, 5},
              {CFG_BOOL_TPM_CLOCK_SKEW_ENABLED, true},
              {CFG_STR_TPM_BACKOFF, "E,3000,300000,2,1"},
@@ -233,4 +234,3 @@ namespace MAT_NS_BEGIN
 
 }
 MAT_NS_END
-

lib/include/public/ILogConfiguration.hpp

@@ -391,6 +391,11 @@ namespace MAT_NS_BEGIN
     /// </summary>
     static constexpr const char* const CFG_INT_TPM_MAX_BLOB_BYTES = "maxBlobSize";
 
+    /// <summary>
+    /// TPM configuration: maximum number of events per upload request. A value of 0 means unlimited.
+    /// </summary>
+    static constexpr const char* const CFG_INT_TPM_MAX_EVENTS_PER_UPLOAD = "maxEventsPerUpload";
+
     /// <summary>
     /// TPM configuration map
     /// </summary>
@@ -471,4 +476,3 @@ namespace MAT_NS_BEGIN
 }
 MAT_NS_END
 #endif
-

lib/tpm/TransmissionPolicyManager.cpp

@@ -41,6 +41,20 @@ namespace MAT_NS_BEGIN {
         return (a > b) ? (a - b) : (b - a);
     }
 
+    unsigned getMaxEventsPerUpload(IRuntimeConfig& config)
+    {
+        const int64_t configuredMaxCount = config[CFG_MAP_TPM][CFG_INT_TPM_MAX_EVENTS_PER_UPLOAD];
+        if (configuredMaxCount <= 0)
+        {
+            return 0;
+        }
+        if (configuredMaxCount > static_cast<int64_t>(std::numeric_limits<unsigned>::max()))
+        {
+            return std::numeric_limits<unsigned>::max();
+        }
+        return static_cast<unsigned>(configuredMaxCount);
+    }
+
     MATSDK_LOG_INST_COMPONENT_CLASS(TransmissionPolicyManager, "EventsSDK.TPM", "Events telemetry client - TransmissionPolicyManager class")
 
     TransmissionPolicyManager::TransmissionPolicyManager(ITelemetrySystem& system, ITaskDispatcher& taskDispatcher, IBandwidthController* bandwidthController) :
@@ -218,6 +232,7 @@ namespace MAT_NS_BEGIN {
 
         auto ctx = m_system.createEventsUploadContext();
         ctx->requestedMinLatency = m_runningLatency;
+        ctx->requestedMaxCount = getMaxEventsPerUpload(m_config);
         addUpload(ctx);
         initiateUpload(ctx);
     }
@@ -336,6 +351,7 @@ namespace MAT_NS_BEGIN {
         if (event->record.latency > EventLatency_RealTime) {
             auto ctx = m_system.createEventsUploadContext();
             ctx->requestedMinLatency = event->record.latency;
+            ctx->requestedMaxCount = getMaxEventsPerUpload(m_config);
             addUpload(ctx);
             initiateUpload(ctx);
             return;

tests/unittests/TransmissionPolicyManagerTests.cpp

@@ -10,11 +10,40 @@
 #include "common/MockIBandwidthController.hpp"
 #include "tpm/TransmissionPolicyManager.hpp"
 #include "TransmitProfiles.hpp"
+#include "offline/MemoryStorage.hpp"
+#include "offline/StorageObserver.hpp"
+#include "packager/Packager.hpp"
+#include "bond/All.hpp"
+#include "bond/generated/CsProtocol_writers.hpp"
+
+#include <limits>
 
 using namespace testing;
 using namespace MAT;
 
 
+class ScopedTpmMaxEventsPerUploadConfig {
+  public:
+    ScopedTpmMaxEventsPerUploadConfig(IRuntimeConfig& config, Variant value)
+      : m_config(config),
+        m_previousMaxCount(config[CFG_MAP_TPM][CFG_INT_TPM_MAX_EVENTS_PER_UPLOAD])
+    {
+        m_config[CFG_MAP_TPM][CFG_INT_TPM_MAX_EVENTS_PER_UPLOAD] = value;
+    }
+
+    ~ScopedTpmMaxEventsPerUploadConfig()
+    {
+        m_config[CFG_MAP_TPM][CFG_INT_TPM_MAX_EVENTS_PER_UPLOAD] = m_previousMaxCount;
+    }
+
+    ScopedTpmMaxEventsPerUploadConfig(const ScopedTpmMaxEventsPerUploadConfig&) = delete;
+    ScopedTpmMaxEventsPerUploadConfig& operator=(const ScopedTpmMaxEventsPerUploadConfig&) = delete;
+
+  private:
+    IRuntimeConfig& m_config;
+    Variant         m_previousMaxCount;
+};
+
 class TransmissionPolicyManager4Test : public TransmissionPolicyManager {
   public:
     TransmissionPolicyManager4Test(ITelemetrySystem& system, IBandwidthController* bandwidthController)
@@ -77,6 +106,7 @@ class TransmissionPolicyManagerTests : public StrictMock<Test> {
 
     RouteSink<TransmissionPolicyManagerTests, EventsUploadContextPtr const&> initiateUpload{this, &TransmissionPolicyManagerTests::resultInitiateUpload};
     RouteSink<TransmissionPolicyManagerTests>                                allUploadsFinished{this, &TransmissionPolicyManagerTests::resultAllUploadsFinished};
+    RouteSink<TransmissionPolicyManagerTests, EventsUploadContextPtr const&> packagedEvents{this, &TransmissionPolicyManagerTests::resultPackagedEvents};
 
   protected:
     TransmissionPolicyManagerTests()
@@ -88,6 +118,7 @@ class TransmissionPolicyManagerTests : public StrictMock<Test> {
 
     MOCK_METHOD1(resultInitiateUpload, void(EventsUploadContextPtr const &));
     MOCK_METHOD0(resultAllUploadsFinished, void());
+    MOCK_METHOD1(resultPackagedEvents, void(EventsUploadContextPtr const &));
 
     virtual void SetUp() override
     {
@@ -246,6 +277,27 @@ TEST_F(TransmissionPolicyManagerTests, ImmediateIncomingEventStartsUploadImmedia
 
     ASSERT_THAT(upload, NotNull());
     EXPECT_THAT(upload->requestedMinLatency, EventLatency_Max);
+    EXPECT_THAT(upload->requestedMaxCount, 1500u);
+}
+
+TEST_F(TransmissionPolicyManagerTests, ImmediateIncomingEventTreatsNonPositiveMaxEventCountAsUnlimited)
+{
+    auto& config = testing::getSystem().getConfig();
+    ScopedTpmMaxEventsPerUploadConfig maxEventsConfig(config, static_cast<int64_t>(-1));
+    UNREFERENCED_PARAMETER(maxEventsConfig);
+
+    tpm.paused(false);
+
+    auto event = new IncomingEventContext();
+    event->record.latency = EventLatency_Max;
+    EventsUploadContextPtr upload;
+    EXPECT_CALL(*this, resultInitiateUpload(_))
+        .WillOnce(SaveArg<0>(&upload));
+    tpm.eventArrived(event);
+
+    ASSERT_THAT(upload, NotNull());
+    EXPECT_THAT(upload->requestedMinLatency, EventLatency_Max);
+    EXPECT_THAT(upload->requestedMaxCount, 0u);
 }
 
 TEST_F(TransmissionPolicyManagerTests, UploadDoesNothingWhenPaused)
@@ -288,13 +340,125 @@ TEST_F(TransmissionPolicyManagerTests, UploadInitiatesUpload)
     EventsUploadContextPtr upload;
     EXPECT_CALL(*this, resultInitiateUpload(_))
         .WillOnce(SaveArg<0>(&upload));
-    tpm.uploadAsync(EventLatency_Normal);
+    tpm.uploadAsyncParent(EventLatency_Normal);
 
     EXPECT_THAT(tpm.uploadScheduled(), false);
     EXPECT_THAT(upload, NotNull());
+    EXPECT_THAT(upload->requestedMaxCount, 1500u);
     EXPECT_THAT(tpm.activeUploads(), Contains(upload));
 }
 
+TEST_F(TransmissionPolicyManagerTests, UploadUsesConfiguredMaxEventCount)
+{
+    auto& config = testing::getSystem().getConfig();
+    ScopedTpmMaxEventsPerUploadConfig maxEventsConfig(config, 3);
+    UNREFERENCED_PARAMETER(maxEventsConfig);
+
+    tpm.uploadScheduled(true);
+    tpm.paused(false);
+
+    EventsUploadContextPtr upload;
+    EXPECT_CALL(*this, resultInitiateUpload(_))
+        .WillOnce(SaveArg<0>(&upload));
+    tpm.uploadAsyncParent(EventLatency_Normal);
+
+    unsigned requestedMaxCount = upload ? upload->requestedMaxCount : 0;
+
+    ASSERT_THAT(upload, NotNull());
+    EXPECT_THAT(requestedMaxCount, 3u);
+}
+
+TEST_F(TransmissionPolicyManagerTests, UploadTreatsNonPositiveMaxEventCountAsUnlimited)
+{
+    auto& config = testing::getSystem().getConfig();
+    ScopedTpmMaxEventsPerUploadConfig maxEventsConfig(config, static_cast<int64_t>(-1));
+    UNREFERENCED_PARAMETER(maxEventsConfig);
+
+    tpm.uploadScheduled(true);
+    tpm.paused(false);
+
+    EventsUploadContextPtr upload;
+    EXPECT_CALL(*this, resultInitiateUpload(_))
+        .WillOnce(SaveArg<0>(&upload));
+    tpm.uploadAsyncParent(EventLatency_Normal);
+
+    ASSERT_THAT(upload, NotNull());
+    EXPECT_THAT(upload->requestedMaxCount, 0u);
+}
+
+TEST_F(TransmissionPolicyManagerTests, UploadCapsOversizedMaxEventCount)
+{
+    auto& config = testing::getSystem().getConfig();
+    int64_t oversizedMaxCount = static_cast<int64_t>(std::numeric_limits<unsigned>::max()) + 1;
+    ScopedTpmMaxEventsPerUploadConfig maxEventsConfig(config, oversizedMaxCount);
+    UNREFERENCED_PARAMETER(maxEventsConfig);
+
+    tpm.uploadScheduled(true);
+    tpm.paused(false);
+
+    EventsUploadContextPtr upload;
+    EXPECT_CALL(*this, resultInitiateUpload(_))
+        .WillOnce(SaveArg<0>(&upload));
+    tpm.uploadAsyncParent(EventLatency_Normal);
+
+    ASSERT_THAT(upload, NotNull());
+    EXPECT_THAT(upload->requestedMaxCount, std::numeric_limits<unsigned>::max());
+}
+
+TEST_F(TransmissionPolicyManagerTests, UploadPackagesNoMoreThanConfiguredMaxEventCount)
+{
+    auto& system = testing::getSystem();
+    auto& config = system.getConfig();
+    ScopedTpmMaxEventsPerUploadConfig maxEventsConfig(config, 3);
+    UNREFERENCED_PARAMETER(maxEventsConfig);
+
+    MemoryStorage storage(system.getLogManager(), config);
+    StorageObserver storageObserver(system, storage);
+    Packager packager(config);
+
+    storageObserver.retrievedEvent >> packager.addEventToPackage;
+    storageObserver.retrievalFinished >> packager.finalizePackage;
+    packager.packagedEvents >> packagedEvents;
+
+    ASSERT_THAT(storageObserver.start(), true);
+    for (unsigned i = 0; i < 10; ++i)
+    {
+        ::CsProtocol::Record sourceRecord;
+        sourceRecord.name = "testEvent" + std::to_string(i);
+        std::vector<uint8_t> recordBlob;
+        bond_lite::CompactBinaryProtocolWriter writer(recordBlob);
+        bond_lite::Serialize(writer, sourceRecord);
+
+        StorageRecord record(
+            "r" + std::to_string(i),
+            "tenant-token",
+            EventLatency_Normal,
+            EventPersistence_Normal,
+            1234567890 + i,
+            std::move(recordBlob));
+        ASSERT_THAT(storage.StoreRecord(record), true);
+    }
+
+    EventsUploadContextPtr packaged;
+    EXPECT_CALL(*this, resultInitiateUpload(_))
+        .WillOnce(Invoke([&storageObserver](EventsUploadContextPtr const& ctx) {
+            storageObserver.retrieveEvents(ctx);
+        }));
+    EXPECT_CALL(*this, resultPackagedEvents(_))
+        .WillOnce(SaveArg<0>(&packaged));
+    tpm.uploadScheduled(true);
+    tpm.paused(false);
+    tpm.uploadAsyncParent(EventLatency_Normal);
+
+    ASSERT_THAT(packaged, NotNull());
+    EXPECT_THAT(packaged->requestedMaxCount, 3u);
+    EXPECT_THAT(packaged->recordIdsAndTenantIds, SizeIs(3));
+    EXPECT_THAT(storage.LastReadRecordCount(), 3u);
+    EXPECT_THAT(storage.GetRecordCount(), 7u);
+    EXPECT_THAT(storage.GetReservedCount(), 3u);
+    EXPECT_THAT(packaged->fromMemory, true);
+}
+
 TEST_F(TransmissionPolicyManagerTests, EmptyUploadCeasesUploadingForRunningLatencyNormal)
 {
     auto upload = tpm.fakeActiveUpload(EventLatency_Normal);

wrappers/obj-c/ODWLogConfiguration.h

@@ -271,6 +271,11 @@ extern NSString * _Nonnull const ODWCFG_STR_TPM_BACKOFF;
 */
 extern NSString * _Nonnull const ODWCFG_INT_TPM_MAX_BLOB_BYTES;
 
+/*!
+ TPM configuration: maximum events per upload request. Set to 0 for unlimited.
+*/
+extern NSString * _Nonnull const ODWCFG_INT_TPM_MAX_EVENTS_PER_UPLOAD;
+
 /*!
  TPM configuration map
 */

wrappers/obj-c/ODWLogConfiguration.mm

@@ -277,6 +277,11 @@ The minimum time (ms) between storage full notifications.
 */
 NSString *const ODWCFG_INT_TPM_MAX_BLOB_BYTES = @"maxBlobSize";
 
+/*!
+ TPM configuration: maximum events per upload request. Set to 0 for unlimited.
+*/
+NSString *const ODWCFG_INT_TPM_MAX_EVENTS_PER_UPLOAD = @"maxEventsPerUpload";
+
 /*!
  TPM configuration map
 */