[AAP-73135] Revert to sync_mode after batch-limit approach proven insufficient

Testing showed the batch-limit heuristic still dropped events:
- 17 chunks in one batch (~425 KB tracked): only 12 arrived
- 26 chunks split into two batches: only 17 arrived

Root cause: the SDK adds ~2-3 KB of per-event metadata (context,
timestamps, messageId, integrations) that our data-size estimate did
not account for, pushing actual batch bodies over Segment's 500 KB
limit despite our 450 KB threshold.

sync_mode=True sends each track() as a separate blocking HTTP request
(~25 KB each) instead of batching, which eliminates the batch-size
problem entirely. Local end-to-end testing confirmed all 15 chunks
arrive reliably with sync_mode=True.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

Author: cshiels-ie

metrics_utility/library/storage/segment.py

@@ -21,12 +21,6 @@ class StorageSegment:
     # including `properties` wrapper, event name, and segment_meta; keep this conservative.
     REGULAR_MESSAGE_LIMIT = 24 * 1024
 
-    # Segment enforces a 500KB limit per batch POST. We flush before reaching it,
-    # leaving headroom for per-event metadata the SDK adds (anonymousId, timestamp,
-    # context, etc.). EVENT_OVERHEAD is a conservative per-event estimate.
-    BATCH_SIZE_LIMIT = 450 * 1024
-    EVENT_OVERHEAD = 512
-
     def __init__(self, **settings):
         self.debug = settings.get('debug', False)
         self.user_id = settings.get('user_id', 'unknown')
@@ -136,6 +130,11 @@ def put(self, artifact_name, *, filename=None, fileobj=None, dict=None, event_na
         # Configure Segment client
         analytics.write_key = self.write_key
         analytics.debug = self.debug
+        # sync_mode makes each track() a blocking HTTP request instead of queuing to a
+        # background thread. Without it the SDK batches all chunks into one POST which
+        # can silently exceed Segment's 500 KB batch limit and drop events, returning
+        # HTTP 200 with no error callback fired.
+        analytics.sync_mode = True
 
         max_size = self.REGULAR_MESSAGE_LIMIT
         chunks = self._split_into_chunks(dict, max_size)
@@ -150,17 +149,9 @@ def put(self, artifact_name, *, filename=None, fileobj=None, dict=None, event_na
             segment_meta = {}
         message_id = segment_meta.get('message_id', None)
 
-        # Send each chunk, flushing before the batch would exceed Segment's 500KB limit
-        batch_bytes = 0
+        # Send each chunk
         for i, chunk in enumerate(chunks, 1):
             chunk_size = self._calculate_size(chunk)
-            event_size = chunk_size + self.EVENT_OVERHEAD
-
-            if batch_bytes + event_size > self.BATCH_SIZE_LIMIT and batch_bytes > 0:
-                if self.debug:
-                    print(f'Flushing batch at {batch_bytes} bytes before adding chunk {i}', file=sys.stderr)
-                analytics.flush()
-                batch_bytes = 0
 
             # chunk hash = sha256(message hash + chunk index)
             if message_id:
@@ -187,9 +178,8 @@ def put(self, artifact_name, *, filename=None, fileobj=None, dict=None, event_na
                 },
                 **segment_meta,
             )
-            batch_bytes += event_size
 
-        # Flush remaining queued events
+        # Flush to ensure all events are sent
         analytics.flush()
 
         return chunks

metrics_utility/test/library/test_storage_segment.py

@@ -133,7 +133,6 @@ def test_put_sends_multiple_chunks_for_large_data(self, mock_analytics):
         # Should split into 7 chunks as tested earlier
         assert len(chunks) == 7
         assert mock_analytics.track.call_count == 7
-        # 7 chunks well under 450 KB batch limit — only the final flush should fire
         assert mock_analytics.flush.call_count == 1
 
         # Verify chunk numbering in the calls
@@ -145,43 +144,20 @@ def test_put_sends_multiple_chunks_for_large_data(self, mock_analytics):
 
     @patch('metrics_utility.library.storage.segment.analytics')
     @patch('metrics_utility.library.storage.segment.SEGMENT_AVAILABLE', True)
-    def test_put_single_flush_when_chunks_fit_in_one_batch(self, mock_analytics):
-        """All chunks are sent in one batch when total size is below BATCH_SIZE_LIMIT."""
-        mock_analytics.track = Mock()
-        mock_analytics.flush = Mock()
-
-        storage_segment = StorageSegment(write_key='test_write_key', debug=False)
-        chunks = storage_segment.put(
-            artifact_name='test_artifact',
-            dict=segment_data_large,
-            event_name='Test Event',
-        )
+    def test_put_sync_mode_enabled(self, mock_analytics):
+        """sync_mode is set to True so each track() is a blocking HTTP request.
 
-        assert mock_analytics.track.call_count == len(chunks)
-        assert mock_analytics.flush.call_count == 1
-
-    @patch('metrics_utility.library.storage.segment.analytics')
-    @patch('metrics_utility.library.storage.segment.SEGMENT_AVAILABLE', True)
-    def test_put_flushes_mid_loop_when_batch_limit_reached(self, mock_analytics):
-        """flush() is called before queuing a chunk that would push the batch over BATCH_SIZE_LIMIT.
-
-        Segment silently drops events from batch POSTs that exceed 500 KB and returns
-        HTTP 200, so we flush proactively before reaching that threshold.
+        Without sync_mode the SDK batches all chunks into one POST which can silently
+        exceed Segment's 500 KB batch limit and drop events with no error raised.
         """
         mock_analytics.track = Mock()
         mock_analytics.flush = Mock()
 
         storage_segment = StorageSegment(write_key='test_write_key', debug=False)
-        # Lower the limit so each large (~24 KB) chunk forces its own batch
-        storage_segment.BATCH_SIZE_LIMIT = storage_segment.REGULAR_MESSAGE_LIMIT + storage_segment.EVENT_OVERHEAD
-
-        chunks = storage_segment.put(
+        storage_segment.put(
             artifact_name='test_artifact',
             dict=segment_data_large,
             event_name='Test Event',
         )
 
-        # Every chunk must still be tracked — none dropped
-        assert mock_analytics.track.call_count == len(chunks)
-        # Multiple flushes confirm that mid-loop splitting occurred
-        assert mock_analytics.flush.call_count > 1
+        assert mock_analytics.sync_mode is True