fix(core): build chunks by appending so oversized events don't stall …

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

@@ -42,6 +42,27 @@ describe('#chunk', () => {
       chunk([about500bString, about500bString, about500bString], 2, 1)
     ).toEqual([[about500bString, about500bString], [about500bString]]);
   });
+
+  it('returns a dense array when a single item exceeds max kb', () => {
+    const overMaxKBString = 'x'.repeat(2 * 1024);
+    const batches = chunk([overMaxKBString, 'b', 'c'], 100, 1);
+
+    expect(0 in batches).toBe(true);
+    expect(batches).toEqual([[overMaxKBString], ['b', 'c']]);
+  });
+
+  it('keeps honouring count after a max kb overflow', () => {
+    const about500bString = 'x'.repeat(500);
+    const items = new Array<string>(10).fill(about500bString);
+
+    expect(chunk(items, 2, 1)).toEqual([
+      [about500bString, about500bString],
+      [about500bString, about500bString],
+      [about500bString, about500bString],
+      [about500bString, about500bString],
+      [about500bString, about500bString],
+    ]);
+  });
 });
 
 describe('allSettled', () => {

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

@@ -347,6 +347,26 @@ describe('SegmentDestination', () => {
       });
     });
 
+    it('drains the queue when a single event exceeds the max payload size', async () => {
+      const events = [
+        { messageId: 'big', properties: { blob: 'x'.repeat(520 * 1024) } },
+        { messageId: 'small-1' },
+        { messageId: 'small-2' },
+      ] as unknown as SegmentEvent[];
+
+      const { plugin } = createTestWith({ events });
+
+      const dequeueSpy = jest.spyOn(
+        // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+        // @ts-ignore
+        plugin.queuePlugin,
+        'dequeueByMessageIds'
+      );
+
+      await expect(plugin.flush()).resolves.not.toThrow();
+      expect(dequeueSpy).toHaveBeenCalledWith(['big', 'small-1', 'small-2']);
+    });
+
     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,29 @@ export const chunk = <T>(array: T[], count: number, maxKB?: number): T[][] => {
     return [];
   }
 
-  let currentChunk = 0;
   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 array.reduce((chunks: T[][], item: T) => {
+    const itemKBSize = maxKB === undefined ? 0 : sizeOf(item);
+    const currentChunk = chunks[chunks.length - 1];
+    const isOverMaxKB =
+      maxKB !== undefined && rollingKBSize + itemKBSize >= maxKB;
+
+    // Always append rather than assign by index, otherwise an item that alone exceeds maxKB leaves a hole at index 0
+    if (
+      currentChunk === undefined ||
+      currentChunk.length >= count ||
+      isOverMaxKB
+    ) {
+      rollingKBSize = itemKBSize;
+      chunks.push([item]);
       return chunks;
-    },
-    []
-  );
+    }
 
-  return result;
+    rollingKBSize += itemKBSize;
+    currentChunk.push(item);
+    return chunks;
+  }, []);
 };
 
 export const getAllPlugins = (timeline: Timeline) => {