From d45668f4a4240881ddea39a8dc8d5a4c3426ef24 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Thu, 15 Oct 2026 22:00:45 +0200 Subject: [PATCH 1/3] fix: cap total activity attributes size after per-string trim MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/activities/insert.test.ts | 96 ++++++++++++++++++- .../src/activities/insert.ts | 76 ++++++++++++++- 2 files changed, 166 insertions(+), 6 deletions(-) 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..63dfaa7978 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 { ATTRIBUTES_MAX_BYTES, serializeAttributes, trimAttributeStrings } from './insert' describe('trimAttributeStrings', () => { it('keeps every key of an oversized attributes object instead of wiping it', () => { @@ -46,3 +46,97 @@ describe('trimAttributeStrings', () => { expect(trimmed).toBe(`${'a'.repeat(499)}...`) }) }) + +describe('serializeAttributes', () => { + const bytes = (s: string) => Buffer.byteLength(s, 'utf8') + + it('leaves attributes under the cap unchanged apart from the per-string trim', () => { + const attributes = { + labels: Array.from({ length: 50 }, (_, i) => ({ name: `label-${i}`, color: 'ff0000' })), + state: 'open', + additions: 12, + merged: false, + missing: null, + comment: 'x'.repeat(600), + } + + expect(JSON.parse(serializeAttributes(attributes))).toEqual(trimAttributeStrings(attributes)) + }) + + it('tightens the per-string limit before dropping anything', () => { + // 60 strings of 500 chars ≈ 31 KB after the default trim, ≈ 12.5 KB at 200 chars + const attributes = Object.fromEntries( + Array.from({ length: 60 }, (_, i) => [`field${i}`, 'a'.repeat(1000)]), + ) + + const json = serializeAttributes(attributes) + const parsed = JSON.parse(json) as Record + + expect(bytes(json)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) + expect(Object.keys(parsed)).toEqual(Object.keys(attributes)) + expect(parsed.field0).toBe(`${'a'.repeat(200)}...`) + }) + + it('drops the largest top-level attributes first when shorter strings are not enough', () => { + const attributes = { + state: 'open', + count: 3, + bigNumbers: Array(5000).fill(123456), + mediumNumbers: Array(1000).fill(1), + small: { nested: ['x'.repeat(30)] }, + } + + const json = serializeAttributes(attributes) + const parsed = JSON.parse(json) + + expect(bytes(json)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) + expect(parsed).toEqual({ + state: 'open', + count: 3, + mediumNumbers: attributes.mediumNumbers, + small: { nested: ['x'.repeat(30)] }, + }) + }) + + it('keeps nested objects and arrays intact, with strings trimmed, when shortening strings is enough', () => { + const attributes = { + review: Array.from({ length: 40 }, (_, i) => ({ id: i, text: 'r'.repeat(900), ok: true })), + tags: Array.from({ length: 40 }, () => 't'.repeat(900)), + } + + const json = serializeAttributes(attributes) + const parsed = JSON.parse(json) as typeof attributes + + expect(bytes(json)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) + expect(parsed.review).toHaveLength(40) + expect(parsed.review[0]).toEqual({ id: 0, text: `${'r'.repeat(100)}...`, ok: true }) + expect(parsed.tags).toHaveLength(40) + expect(parsed.tags[0]).toBe(`${'t'.repeat(100)}...`) + }) + + it('measures the cap in UTF-8 bytes, not characters', () => { + const attributes = Object.fromEntries( + Array.from({ length: 20 }, (_, i) => [`k${i}`, 'é'.repeat(500)]), + ) + + const json = serializeAttributes(attributes) + + // 20 × 500 two-byte chars ≈ 20 KB, over the cap even though it is ~10k characters + expect(bytes(json)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) + expect(Object.keys(JSON.parse(json))).toHaveLength(20) + }) + + it('is deterministic, including the tie-break between equally large attributes', () => { + const build = (order: string[]) => + Object.fromEntries(order.map((k) => [k, Array(3000).fill(1)])) + + const a = serializeAttributes(build(['b', 'a', 'c'])) + const b = serializeAttributes(build(['b', 'a', 'c'])) + const reordered = JSON.parse(serializeAttributes(build(['c', 'b', 'a']))) + + expect(a).toBe(b) + expect(bytes(a)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) + expect(Object.keys(JSON.parse(a)).sort()).toEqual(['b', 'c']) + expect(Object.keys(reordered).sort()).toEqual(['b', 'c']) + }) +}) diff --git a/services/libs/data-access-layer/src/activities/insert.ts b/services/libs/data-access-layer/src/activities/insert.ts index a5d5651fd7..7028c604d2 100644 --- a/services/libs/data-access-layer/src/activities/insert.ts +++ b/services/libs/data-access-layer/src/activities/insert.ts @@ -141,27 +141,93 @@ export async function insertActivities( const ATTRIBUTE_STRING_MAX_LENGTH = 500 +// Upper bound on the serialized (UTF-8) size of an activity's attributes. Every platform goes through the same cap so +// a single activity with many attributes can't bloat the Kafka message or the Tinybird row. 16 KiB is 8x the old +// 2000-byte QuestDB-era limit and keeps a 10-message Kafka batch far below the 1 MB broker default. +export const ATTRIBUTES_MAX_BYTES = 16 * 1024 + +// Per-string limits tried, in order, when the attributes are still over ATTRIBUTES_MAX_BYTES after the default trim. +const ATTRIBUTE_STRING_FALLBACK_LENGTHS = [200, 100, 50] + function objectToBytes(input: object): string { if (typeof input !== 'object') { return input } - return JSON.stringify(trimAttributeStrings(input ?? {})) + return serializeAttributes(input ?? {}) +} + +/** + * Serializes activity attributes to JSON no larger than ATTRIBUTES_MAX_BYTES. + * 1. Every string is trimmed to ATTRIBUTE_STRING_MAX_LENGTH characters; under the cap that is the result. + * 2. Otherwise strings are trimmed to progressively shorter lengths, keeping every key. + * 3. If still over, top-level attributes are dropped largest first (ties by key) until the result fits. + */ +export function serializeAttributes(attributes: object): string { + let json = JSON.stringify(trimAttributeStrings(attributes)) + if (byteLength(json) <= ATTRIBUTES_MAX_BYTES) { + return json + } + + let trimmed: unknown = attributes + for (const maxLength of ATTRIBUTE_STRING_FALLBACK_LENGTHS) { + trimmed = trimAttributeStrings(attributes, maxLength) + json = JSON.stringify(trimmed) + if (byteLength(json) <= ATTRIBUTES_MAX_BYTES) { + return json + } + } + + return JSON.stringify(dropLargestEntries(trimmed as object, ATTRIBUTES_MAX_BYTES)) } -export function trimAttributeStrings(value: unknown): unknown { +export function trimAttributeStrings( + value: unknown, + maxLength: number = ATTRIBUTE_STRING_MAX_LENGTH, +): unknown { if (typeof value === 'string') { - return truncateWithDots(value, ATTRIBUTE_STRING_MAX_LENGTH) + return truncateWithDots(value, maxLength) } if (Array.isArray(value)) { - return value.map(trimAttributeStrings) + return value.map((item) => trimAttributeStrings(item, maxLength)) } if (value && typeof value === 'object') { - return Object.fromEntries(Object.entries(value).map(([k, v]) => [k, trimAttributeStrings(v)])) + return Object.fromEntries( + Object.entries(value).map(([k, v]) => [k, trimAttributeStrings(v, maxLength)]), + ) } return value } +function dropLargestEntries(value: object, maxBytes: number): object { + const isArray = Array.isArray(value) + const entries = Object.entries(value) + const bySizeDesc = entries + .map(([key, v]) => ({ key, size: byteLength(JSON.stringify(v) ?? '') })) + .sort((a, b) => b.size - a.size || (a.key < b.key ? -1 : a.key > b.key ? 1 : 0)) + + const dropped = new Set() + const rebuild = (): object => { + const kept = entries.filter(([key]) => !dropped.has(key)) + return isArray ? kept.map(([, v]) => v) : Object.fromEntries(kept) + } + + let result = rebuild() + for (const { key } of bySizeDesc) { + if (byteLength(JSON.stringify(result)) <= maxBytes) { + break + } + dropped.add(key) + result = rebuild() + } + + return result +} + +function byteLength(s: string): number { + return Buffer.byteLength(s, 'utf8') +} + function tryToUnwrapAttributes(attributes: string | object): object { // eslint-disable-next-line no-constant-condition while (true) { From 0695aebf275b0483781aa03fc33becd5bdc5bc1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Thu, 15 Oct 2026 22:18:38 +0200 Subject: [PATCH 2/3] fix: prune oversized attributes from the inside and log when capped MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/activities/insert.test.ts | 158 ++++++++++++++---- .../src/activities/insert.ts | 124 ++++++++++---- 2 files changed, 221 insertions(+), 61 deletions(-) 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 63dfaa7978..d8be951c9d 100644 --- a/services/libs/data-access-layer/src/activities/insert.test.ts +++ b/services/libs/data-access-layer/src/activities/insert.test.ts @@ -49,6 +49,7 @@ describe('trimAttributeStrings', () => { describe('serializeAttributes', () => { const bytes = (s: string) => Buffer.byteLength(s, 'utf8') + const fitsCap = (json: string) => expect(bytes(json)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) it('leaves attributes under the cap unchanged apart from the per-string trim', () => { const attributes = { @@ -60,58 +61,125 @@ describe('serializeAttributes', () => { comment: 'x'.repeat(600), } - expect(JSON.parse(serializeAttributes(attributes))).toEqual(trimAttributeStrings(attributes)) + const { json, degradation } = serializeAttributes(attributes) + + expect(degradation).toBe('none') + expect(JSON.parse(json)).toEqual(trimAttributeStrings(attributes)) }) - it('tightens the per-string limit before dropping anything', () => { + it('tightens the per-string limit before pruning anything', () => { // 60 strings of 500 chars ≈ 31 KB after the default trim, ≈ 12.5 KB at 200 chars const attributes = Object.fromEntries( Array.from({ length: 60 }, (_, i) => [`field${i}`, 'a'.repeat(1000)]), ) - const json = serializeAttributes(attributes) + const { json, degradation } = serializeAttributes(attributes) const parsed = JSON.parse(json) as Record - expect(bytes(json)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) + fitsCap(json) + expect(degradation).toBe('strings-shortened') expect(Object.keys(parsed)).toEqual(Object.keys(attributes)) expect(parsed.field0).toBe(`${'a'.repeat(200)}...`) }) - it('drops the largest top-level attributes first when shorter strings are not enough', () => { + it('keeps nested objects and arrays, with strings trimmed, when shortening strings is enough', () => { + const attributes = { + review: Array.from({ length: 40 }, (_, i) => ({ id: i, text: 'r'.repeat(900), ok: true })), + tags: Array.from({ length: 40 }, () => 't'.repeat(900)), + } + + const { json } = serializeAttributes(attributes) + const parsed = JSON.parse(json) as typeof attributes + + fitsCap(json) + expect(parsed.review).toHaveLength(40) + expect(parsed.review[0]).toEqual({ id: 0, text: `${'r'.repeat(100)}...`, ok: true }) + expect(parsed.tags).toHaveLength(40) + expect(parsed.tags[0]).toBe(`${'t'.repeat(100)}...`) + }) + + it('prunes the largest value from the inside and leaves smaller attributes alone', () => { const attributes = { state: 'open', count: 3, + isIndirectFork: true, bigNumbers: Array(5000).fill(123456), mediumNumbers: Array(1000).fill(1), small: { nested: ['x'.repeat(30)] }, } - const json = serializeAttributes(attributes) + const { json, degradation } = serializeAttributes(attributes) const parsed = JSON.parse(json) - expect(bytes(json)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) - expect(parsed).toEqual({ - state: 'open', - count: 3, - mediumNumbers: attributes.mediumNumbers, - small: { nested: ['x'.repeat(30)] }, - }) + fitsCap(json) + expect(degradation).toBe('values-pruned') + expect(parsed.bigNumbers.length).toBeGreaterThan(0) + 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('keeps nested objects and arrays intact, with strings trimmed, when shortening strings is enough', () => { + it('shrinks a single nested value that alone exceeds the cap instead of wiping it', () => { const attributes = { - review: Array.from({ length: 40 }, (_, i) => ({ id: i, text: 'r'.repeat(900), ok: true })), - tags: Array.from({ length: 40 }, () => 't'.repeat(900)), + payload: { items: Array.from({ length: 3000 }, (_, i) => ({ i })), note: 'n' }, } - const json = serializeAttributes(attributes) - const parsed = JSON.parse(json) as typeof attributes + const { json } = serializeAttributes(attributes) + const parsed = JSON.parse(json) - expect(bytes(json)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) - expect(parsed.review).toHaveLength(40) - expect(parsed.review[0]).toEqual({ id: 0, text: `${'r'.repeat(100)}...`, ok: true }) - expect(parsed.tags).toHaveLength(40) - expect(parsed.tags[0]).toBe(`${'t'.repeat(100)}...`) + fitsCap(json) + expect(parsed.payload.note).toBe('n') + expect(parsed.payload.items.length).toBeGreaterThan(1000) + }) + + it('drops the largest strings first, ties by key, keeping small scalars', () => { + // 400 strings × ~60 bytes even at the 50-character limit ≈ 24 KB + const attributes: Record = Object.fromEntries( + Array.from({ length: 400 }, (_, i) => [`s${String(i).padStart(3, '0')}`, 's'.repeat(1000)]), + ) + attributes.isIndirectFork = true + + const { json } = serializeAttributes(attributes) + const parsed = JSON.parse(json) + + fitsCap(json) + expect(parsed.isIndirectFork).toBe(true) + expect(parsed.s000).toBeUndefined() + expect(parsed.s399).toBe(`${'s'.repeat(50)}...`) + }) + + it('caps a top-level array', () => { + const attributes = Array.from({ length: 3000 }, (_, i) => ({ id: i, name: `item-${i}` })) + + const { json } = serializeAttributes(attributes) + const parsed = JSON.parse(json) + + fitsCap(json) + expect(Array.isArray(parsed)).toBe(true) + expect(parsed.length).toBeGreaterThan(0) + // equal-sized items are dropped from the end, so the head of the list survives in order + expect(parsed[0]).toEqual({ id: 0, name: 'item-0' }) + }) + + it('handles undefined, function and Date values like JSON.stringify does', () => { + const date = new Date('2026-01-02T03:04:05.000Z') + const attributes = { + missing: undefined, + fn: () => 1, + when: date, + list: [undefined, 1], + big: Array(5000).fill(123456), + } + + const small = JSON.parse(serializeAttributes({ when: date, list: [undefined, 1] }).json) + const { json } = serializeAttributes(attributes) + const parsed = JSON.parse(json) + + expect(small).toEqual({ when: date.toISOString(), list: [null, 1] }) + fitsCap(json) + expect(parsed.when).toBe(date.toISOString()) + expect(parsed.list).toEqual([null, 1]) + expect('missing' in parsed || 'fn' in parsed).toBe(false) }) it('measures the cap in UTF-8 bytes, not characters', () => { @@ -119,24 +187,50 @@ describe('serializeAttributes', () => { Array.from({ length: 20 }, (_, i) => [`k${i}`, 'é'.repeat(500)]), ) - const json = serializeAttributes(attributes) + const { json } = serializeAttributes(attributes) // 20 × 500 two-byte chars ≈ 20 KB, over the cap even though it is ~10k characters - expect(bytes(json)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) + fitsCap(json) expect(Object.keys(JSON.parse(json))).toHaveLength(20) }) - it('is deterministic, including the tie-break between equally large attributes', () => { + it('is deterministic, independent 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'])) - const b = serializeAttributes(build(['b', 'a', 'c'])) - const reordered = JSON.parse(serializeAttributes(build(['c', 'b', 'a']))) + const a = serializeAttributes(build(['b', 'a', 'c'])).json + const b = serializeAttributes(build(['b', 'a', 'c'])).json + const reordered = serializeAttributes(build(['c', 'b', 'a'])).json expect(a).toBe(b) - expect(bytes(a)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) - expect(Object.keys(JSON.parse(a)).sort()).toEqual(['b', 'c']) - expect(Object.keys(reordered).sort()).toEqual(['b', 'c']) + fitsCap(a) + expect(JSON.parse(reordered)).toEqual(JSON.parse(a)) + // equal sizes: the tie goes to the first key, so only `a` is shortened + expect(JSON.parse(a).a.length).toBeLessThan(3000) + expect(JSON.parse(a).b).toHaveLength(3000) + expect(JSON.parse(a).c).toHaveLength(3000) + }) + + it('stays 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 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) }) + fitsCap(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 7028c604d2..c00ea77d59 100644 --- a/services/libs/data-access-layer/src/activities/insert.ts +++ b/services/libs/data-access-layer/src/activities/insert.ts @@ -92,7 +92,19 @@ 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: objectToBytes(tryToUnwrapAttributes(activity.attributes), (degradation) => + logger.warn( + { + activityId: id, + platform: activity.platform, + type: activity.type, + sourceId: activity.sourceId, + degradation, + maxBytes: ATTRIBUTES_MAX_BYTES, + }, + 'Activity attributes exceeded the size cap and were reduced!', + ), + ), body: truncateWithDots(activity.body, 2000), // This is being added here temporarily, just to ensure we keep sending isContribution to Kafka @@ -149,36 +161,54 @@ export const ATTRIBUTES_MAX_BYTES = 16 * 1024 // Per-string limits tried, in order, when the attributes are still over ATTRIBUTES_MAX_BYTES after the default trim. const ATTRIBUTE_STRING_FALLBACK_LENGTHS = [200, 100, 50] -function objectToBytes(input: object): string { +export type AttributesDegradation = 'none' | 'strings-shortened' | 'values-pruned' + +function objectToBytes( + input: object, + onDegraded: (degradation: AttributesDegradation) => void, +): string { if (typeof input !== 'object') { return input } - return serializeAttributes(input ?? {}) + const { json, degradation } = serializeAttributes(input ?? {}) + if (degradation !== 'none') { + onDegraded(degradation) + } + return json } /** * Serializes activity attributes to JSON no larger than ATTRIBUTES_MAX_BYTES. * 1. Every string is trimmed to ATTRIBUTE_STRING_MAX_LENGTH characters; under the cap that is the result. * 2. Otherwise strings are trimmed to progressively shorter lengths, keeping every key. - * 3. If still over, top-level attributes are dropped largest first (ties by key) until the result fits. + * 3. If still over, the largest values are pruned (largest first; ties by key, array items from the end): a nested + * object or array is pruned + * the same way from the inside, anything else is dropped. Small scalars are the last to go. */ -export function serializeAttributes(attributes: object): string { - let json = JSON.stringify(trimAttributeStrings(attributes)) - if (byteLength(json) <= ATTRIBUTES_MAX_BYTES) { - return json +export function serializeAttributes(attributes: object): { + json: string + degradation: AttributesDegradation +} { + let trimmed = trimAttributeStrings(attributes) + let json = JSON.stringify(trimmed) + if (fitsCap(json)) { + return { json, degradation: 'none' } } - let trimmed: unknown = attributes for (const maxLength of ATTRIBUTE_STRING_FALLBACK_LENGTHS) { - trimmed = trimAttributeStrings(attributes, maxLength) + // trimming the already-trimmed copy gives the same result as trimming the original, on a smaller input + trimmed = trimAttributeStrings(trimmed, maxLength) json = JSON.stringify(trimmed) - if (byteLength(json) <= ATTRIBUTES_MAX_BYTES) { - return json + if (fitsCap(json)) { + return { json, degradation: 'strings-shortened' } } } - return JSON.stringify(dropLargestEntries(trimmed as object, ATTRIBUTES_MAX_BYTES)) + return { + json: JSON.stringify(pruneToFit(trimmed as object, ATTRIBUTES_MAX_BYTES)), + degradation: 'values-pruned', + } } export function trimAttributeStrings( @@ -192,6 +222,10 @@ export function trimAttributeStrings( return value.map((item) => trimAttributeStrings(item, maxLength)) } if (value && typeof value === 'object') { + // keep what JSON.stringify would produce for values like Date instead of turning them into {} + if (typeof (value as { toJSON?: unknown }).toJSON === 'function') { + return trimAttributeStrings((value as { toJSON: () => unknown }).toJSON(), maxLength) + } return Object.fromEntries( Object.entries(value).map(([k, v]) => [k, trimAttributeStrings(v, maxLength)]), ) @@ -199,29 +233,61 @@ export function trimAttributeStrings( return value } -function dropLargestEntries(value: object, maxBytes: number): object { +// A UTF-8 string is never more than 3 bytes per UTF-16 code unit, so short strings skip the byte count. +function fitsCap(json: string): boolean { + return json.length * 3 <= ATTRIBUTES_MAX_BYTES || byteLength(json) <= ATTRIBUTES_MAX_BYTES +} + +/** + * Returns a copy of a (JSON-safe) object or array whose serialization is at most maxBytes. Entry sizes are computed + * once and subtracted as entries are dropped, so the cost stays linear in the size of the input per nesting level. + */ +function pruneToFit(value: object, maxBytes: number): object { const isArray = Array.isArray(value) - const entries = Object.entries(value) - const bySizeDesc = entries - .map(([key, v]) => ({ key, size: byteLength(JSON.stringify(v) ?? '') })) - .sort((a, b) => b.size - a.size || (a.key < b.key ? -1 : a.key > b.key ? 1 : 0)) + const entries = Object.entries(value).map(([key, v]) => { + const valueJson = JSON.stringify(v) + // undefined / functions: skipped inside objects, serialized as null inside arrays + const valueBytes = valueJson === undefined ? (isArray ? 4 : 0) : byteLength(valueJson) + const size = + isArray || valueJson === undefined + ? valueBytes + : byteLength(JSON.stringify(key)) + 1 + valueBytes + return { key, value: v as unknown, valueBytes, size, keep: true } + }) - const dropped = new Set() - const rebuild = (): object => { - const kept = entries.filter(([key]) => !dropped.has(key)) - return isArray ? kept.map(([, v]) => v) : Object.fromEntries(kept) - } + 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 + + // ties: object keys alphabetically, array items from the end so the head of a list survives + 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)) - let result = rebuild() - for (const { key } of bySizeDesc) { - if (byteLength(JSON.stringify(result)) <= maxBytes) { + for (const entry of bySizeDesc) { + if (total <= maxBytes) { break } - dropped.add(key) - result = rebuild() + + const excess = total - maxBytes + const child = entry.value + if (child && typeof child === 'object' && entry.valueBytes - excess >= 2) { + // shrink the nested container just enough instead of losing all of it + entry.value = pruneToFit(child, entry.valueBytes - excess) + const newBytes = 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 } - return result + const kept = entries.filter((e) => e.keep) + return isArray ? kept.map((e) => e.value) : Object.fromEntries(kept.map((e) => [e.key, e.value])) } function byteLength(s: string): number { From 527268a32f2c637075aa983e23c5687b1e23c482 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Uro=C5=A1=20Marolt?= Date: Thu, 15 Oct 2026 22:34:01 +0200 Subject: [PATCH 3/3] fix: simplify the attributes size cap to trim, measure and prune MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Refs linuxfoundation/lfx-cdp-insights-ops#748 Signed-off-by: Uroš Marolt --- .../src/activities/insert.test.ts | 161 ++++++------------ .../src/activities/insert.ts | 136 +++++---------- 2 files changed, 87 insertions(+), 210 deletions(-) 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 d8be951c9d..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 { ATTRIBUTES_MAX_BYTES, serializeAttributes, trimAttributeStrings } from './insert' +import { serializeAttributes, trimAttributeStrings } from './insert' describe('trimAttributeStrings', () => { it('keeps every key of an oversized attributes object instead of wiping it', () => { @@ -48,170 +48,105 @@ describe('trimAttributeStrings', () => { }) describe('serializeAttributes', () => { - const bytes = (s: string) => Buffer.byteLength(s, 'utf8') - const fitsCap = (json: string) => expect(bytes(json)).toBeLessThanOrEqual(ATTRIBUTES_MAX_BYTES) + const MAX_BYTES = 16 * 1024 + const expectUnderCap = (json: string) => + expect(Buffer.byteLength(json)).toBeLessThanOrEqual(MAX_BYTES) - it('leaves attributes under the cap unchanged apart from the per-string trim', () => { - const attributes = { - labels: Array.from({ length: 50 }, (_, i) => ({ name: `label-${i}`, color: 'ff0000' })), + 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), - } + }) - const { json, degradation } = serializeAttributes(attributes) - - expect(degradation).toBe('none') - expect(JSON.parse(json)).toEqual(trimAttributeStrings(attributes)) - }) - - it('tightens the per-string limit before pruning anything', () => { - // 60 strings of 500 chars ≈ 31 KB after the default trim, ≈ 12.5 KB at 200 chars - const attributes = Object.fromEntries( - Array.from({ length: 60 }, (_, i) => [`field${i}`, 'a'.repeat(1000)]), + expect(result.pruned).toBe(false) + expect(result.json).toBe( + `{"state":"open","additions":12,"merged":false,"missing":null,"labels":[{"name":"bug"}],"comment":"${'x'.repeat(500)}..."}`, ) - - const { json, degradation } = serializeAttributes(attributes) - const parsed = JSON.parse(json) as Record - - fitsCap(json) - expect(degradation).toBe('strings-shortened') - expect(Object.keys(parsed)).toEqual(Object.keys(attributes)) - expect(parsed.field0).toBe(`${'a'.repeat(200)}...`) - }) - - it('keeps nested objects and arrays, with strings trimmed, when shortening strings is enough', () => { - const attributes = { - review: Array.from({ length: 40 }, (_, i) => ({ id: i, text: 'r'.repeat(900), ok: true })), - tags: Array.from({ length: 40 }, () => 't'.repeat(900)), - } - - const { json } = serializeAttributes(attributes) - const parsed = JSON.parse(json) as typeof attributes - - fitsCap(json) - expect(parsed.review).toHaveLength(40) - expect(parsed.review[0]).toEqual({ id: 0, text: `${'r'.repeat(100)}...`, ok: true }) - expect(parsed.tags).toHaveLength(40) - expect(parsed.tags[0]).toBe(`${'t'.repeat(100)}...`) }) - it('prunes the largest value from the inside and leaves smaller attributes alone', () => { + 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), - small: { nested: ['x'.repeat(30)] }, } - const { json, degradation } = serializeAttributes(attributes) - const parsed = JSON.parse(json) + const result = serializeAttributes(attributes) + const parsed = JSON.parse(result.json) - fitsCap(json) - expect(degradation).toBe('values-pruned') - expect(parsed.bigNumbers.length).toBeGreaterThan(0) + 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('shrinks a single nested value that alone exceeds the cap instead of wiping it', () => { - const attributes = { - payload: { items: Array.from({ length: 3000 }, (_, i) => ({ i })), note: 'n' }, - } - - const { json } = serializeAttributes(attributes) - const parsed = JSON.parse(json) - - fitsCap(json) - expect(parsed.payload.note).toBe('n') - expect(parsed.payload.items.length).toBeGreaterThan(1000) - }) - - it('drops the largest strings first, ties by key, keeping small scalars', () => { - // 400 strings × ~60 bytes even at the 50-character limit ≈ 24 KB + it('drops the largest strings first, alphabetically on ties, before small scalars', () => { const attributes: Record = Object.fromEntries( - Array.from({ length: 400 }, (_, i) => [`s${String(i).padStart(3, '0')}`, 's'.repeat(1000)]), + Array.from({ length: 40 }, (_, i) => [`s${String(i).padStart(2, '0')}`, 's'.repeat(1000)]), ) attributes.isIndirectFork = true - const { json } = serializeAttributes(attributes) - const parsed = JSON.parse(json) + const parsed = JSON.parse(serializeAttributes(attributes).json) - fitsCap(json) expect(parsed.isIndirectFork).toBe(true) - expect(parsed.s000).toBeUndefined() - expect(parsed.s399).toBe(`${'s'.repeat(50)}...`) + expect(parsed.s00).toBeUndefined() + expect(parsed.s39).toBe(`${'s'.repeat(500)}...`) }) - it('caps a top-level array', () => { - const attributes = Array.from({ length: 3000 }, (_, i) => ({ id: i, name: `item-${i}` })) - - const { json } = serializeAttributes(attributes) - const parsed = JSON.parse(json) + 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) - fitsCap(json) - expect(Array.isArray(parsed)).toBe(true) - expect(parsed.length).toBeGreaterThan(0) - // equal-sized items are dropped from the end, so the head of the list survives in order - expect(parsed[0]).toEqual({ id: 0, name: 'item-0' }) + expectUnderCap(result.json) + expect(parsed.payload.note).toBe('n') + expect(parsed.payload.items.length).toBeGreaterThan(1000) }) - it('handles undefined, function and Date values like JSON.stringify does', () => { - const date = new Date('2026-01-02T03:04:05.000Z') - const attributes = { - missing: undefined, - fn: () => 1, - when: date, - list: [undefined, 1], - big: Array(5000).fill(123456), - } - - const small = JSON.parse(serializeAttributes({ when: date, list: [undefined, 1] }).json) - const { json } = serializeAttributes(attributes) - const parsed = JSON.parse(json) + 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) - expect(small).toEqual({ when: date.toISOString(), list: [null, 1] }) - fitsCap(json) - expect(parsed.when).toBe(date.toISOString()) - expect(parsed.list).toEqual([null, 1]) - expect('missing' in parsed || 'fn' in parsed).toBe(false) + 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 attributes = Object.fromEntries( - Array.from({ length: 20 }, (_, i) => [`k${i}`, 'é'.repeat(500)]), + const result = serializeAttributes( + Object.fromEntries(Array.from({ length: 20 }, (_, i) => [`k${i}`, 'é'.repeat(500)])), ) - const { json } = serializeAttributes(attributes) - - // 20 × 500 two-byte chars ≈ 20 KB, over the cap even though it is ~10k characters - fitsCap(json) - expect(Object.keys(JSON.parse(json))).toHaveLength(20) + expect(result.pruned).toBe(true) + expectUnderCap(result.json) }) - it('is deterministic, independent of input key order', () => { + 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 b = serializeAttributes(build(['b', 'a', 'c'])).json const reordered = serializeAttributes(build(['c', 'b', 'a'])).json - expect(a).toBe(b) - fitsCap(a) + expect(serializeAttributes(build(['b', 'a', 'c'])).json).toBe(a) expect(JSON.parse(reordered)).toEqual(JSON.parse(a)) - // equal sizes: the tie goes to the first key, so only `a` is shortened expect(JSON.parse(a).a.length).toBeLessThan(3000) expect(JSON.parse(a).b).toHaveLength(3000) - expect(JSON.parse(a).c).toHaveLength(3000) }) - it('stays under the cap for randomly shaped oversized input', () => { + it('always returns valid JSON under the cap for randomly shaped oversized input', () => { let seed = 42 const rand = () => { seed = (seed * 1103515245 + 12345) % 2147483648 @@ -220,7 +155,7 @@ describe('serializeAttributes', () => { 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 Math.floor(rand() * 1e9) + 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)]), @@ -229,7 +164,7 @@ describe('serializeAttributes', () => { for (let i = 0; i < 30; i++) { const { json } = serializeAttributes({ root: gen(0), other: gen(1) }) - fitsCap(json) + 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 c00ea77d59..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,19 +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), (degradation) => - logger.warn( - { - activityId: id, - platform: activity.platform, - type: activity.type, - sourceId: activity.sourceId, - degradation, - maxBytes: ATTRIBUTES_MAX_BYTES, - }, - 'Activity attributes exceeded the size cap and were reduced!', - ), - ), + attributes: attributes.json, body: truncateWithDots(activity.body, 2000), // This is being added here temporarily, just to ensure we keep sending isContribution to Kafka @@ -153,105 +155,52 @@ export async function insertActivities( const ATTRIBUTE_STRING_MAX_LENGTH = 500 -// Upper bound on the serialized (UTF-8) size of an activity's attributes. Every platform goes through the same cap so -// a single activity with many attributes can't bloat the Kafka message or the Tinybird row. 16 KiB is 8x the old -// 2000-byte QuestDB-era limit and keeps a 10-message Kafka batch far below the 1 MB broker default. -export const ATTRIBUTES_MAX_BYTES = 16 * 1024 - -// Per-string limits tried, in order, when the attributes are still over ATTRIBUTES_MAX_BYTES after the default trim. -const ATTRIBUTE_STRING_FALLBACK_LENGTHS = [200, 100, 50] - -export type AttributesDegradation = 'none' | 'strings-shortened' | 'values-pruned' - -function objectToBytes( - input: object, - onDegraded: (degradation: AttributesDegradation) => void, -): 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 - const { json, degradation } = serializeAttributes(input ?? {}) - if (degradation !== 'none') { - onDegraded(degradation) - } - return json -} - -/** - * Serializes activity attributes to JSON no larger than ATTRIBUTES_MAX_BYTES. - * 1. Every string is trimmed to ATTRIBUTE_STRING_MAX_LENGTH characters; under the cap that is the result. - * 2. Otherwise strings are trimmed to progressively shorter lengths, keeping every key. - * 3. If still over, the largest values are pruned (largest first; ties by key, array items from the end): a nested - * object or array is pruned - * the same way from the inside, anything else is dropped. Small scalars are the last to go. - */ -export function serializeAttributes(attributes: object): { +export function serializeAttributes(input: object): { json: string - degradation: AttributesDegradation + pruned: boolean + originalBytes: number } { - let trimmed = trimAttributeStrings(attributes) - let json = JSON.stringify(trimmed) - if (fitsCap(json)) { - return { json, degradation: 'none' } - } - - for (const maxLength of ATTRIBUTE_STRING_FALLBACK_LENGTHS) { - // trimming the already-trimmed copy gives the same result as trimming the original, on a smaller input - trimmed = trimAttributeStrings(trimmed, maxLength) - json = JSON.stringify(trimmed) - if (fitsCap(json)) { - return { json, degradation: 'strings-shortened' } - } + 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: JSON.stringify(pruneToFit(trimmed as object, ATTRIBUTES_MAX_BYTES)), - degradation: 'values-pruned', + json: JSON.stringify(pruneToFit(trimmed, ATTRIBUTES_MAX_BYTES)), + pruned: true, + originalBytes, } } -export function trimAttributeStrings( - value: unknown, - maxLength: number = ATTRIBUTE_STRING_MAX_LENGTH, -): unknown { +export function trimAttributeStrings(value: unknown): unknown { if (typeof value === 'string') { - return truncateWithDots(value, maxLength) + return truncateWithDots(value, ATTRIBUTE_STRING_MAX_LENGTH) } if (Array.isArray(value)) { - return value.map((item) => trimAttributeStrings(item, maxLength)) + return value.map(trimAttributeStrings) } if (value && typeof value === 'object') { - // keep what JSON.stringify would produce for values like Date instead of turning them into {} - if (typeof (value as { toJSON?: unknown }).toJSON === 'function') { - return trimAttributeStrings((value as { toJSON: () => unknown }).toJSON(), maxLength) - } - return Object.fromEntries( - Object.entries(value).map(([k, v]) => [k, trimAttributeStrings(v, maxLength)]), - ) + return Object.fromEntries(Object.entries(value).map(([k, v]) => [k, trimAttributeStrings(v)])) } return value } -// A UTF-8 string is never more than 3 bytes per UTF-16 code unit, so short strings skip the byte count. -function fitsCap(json: string): boolean { - return json.length * 3 <= ATTRIBUTES_MAX_BYTES || byteLength(json) <= ATTRIBUTES_MAX_BYTES -} - -/** - * Returns a copy of a (JSON-safe) object or array whose serialization is at most maxBytes. Entry sizes are computed - * once and subtracted as entries are dropped, so the cost stays linear in the size of the input per nesting level. - */ +// 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) - // undefined / functions: skipped inside objects, serialized as null inside arrays - const valueBytes = valueJson === undefined ? (isArray ? 4 : 0) : byteLength(valueJson) + const valueBytes = valueJson === undefined ? (isArray ? 4 : 0) : Buffer.byteLength(valueJson) const size = isArray || valueJson === undefined ? valueBytes - : byteLength(JSON.stringify(key)) + 1 + valueBytes + : Buffer.byteLength(JSON.stringify(key)) + 1 + valueBytes return { key, value: v as unknown, valueBytes, size, keep: true } }) @@ -259,7 +208,6 @@ function pruneToFit(value: object, maxBytes: number): object { let total = 2 + present.reduce((sum, e) => sum + e.size, 0) + Math.max(present.length - 1, 0) let count = present.length - // ties: object keys alphabetically, array items from the end so the head of a list survives 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) @@ -271,11 +219,9 @@ function pruneToFit(value: object, maxBytes: number): object { } const excess = total - maxBytes - const child = entry.value - if (child && typeof child === 'object' && entry.valueBytes - excess >= 2) { - // shrink the nested container just enough instead of losing all of it - entry.value = pruneToFit(child, entry.valueBytes - excess) - const newBytes = byteLength(JSON.stringify(entry.value)) + 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 @@ -290,10 +236,6 @@ function pruneToFit(value: object, maxBytes: number): object { return isArray ? kept.map((e) => e.value) : Object.fromEntries(kept.map((e) => [e.key, e.value])) } -function byteLength(s: string): number { - return Buffer.byteLength(s, 'utf8') -} - function tryToUnwrapAttributes(attributes: string | object): object { // eslint-disable-next-line no-constant-condition while (true) {