diff --git a/services/libs/data-access-layer/src/activities/insert.test.ts b/services/libs/data-access-layer/src/activities/insert.test.ts index c8082870a4..5f0988bfac 100644 --- a/services/libs/data-access-layer/src/activities/insert.test.ts +++ b/services/libs/data-access-layer/src/activities/insert.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' -import { trimAttributeStrings } from './insert' +import { serializeAttributes, trimAttributeStrings } from './insert' describe('trimAttributeStrings', () => { it('keeps every key of an oversized attributes object instead of wiping it', () => { @@ -46,3 +46,126 @@ describe('trimAttributeStrings', () => { expect(trimmed).toBe(`${'a'.repeat(499)}...`) }) }) + +describe('serializeAttributes', () => { + const MAX_BYTES = 16 * 1024 + const expectUnderCap = (json: string) => + expect(Buffer.byteLength(json)).toBeLessThanOrEqual(MAX_BYTES) + + it('serializes attributes under the cap exactly as before, with only the per-string trim', () => { + const result = serializeAttributes({ + state: 'open', + additions: 12, + merged: false, + missing: null, + labels: [{ name: 'bug' }], + comment: 'x'.repeat(600), + }) + + expect(result.pruned).toBe(false) + expect(result.json).toBe( + `{"state":"open","additions":12,"merged":false,"missing":null,"labels":[{"name":"bug"}],"comment":"${'x'.repeat(500)}..."}`, + ) + }) + + it('shrinks the largest value and leaves small attributes like isIndirectFork intact', () => { + const attributes = { + state: 'open', + count: 3, + isIndirectFork: true, + url: 'https://github.com/org/repo/pull/1', + bigNumbers: Array(5000).fill(123456), + mediumNumbers: Array(1000).fill(1), + } + + const result = serializeAttributes(attributes) + const parsed = JSON.parse(result.json) + + expect(result.pruned).toBe(true) + expect(result.originalBytes).toBeGreaterThan(MAX_BYTES) + expectUnderCap(result.json) + expect(parsed.bigNumbers.length).toBeLessThan(5000) + expect(parsed.bigNumbers).toEqual(attributes.bigNumbers.slice(0, parsed.bigNumbers.length)) + expect(parsed).toEqual({ ...attributes, bigNumbers: parsed.bigNumbers }) + }) + + it('drops the largest strings first, alphabetically on ties, before small scalars', () => { + const attributes: Record = Object.fromEntries( + Array.from({ length: 40 }, (_, i) => [`s${String(i).padStart(2, '0')}`, 's'.repeat(1000)]), + ) + attributes.isIndirectFork = true + + const parsed = JSON.parse(serializeAttributes(attributes).json) + + expect(parsed.isIndirectFork).toBe(true) + expect(parsed.s00).toBeUndefined() + expect(parsed.s39).toBe(`${'s'.repeat(500)}...`) + }) + + it('shrinks a single nested value over the cap instead of dropping it', () => { + const result = serializeAttributes({ + payload: { items: Array.from({ length: 3000 }, (_, i) => ({ i })), note: 'n' }, + }) + const parsed = JSON.parse(result.json) + + expectUnderCap(result.json) + expect(parsed.payload.note).toBe('n') + expect(parsed.payload.items.length).toBeGreaterThan(1000) + }) + + it('keeps a top-level array an array, dropping items from the end', () => { + const result = serializeAttributes( + Array.from({ length: 3000 }, (_, i) => ({ id: i, name: `item-${i}` })), + ) + const parsed = JSON.parse(result.json) + + expectUnderCap(result.json) + expect(Array.isArray(parsed)).toBe(true) + expect(parsed[0]).toEqual({ id: 0, name: 'item-0' }) + }) + + it('measures the cap in UTF-8 bytes, not characters', () => { + const result = serializeAttributes( + Object.fromEntries(Array.from({ length: 20 }, (_, i) => [`k${i}`, 'é'.repeat(500)])), + ) + + expect(result.pruned).toBe(true) + expectUnderCap(result.json) + }) + + it('gives the same output regardless of input key order', () => { + const build = (order: string[]) => + Object.fromEntries(order.map((k) => [k, Array(3000).fill(1)])) + + const a = serializeAttributes(build(['b', 'a', 'c'])).json + const reordered = serializeAttributes(build(['c', 'b', 'a'])).json + + expect(serializeAttributes(build(['b', 'a', 'c'])).json).toBe(a) + expect(JSON.parse(reordered)).toEqual(JSON.parse(a)) + expect(JSON.parse(a).a.length).toBeLessThan(3000) + expect(JSON.parse(a).b).toHaveLength(3000) + }) + + it('always returns valid JSON under the cap for randomly shaped oversized input', () => { + let seed = 42 + const rand = () => { + seed = (seed * 1103515245 + 12345) % 2147483648 + return seed / 2147483648 + } + const gen = (depth: number): unknown => { + const r = rand() + if (depth > 2 || r < 0.3) return 'ü'.repeat(Math.floor(rand() * 800)) + if (r < 0.4) return r < 0.35 ? undefined : Math.floor(rand() * 1e9) + if (r < 0.7) return Array.from({ length: Math.floor(rand() * 40) }, () => gen(depth + 1)) + return Object.fromEntries( + Array.from({ length: Math.floor(rand() * 40) }, (_, i) => [`key-${i}`, gen(depth + 1)]), + ) + } + + for (let i = 0; i < 30; i++) { + const { json } = serializeAttributes({ root: gen(0), other: gen(1) }) + expectUnderCap(json) + expect(() => JSON.parse(json)).not.toThrow() + } + }) +}) diff --git a/services/libs/data-access-layer/src/activities/insert.ts b/services/libs/data-access-layer/src/activities/insert.ts index a5d5651fd7..6927b9c892 100644 --- a/services/libs/data-access-layer/src/activities/insert.ts +++ b/services/libs/data-access-layer/src/activities/insert.ts @@ -72,6 +72,20 @@ export async function insertActivities( const toInsert = activities .map((activity) => { const id = activity.id || generateUUIDv4() + const attributes = serializeAttributes(tryToUnwrapAttributes(activity.attributes)) + if (attributes.pruned) { + logger.warn( + { + activityId: id, + platform: activity.platform, + type: activity.type, + sourceId: activity.sourceId, + originalBytes: attributes.originalBytes, + maxBytes: ATTRIBUTES_MAX_BYTES, + }, + 'Activity attributes exceeded the size cap and were pruned!', + ) + } return { // we keep these ones in front of `...activity` because these fields might exist in the activity object @@ -92,7 +106,7 @@ export async function insertActivities( updatedAt: now, createdAt: activity.createdAt ? moment(activity.createdAt).toISOString() : now, timestamp: activity.timestamp ? moment(activity.timestamp).toISOString() : now, - attributes: objectToBytes(tryToUnwrapAttributes(activity.attributes)), + attributes: attributes.json, body: truncateWithDots(activity.body, 2000), // This is being added here temporarily, just to ensure we keep sending isContribution to Kafka @@ -141,12 +155,26 @@ export async function insertActivities( const ATTRIBUTE_STRING_MAX_LENGTH = 500 -function objectToBytes(input: object): string { - if (typeof input !== 'object') { - return input +// keeps a 10-message Kafka batch far below the 1 MB broker default +const ATTRIBUTES_MAX_BYTES = 16 * 1024 + +export function serializeAttributes(input: object): { + json: string + pruned: boolean + originalBytes: number +} { + const trimmed = trimAttributeStrings(input ?? {}) as object + const json = JSON.stringify(trimmed) + const originalBytes = Buffer.byteLength(json) + if (originalBytes <= ATTRIBUTES_MAX_BYTES) { + return { json, pruned: false, originalBytes } } - return JSON.stringify(trimAttributeStrings(input ?? {})) + return { + json: JSON.stringify(pruneToFit(trimmed, ATTRIBUTES_MAX_BYTES)), + pruned: true, + originalBytes, + } } export function trimAttributeStrings(value: unknown): unknown { @@ -162,6 +190,52 @@ export function trimAttributeStrings(value: unknown): unknown { return value } +// Largest entries go first (ties: object keys alphabetically, array items from the end); containers shrink from the +// inside, anything else is dropped. +function pruneToFit(value: object, maxBytes: number): object { + const isArray = Array.isArray(value) + const entries = Object.entries(value).map(([key, v]) => { + const valueJson = JSON.stringify(v) + const valueBytes = valueJson === undefined ? (isArray ? 4 : 0) : Buffer.byteLength(valueJson) + const size = + isArray || valueJson === undefined + ? valueBytes + : Buffer.byteLength(JSON.stringify(key)) + 1 + valueBytes + return { key, value: v as unknown, valueBytes, size, keep: true } + }) + + const present = entries.filter((e) => e.size > 0) + let total = 2 + present.reduce((sum, e) => sum + e.size, 0) + Math.max(present.length - 1, 0) + let count = present.length + + const tieBreak = isArray + ? (a: { key: string }, b: { key: string }) => Number(b.key) - Number(a.key) + : (a: { key: string }, b: { key: string }) => (a.key < b.key ? -1 : a.key > b.key ? 1 : 0) + const bySizeDesc = [...present].sort((a, b) => b.size - a.size || tieBreak(a, b)) + + for (const entry of bySizeDesc) { + if (total <= maxBytes) { + break + } + + const excess = total - maxBytes + if (entry.value && typeof entry.value === 'object' && entry.valueBytes - excess >= 2) { + entry.value = pruneToFit(entry.value, entry.valueBytes - excess) + const newBytes = Buffer.byteLength(JSON.stringify(entry.value)) + total -= entry.valueBytes - newBytes + entry.valueBytes = newBytes + continue + } + + entry.keep = false + total -= entry.size + (count > 1 ? 1 : 0) + count -= 1 + } + + const kept = entries.filter((e) => e.keep) + return isArray ? kept.map((e) => e.value) : Object.fromEntries(kept.map((e) => [e.key, e.value])) +} + function tryToUnwrapAttributes(attributes: string | object): object { // eslint-disable-next-line no-constant-condition while (true) {