fix: cap the total size of activity attributes

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<string, unknown> = 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()
+    }
+  })
+})

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) {