fix(core): resolve sparse array and rolling size accumulator regressions in event chunking (#1309)

packages/core/src/__tests__/util.test.ts

@@ -42,6 +42,76 @@ describe('#chunk', () => {
       chunk([about500bString, about500bString, about500bString], 2, 1)
     ).toEqual([[about500bString, about500bString], [about500bString]]);
   });
+
+  it('handles oversized first item without creating a sparse array', () => {
+    const oversizedString = 'x'.repeat(2000); // ~2KB > 1KB maxKB
+    const result = chunk([oversizedString, 'small1', 'small2'], 5, 1);
+
+    expect(0 in result).toBe(true);
+    expect(result.length).toBe(2);
+    expect(result).toEqual([[oversizedString], ['small1', 'small2']]);
+  });
+
+  it('resets rolling size accumulator when starting a new chunk', () => {
+    const item400b = 'x'.repeat(400); // ~0.4KB
+    const result = chunk([item400b, item400b, item400b, item400b], 5, 1);
+
+    expect(result.length).toBe(2);
+    expect(result).toEqual([
+      [item400b, item400b],
+      [item400b, item400b],
+    ]);
+  });
+
+  it('handles multiple consecutive oversized items', () => {
+    const oversizedString = 'x'.repeat(2000);
+    const result = chunk(
+      [oversizedString, oversizedString, 'small1', 'small2'],
+      5,
+      1
+    );
+
+    expect(result).toEqual([
+      [oversizedString],
+      [oversizedString],
+      ['small1', 'small2'],
+    ]);
+  });
+
+  it('handles oversized item in the middle of normal items', () => {
+    const oversizedString = 'x'.repeat(2000);
+    const result = chunk(
+      ['small1', 'small2', oversizedString, 'small3'],
+      5,
+      1
+    );
+
+    expect(result).toEqual([
+      ['small1', 'small2'],
+      [oversizedString],
+      ['small3'],
+    ]);
+  });
+
+  it('handles exact max kb boundary', () => {
+    const halfKBString = 'a'.repeat(510); // exactly 0.5 KB (512 bytes with JSON quotes)
+    const result = chunk([halfKBString, halfKBString, halfKBString], 5, 1);
+
+    expect(result).toEqual([
+      [halfKBString, halfKBString],
+      [halfKBString],
+    ]);
+  });
+
+  it('handles single item exactly at max kb boundary', () => {
+    const exact1KBString = 'a'.repeat(1022); // exactly 1.0 KB (1024 bytes with JSON quotes)
+    const result = chunk([exact1KBString, 'small'], 5, 1);
+
+    expect(result).toEqual([
+      [exact1KBString],
+      ['small'],
+    ]);
+  });
 });
 
 describe('allSettled', () => {

packages/core/src/plugins/__tests__/SegmentDestination.test.ts

@@ -347,6 +347,55 @@ describe('SegmentDestination', () => {
       });
     });
 
+    it('flushes successfully when first event exceeds max payload size', async () => {
+      const oversizedPayload = 'x'.repeat(600 * 1024); // > 500 KB MAX_PAYLOAD_SIZE_IN_KB
+      const events = [
+        {
+          messageId: 'message-oversized',
+          type: EventType.TrackEvent,
+          event: 'Oversized Event',
+          properties: { data: oversizedPayload },
+        },
+        {
+          messageId: 'message-normal',
+          type: EventType.TrackEvent,
+          event: 'Normal Event',
+        },
+      ] as SegmentEvent[];
+
+      const { plugin, sendEventsSpy } = createTestWith({
+        events: events,
+      });
+
+      await expect(plugin.flush()).resolves.not.toThrow();
+      expect(sendEventsSpy).toHaveBeenCalledTimes(2);
+      expect(sendEventsSpy).toHaveBeenCalledWith({
+        url: getURL(defaultApiHost, ''),
+        writeKey: '123-456',
+        retryCount: 0,
+        events: [
+          {
+            messageId: 'message-oversized',
+            type: EventType.TrackEvent,
+            event: 'Oversized Event',
+            properties: { data: oversizedPayload },
+          },
+        ],
+      });
+      expect(sendEventsSpy).toHaveBeenCalledWith({
+        url: getURL(defaultApiHost, ''),
+        writeKey: '123-456',
+        retryCount: 0,
+        events: [
+          {
+            messageId: 'message-normal',
+            type: EventType.TrackEvent,
+            event: 'Normal Event',
+          },
+        ],
+      });
+    });
+
     it('uses segment settings apiHost for uploading events', async () => {
       const customEndpoint = 'events.eu1.segmentapis.com';
       const events = [

packages/core/src/util.ts

@@ -29,34 +29,34 @@ export const chunk = <T>(array: T[], count: number, maxKB?: number): T[][] => {
     return [];
   }
 
-  let currentChunk = 0;
+  const chunks: T[][] = [];
+  let currentChunk: T[] = [];
   let rollingKBSize = 0;
-  const result: T[][] = array.reduce(
-    (chunks: T[][], item: T, index: number) => {
-      if (maxKB !== undefined) {
-        rollingKBSize += sizeOf(item);
-        // If we overflow chunk until the previous index, else keep going
-        if (rollingKBSize >= maxKB) {
-          chunks[++currentChunk] = [item];
-          return chunks;
-        }
-      }
-
-      if (index !== 0 && index % count === 0) {
-        chunks[++currentChunk] = [item];
-      } else {
-        if (chunks[currentChunk] === undefined) {
-          chunks[currentChunk] = [];
-        }
-        chunks[currentChunk].push(item);
-      }
-
-      return chunks;
-    },
-    []
-  );
 
-  return result;
+  for (const item of array) {
+    const itemKB = maxKB !== undefined ? sizeOf(item) : 0;
+
+    const exceedsCount = currentChunk.length >= count;
+    const exceedsSize =
+      maxKB !== undefined &&
+      currentChunk.length > 0 &&
+      rollingKBSize + itemKB > maxKB;
+
+    if (exceedsCount || exceedsSize) {
+      chunks.push(currentChunk);
+      currentChunk = [];
+      rollingKBSize = 0;
+    }
+
+    currentChunk.push(item);
+    rollingKBSize += itemKB;
+  }
+
+  if (currentChunk.length > 0) {
+    chunks.push(currentChunk);
+  }
+
+  return chunks;
 };
 
 export const getAllPlugins = (timeline: Timeline) => {