diff --git a/packages/core/src/__tests__/util.test.ts b/packages/core/src/__tests__/util.test.ts index bdc03e049..1f8542464 100644 --- a/packages/core/src/__tests__/util.test.ts +++ b/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(10).fill(about500bString); + + expect(chunk(items, 2, 1)).toEqual([ + [about500bString, about500bString], + [about500bString, about500bString], + [about500bString, about500bString], + [about500bString, about500bString], + [about500bString, about500bString], + ]); + }); }); describe('allSettled', () => { diff --git a/packages/core/src/plugins/__tests__/SegmentDestination.test.ts b/packages/core/src/plugins/__tests__/SegmentDestination.test.ts index e187771a4..726f8bff8 100644 --- a/packages/core/src/plugins/__tests__/SegmentDestination.test.ts +++ b/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 = [ diff --git a/packages/core/src/util.ts b/packages/core/src/util.ts index 8a89a8a78..a0c959396 100644 --- a/packages/core/src/util.ts +++ b/packages/core/src/util.ts @@ -29,34 +29,29 @@ export const chunk = (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) => {