From 288cf61c9108cc295360ab4510ed935caac63046 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 11:10:01 +0000 Subject: [PATCH 01/11] feat(cdp): let a caller bound hog and liquid rendering with a deadline HogInputsService.buildInputs takes an optional deadline signal. A hog render and a liquid render stop at the earlier of their own limit and the deadline, and a passed deadline fails the inputs with the deadline error. The deadline records when it expires, so a render checks the clock as well as the signal. No caller passes a deadline yet, so no send changes. Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- .../cdp/services/hog-inputs.service.test.ts | 243 +++++++++++++++++- nodejs/src/cdp/services/hog-inputs.service.ts | 34 ++- nodejs/src/cdp/utils/deadline.ts | 32 +++ nodejs/src/cdp/utils/hog-exec.ts | 22 +- nodejs/src/cdp/utils/liquid.ts | 9 +- ...5ef197338f8ac69ac218fbbc3c489fc7872f7.json | 1 + 6 files changed, 325 insertions(+), 16 deletions(-) create mode 100644 nodejs/src/cdp/utils/deadline.ts create mode 100644 nodejs/tests/fixtures/hog-bytecode/0a7c9f2282f0c526ba09f04bf535ef197338f8ac69ac218fbbc3c489fc7872f7.json diff --git a/nodejs/src/cdp/services/hog-inputs.service.test.ts b/nodejs/src/cdp/services/hog-inputs.service.test.ts index 73f1c143c0aa..fcf98346420f 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.test.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.test.ts @@ -4,7 +4,9 @@ import { DateTime } from 'luxon' import { register } from 'prom-client' import { inspect } from 'util' +import { CyclotronInputType } from '~/cdp/schema/cyclotron' import { closeHub, createHub } from '~/common/utils/db/hub' +import { configureEventLoopYield, getEventLoopYieldThresholdMs } from '~/common/utils/event-loop-yield' import { createTestTeamFixture } from '~/tests/helpers/sql' import { Hub, Team } from '~/types' @@ -12,10 +14,11 @@ import { HOG_FILTERS_EXAMPLES } from '../_tests/examples' import { createHogExecutionGlobals, createHogFunction, insertIntegration } from '../_tests/fixtures' import { compileHog } from '../templates/compiler' import { HogFunctionInvocationGlobals, HogFunctionType } from '../types' +import { deadlineAfter } from '../utils/deadline' import { HogConfigurationError, classifyHogError } from '../utils/hog-error-classification' import { replaceIntegrationInputsWithRefs, signIntegrationRef } from '../utils/integration-input-refs' import { buildHogFunctionInvocations } from '../utils/invocation-utils' -import { LiquidRenderBudget } from '../utils/liquid' +import { LIQUID_RENDER_LIMIT_MS, LiquidRenderBudget } from '../utils/liquid' import { NATIVE_PUSH_TEMPLATE_ID } from '../utils/push-subscription-utils' import { HogInputsService, formatHogInput, formatLiquidInput, getAppIdentifierForPush } from './hog-inputs.service' import { serializeInvocation } from './job-queue/job-queue-kafka' @@ -297,6 +300,244 @@ describe('Hog Inputs', () => { expect(inputs.hog_templated).toMatchInlineSnapshot(`"event: "test""`) }) + const passedDeadline = (): AbortController => { + const deadline = new AbortController() + deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) + return deadline + } + + it.each([ + ['hog', { value: 'x', templating: 'hog' as const, bytecode: ['_H', 1, 32, 'x'] }], + ['liquid', { value: '{{ event.event }}', templating: 'liquid' as const }], + ])('renders no %s input once the deadline it was given has passed', async (_templating, input) => { + const onlyThatInput = createHogFunction({ ...hogFunction, inputs: { only: input }, inputs_schema: [] }) + + await expect( + hogInputsService.buildInputs(onlyThatInput, globals, undefined, undefined, passedDeadline().signal) + ).rejects.toThrow('The verification deadline passed') + }) + + it.each([ + [ + 'it is aborted', + (): { signal: AbortSignal; pass: () => void } => { + const deadline = new AbortController() + return { + signal: deadline.signal, + pass: () => + deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')), + } + }, + ], + [ + 'the clock passes it before its timer fires', + (): { signal: AbortSignal; pass: () => void } => ({ + signal: deadlineAfter(60_000), + pass: () => jest.advanceTimersByTime(60_000), + }), + ], + ])( + 'fails the inputs with the deadline error when the deadline passes while a hog input waits for its turn on the event loop because %s', + async (_how, deadlineFor) => { + const busyFor = 30_000 + const busy = await compileHog( + `while (toUnixTimestampMilli(now()) < event.properties.busy_until) { } return 'done'` + ) + const onlyHog = createHogFunction({ + ...hogFunction, + inputs: { only: { value: 'x', templating: 'hog', bytecode: busy } }, + inputs_schema: [], + }) + const mockedNow = jest.mocked(Date.now) + const fixedNow = mockedNow() + const threshold = getEventLoopYieldThresholdMs() + await new Promise((resolve) => setTimeout(resolve, 5)) + configureEventLoopYield(0) + jest.useFakeTimers({ + doNotFake: [ + 'hrtime', + 'nextTick', + 'queueMicrotask', + 'setImmediate', + 'clearImmediate', + 'setInterval', + 'clearInterval', + 'Date', + ], + }) + const nextTurn = (): Promise => new Promise((resolve) => setImmediate(resolve)) + try { + mockedNow.mockImplementation(() => Number(process.hrtime.bigint() / 1_000_000n)) + const startedAt = Date.now() + const deadline = deadlineFor() + let outcome: unknown + const busyForSeconds = { + ...globals, + event: { + ...globals.event, + properties: { ...globals.event.properties, busy_until: startedAt + busyFor }, + }, + } + void hogInputsService + .buildInputs(onlyHog, busyForSeconds, undefined, undefined, deadline.signal) + .then( + () => (outcome = 'rendered'), + (error: unknown) => (outcome = error) + ) + for (let turn = 0; jest.getTimerCount() === 0 && turn < 1000; turn++) { + await nextTurn() + } + expect(outcome).toBeUndefined() + + deadline.pass() + for (let turn = 0; outcome === undefined && turn < 1000; turn++) { + jest.runOnlyPendingTimers() + await nextTurn() + } + + expect(outcome).toMatchObject({ name: 'TimeoutError' }) + } finally { + mockedNow.mockReturnValue(fixedNow) + jest.useRealTimers() + configureEventLoopYield(threshold) + } + } + ) + + it.each<[string, Record]>([ + ['an input before the last', { next: { value: '{{ event.event }}', templating: 'liquid', order: 1 } }], + ['the last input', {}], + ])('fails the inputs once %s carries the clock past the deadline in its render', async (_which, after) => { + const deadlineMs = 50 + let now = performance.now() + const clock = jest.spyOn(performance, 'now').mockImplementation(() => now) + try { + const deadline = deadlineAfter(deadlineMs) + const slowThenNext = createHogFunction({ + ...hogFunction, + inputs: { + slow: { value: '{{ event.properties.slow }}', templating: 'liquid', order: 0 }, + ...after, + }, + inputs_schema: [], + }) + const properties = { ...globals.event.properties } + Object.defineProperty(properties, 'slow', { + enumerable: true, + get: () => { + now += 2 * deadlineMs + return 'slow' + }, + }) + + await expect( + hogInputsService.buildInputs( + slowThenNext, + { ...globals, event: { ...globals.event, properties } }, + undefined, + undefined, + deadline + ) + ).rejects.toMatchObject({ name: 'TimeoutError' }) + } finally { + clock.mockRestore() + } + }) + + it.each<[string, number, () => Promise]>([ + [ + 'hog', + 3000, + async () => ({ + value: 'busy', + templating: 'hog', + bytecode: await compileHog( + `while (toUnixTimestampMilli(now()) < event.properties.busy_until) { } return 'done'` + ), + }), + ], + [ + 'liquid', + LIQUID_RENDER_LIMIT_MS, + () => + Promise.resolve({ + value: '{% for a in (1..1000) %}{% for b in (1..1000) %}{% for c in (1..1000) %}x{% endfor %}{% endfor %}{% endfor %}', + templating: 'liquid', + }), + ], + ])( + 'stops a %s render at the deadline rather than at its own longer limit of %d ms', + async (_templating, ownLimitMs, busyInput) => { + const deadlineMs = 50 + const onlyBusy = createHogFunction({ + ...hogFunction, + inputs: { busy: await busyInput() }, + inputs_schema: [], + }) + let now = 0 + const clock = jest.spyOn(performance, 'now').mockImplementation(() => (now += 0.01)) + const mockedNow = jest.mocked(Date.now) + const fixedNow = mockedNow() + mockedNow.mockImplementation(() => Math.floor((now += 0.01))) + try { + const deadline = deadlineAfter(deadlineMs) + const busyLong = { + ...globals, + event: { + ...globals.event, + properties: { ...globals.event.properties, busy_until: 2 * ownLimitMs }, + }, + } + + await expect( + hogInputsService.buildInputs(onlyBusy, busyLong, undefined, undefined, deadline) + ).rejects.toMatchObject({ name: 'TimeoutError' }) + expect(now).toBeLessThan(ownLimitMs / 2) + } finally { + mockedNow.mockReturnValue(fixedNow) + clock.mockRestore() + } + } + ) + + it('fails a hog render that the deadline cuts short with the deadline error on a clock that reads whole milliseconds', async () => { + const deadlineMs = 200 + const busy = await compileHog( + `while (toUnixTimestampMilli(now()) < event.properties.busy_until) { } return 'done'` + ) + const onlyBusy = createHogFunction({ + ...hogFunction, + inputs: { busy: { value: 'busy', templating: 'hog', bytecode: busy } }, + inputs_schema: [], + }) + let now = 0.5 + const clock = jest.spyOn(performance, 'now').mockImplementation(() => now) + const mockedNow = jest.mocked(Date.now) + const fixedNow = mockedNow() + mockedNow.mockImplementation(() => { + now += 0.001 + return Math.floor(now) + }) + try { + const deadline = deadlineAfter(deadlineMs) + now = 0.9 + const busyLong = { + ...globals, + event: { + ...globals.event, + properties: { ...globals.event.properties, busy_until: 10 * deadlineMs }, + }, + } + + await expect( + hogInputsService.buildInputs(onlyBusy, busyLong, undefined, undefined, deadline) + ).rejects.toMatchObject({ name: 'TimeoutError' }) + } finally { + mockedNow.mockReturnValue(fixedNow) + clock.mockRestore() + } + }) + it('should template out liquid inputs', async () => { const inputs = await hogInputsService.buildInputs(hogFunction, globals) expect(inputs.liquid_templated).toMatchInlineSnapshot(`"event: "test""`) diff --git a/nodejs/src/cdp/services/hog-inputs.service.ts b/nodejs/src/cdp/services/hog-inputs.service.ts index 5c170af292f2..928e51cb5eb9 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.ts @@ -12,6 +12,7 @@ import { HogFunctionType, IntegrationType, } from '../types' +import { throwIfDeadlinePassed } from '../utils/deadline' import { EncryptedFields } from '../utils/encryption-utils' import { HogConfigurationError, isHogVMErrorKind, withBytecodeContract } from '../utils/hog-error-classification' import { execHog } from '../utils/hog-exec' @@ -62,7 +63,8 @@ export class HogInputsService { hogFunction: HogFunctionType, globals: HogFunctionInvocationGlobals, additionalInputs?: Record, - preferencesTokenSource?: PreferencesTokenSource + preferencesTokenSource?: PreferencesTokenSource, + signal?: AbortSignal ): Promise> { // TODO: Load the values from the integrationManager const newGlobals: HogFunctionInvocationGlobalsWithInputs = { @@ -83,7 +85,7 @@ export class HogInputsService { } // One budget for the whole invocation, so many small liquid leaves cannot add up to a stall. - const liquidBudget = new LiquidRenderBudget() + const liquidBudget = new LiquidRenderBudget(signal) // Build a lookup of schema types for rendering and post-render coercion const schemaTypes: Record = {} @@ -92,6 +94,7 @@ export class HogInputsService { } const _formatInput = async (input: CyclotronInputType, key: string): Promise => { + throwIfDeadlinePassed(signal) const templating = input.templating ?? 'hog' if (templating === 'liquid') { @@ -108,11 +111,18 @@ export class HogInputsService { : hogFunction.template_id === NATIVE_PUSH_TEMPLATE_ID ? new Set([key]) : undefined - return formatLiquidInput(value, newGlobals, key, liquidBudget, plainTextKeys) + try { + const rendered = formatLiquidInput(value, newGlobals, key, liquidBudget, plainTextKeys) + throwIfDeadlinePassed(signal) + return rendered + } catch (error) { + throwIfDeadlinePassed(signal) + throw error + } } if (templating === 'hog' && input?.bytecode) { try { - return await formatHogInput(input.bytecode, newGlobals, key) + return await formatHogInput(input.bytecode, newGlobals, key, signal) } catch (error) { throw withBytecodeContract(error, input.bytecode_contract) } @@ -172,11 +182,12 @@ export class HogInputsService { hogFunction: HogFunctionType, globals: HogFunctionInvocationGlobals, additionalInputs?: Record, - preferencesTokenSource?: PreferencesTokenSource + preferencesTokenSource?: PreferencesTokenSource, + signal?: AbortSignal ): Promise { return { ...globals, - inputs: await this.buildInputs(hogFunction, globals, additionalInputs, preferencesTokenSource), + inputs: await this.buildInputs(hogFunction, globals, additionalInputs, preferencesTokenSource, signal), } } @@ -453,7 +464,8 @@ const warehouseIntegrationMessage = (key: string): string => export const formatHogInput = async ( bytecode: any, globals: HogFunctionInvocationGlobalsWithInputs, - key?: string + key?: string, + signal?: AbortSignal ): Promise => { // Similar to how we generate the bytecode by iterating over the values, // here we iterate over the object and replace the bytecode with the actual values @@ -464,7 +476,7 @@ export const formatHogInput = async ( } if (Array.isArray(bytecode) && (bytecode[0] === '_h' || bytecode[0] === '_H')) { - const { execResult: result, error } = await execHog(bytecode, { globals }) + const { execResult: result, error } = await execHog(bytecode, { globals }, signal) if (!result || error) { throw error ?? result?.error } @@ -485,12 +497,12 @@ export const formatHogInput = async ( } if (Array.isArray(bytecode)) { - return await Promise.all(bytecode.map((item) => formatHogInput(item, globals, key))) + return await Promise.all(bytecode.map((item) => formatHogInput(item, globals, key, signal))) } else if (typeof bytecode === 'object' && bytecode !== null) { let ret: Record = {} if (bytecode[EXTEND_OBJECT_KEY]) { - const res = await formatHogInput(bytecode[EXTEND_OBJECT_KEY], globals, key) + const res = await formatHogInput(bytecode[EXTEND_OBJECT_KEY], globals, key, signal) if (res && typeof res === 'object') { ret = { ...res, @@ -503,7 +515,7 @@ export const formatHogInput = async ( if (subkey === EXTEND_OBJECT_KEY) { return } - ret[subkey] = await formatHogInput(value, globals, key ? `${key}.${subkey}` : subkey) + ret[subkey] = await formatHogInput(value, globals, key ? `${key}.${subkey}` : subkey, signal) }) ) diff --git a/nodejs/src/cdp/utils/deadline.ts b/nodejs/src/cdp/utils/deadline.ts new file mode 100644 index 000000000000..fb1f2c21921e --- /dev/null +++ b/nodejs/src/cdp/utils/deadline.ts @@ -0,0 +1,32 @@ +const expiresAt = new WeakMap() + +export function deadlineAfter(ms: number): AbortSignal { + const signal = AbortSignal.timeout(ms) + expiresAt.set(signal, performance.now() + ms) + return signal +} + +export function deadlineError(signal: AbortSignal | undefined): unknown { + if (!signal) { + return undefined + } + if (signal.aborted) { + return signal.reason + } + const at = expiresAt.get(signal) + return at !== undefined && performance.now() >= at + ? new DOMException('The operation was aborted due to timeout', 'TimeoutError') + : undefined +} + +export function deadlineRemainingMs(signal: AbortSignal | undefined): number | undefined { + const at = signal && expiresAt.get(signal) + return at === undefined ? undefined : Math.max(0, at - performance.now()) +} + +export function throwIfDeadlinePassed(signal: AbortSignal | undefined): void { + const error = deadlineError(signal) + if (error !== undefined) { + throw error + } +} diff --git a/nodejs/src/cdp/utils/hog-exec.ts b/nodejs/src/cdp/utils/hog-exec.ts index 210770f8da46..6faa3cffa1f3 100644 --- a/nodejs/src/cdp/utils/hog-exec.ts +++ b/nodejs/src/cdp/utils/hog-exec.ts @@ -6,13 +6,15 @@ import { instrumentFn } from '~/common/tracing/tracing-utils' import { yieldEventLoopIfNeeded } from '~/common/utils/event-loop-yield' import { createTrackedRE2 } from '~/common/utils/tracked-re2' +import { deadlineError, deadlineRemainingMs } from './deadline' import { Semaphore } from './sempahore' const semaphore = new Semaphore(1) export async function execHog( bytecode: any, - options?: ExecOptions + options?: ExecOptions, + signal?: AbortSignal ): Promise<{ execResult?: ExecResult error?: any @@ -20,11 +22,27 @@ export async function execHog( }> { return await semaphore.run(async () => { return await instrumentFn(`hog-exec`, async () => { - return await yieldEventLoopIfNeeded('hog-exec', () => execHogImmediate(bytecode, options)) + return await yieldEventLoopIfNeeded('hog-exec', () => { + const passedBefore = deadlineError(signal) + if (passedBefore !== undefined) { + return { error: passedBefore, durationMs: 0 } + } + const result = execHogImmediate(bytecode, withinDeadline(options, signal)) + const passedDuring = deadlineError(signal) + return passedDuring !== undefined ? { error: passedDuring, durationMs: result.durationMs } : result + }) }) }) } +function withinDeadline(options: ExecOptions | undefined, signal: AbortSignal | undefined): ExecOptions | undefined { + const remainingMs = deadlineRemainingMs(signal) + if (remainingMs === undefined) { + return options + } + return { ...options, timeout: Math.min(options?.timeout ?? DEFAULT_TIMEOUT_MS, Math.ceil(remainingMs) + 1) } +} + export function execHogImmediate( bytecode: any, options?: ExecOptions diff --git a/nodejs/src/cdp/utils/liquid.ts b/nodejs/src/cdp/utils/liquid.ts index 6794e5689fb0..ef895051ed40 100644 --- a/nodejs/src/cdp/utils/liquid.ts +++ b/nodejs/src/cdp/utils/liquid.ts @@ -2,6 +2,7 @@ import { Context, Emitter, Liquid, Tag, TagToken, Template, TopLevelToken } from import { rowScopedTriggerTypeForEvent } from '../schema/hogflow' import { HogFunctionInvocationGlobalsWithInputs } from '../types' +import { deadlineRemainingMs } from './deadline' // Rendering is synchronous on shared multi-tenant workers, so an unbounded template would // stall or OOM the whole process. These budgets make an over-budget template fail only its own invocation. @@ -14,7 +15,7 @@ const LIQUID_PARSE_LIMIT_CHARS = 5_000_000 const LIQUID_MEMORY_LIMIT_UNITS = 1_000_000 // Time and output are budgeted per invocation, not per template, because one invocation renders // every string leaf of every liquid input and those leaves would otherwise each get a full budget. -const LIQUID_RENDER_LIMIT_MS = 500 +export const LIQUID_RENDER_LIMIT_MS = 500 const LIQUID_OUTPUT_LIMIT_CHARS = 10_000_000 const LIQUID_TAG_CLOSERS: Record = { '{': '}}', '%': '%}' } @@ -28,8 +29,12 @@ export class LiquidRenderBudget { private usedMs = 0 private outputChars = 0 + constructor(private readonly deadline?: AbortSignal) {} + remainingMs(): number { - return Math.max(0, LIQUID_RENDER_LIMIT_MS - this.usedMs) + const ownMs = Math.max(0, LIQUID_RENDER_LIMIT_MS - this.usedMs) + const deadlineMs = deadlineRemainingMs(this.deadline) + return deadlineMs === undefined ? ownMs : Math.min(ownMs, Math.ceil(deadlineMs) + 1) } useTime(ms: number): void { diff --git a/nodejs/tests/fixtures/hog-bytecode/0a7c9f2282f0c526ba09f04bf535ef197338f8ac69ac218fbbc3c489fc7872f7.json b/nodejs/tests/fixtures/hog-bytecode/0a7c9f2282f0c526ba09f04bf535ef197338f8ac69ac218fbbc3c489fc7872f7.json new file mode 100644 index 000000000000..ecc22a95fd12 --- /dev/null +++ b/nodejs/tests/fixtures/hog-bytecode/0a7c9f2282f0c526ba09f04bf535ef197338f8ac69ac218fbbc3c489fc7872f7.json @@ -0,0 +1 @@ +["_H",1,32,"busy_until",32,"properties",32,"event",1,3,2,"now",0,2,"toUnixTimestampMilli",1,15,40,2,39,-19,32,"done",38] From 781911acafbbf23050d7925b81c3800723dad555 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 11:23:47 +0000 Subject: [PATCH 02/11] fix(cdp): stop rendering once the deadline passes between liquid leaves or after a hog yield A liquid input renders each of its leaves in turn, and a leaf that starts after the deadline still got a millisecond, so a large object could keep rendering toward the invocation budget. Each leaf now checks the deadline before it renders. A hog render could also finish just before the deadline and then return after it, from the event-loop yield, as a success; the deadline is now checked once more after that yield. Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- .../cdp/services/hog-inputs.service.test.ts | 90 +++++++++++++++++++ nodejs/src/cdp/services/hog-inputs.service.ts | 1 + nodejs/src/cdp/utils/hog-exec.ts | 8 +- nodejs/src/cdp/utils/liquid.ts | 6 +- 4 files changed, 100 insertions(+), 5 deletions(-) diff --git a/nodejs/src/cdp/services/hog-inputs.service.test.ts b/nodejs/src/cdp/services/hog-inputs.service.test.ts index fcf98346420f..5f98e1f24c39 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.test.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.test.ts @@ -404,6 +404,58 @@ describe('Hog Inputs', () => { } ) + it('fails the inputs with the deadline error when the deadline passes while a hog input yields the event loop after its render', async () => { + const onlyHog = createHogFunction({ + ...hogFunction, + inputs: { only: { value: 'x', templating: 'hog', bytecode: ['_H', 1, 32, 'x'] } }, + inputs_schema: [], + }) + const threshold = getEventLoopYieldThresholdMs() + await new Promise((resolve) => setTimeout(resolve, 5)) + configureEventLoopYield(0) + jest.useFakeTimers({ + doNotFake: [ + 'hrtime', + 'nextTick', + 'queueMicrotask', + 'setImmediate', + 'clearImmediate', + 'setInterval', + 'clearInterval', + 'Date', + ], + }) + const nextTurn = (): Promise => new Promise((resolve) => setImmediate(resolve)) + const untilAYieldWaits = async (): Promise => { + for (let turn = 0; jest.getTimerCount() === 0 && turn < 1000; turn++) { + await nextTurn() + } + } + try { + const deadline = new AbortController() + let outcome: unknown + void hogInputsService.buildInputs(onlyHog, globals, undefined, undefined, deadline.signal).then( + () => (outcome = 'rendered'), + (error: unknown) => (outcome = error) + ) + await untilAYieldWaits() + jest.runOnlyPendingTimers() + await untilAYieldWaits() + expect(outcome).toBeUndefined() + + deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) + for (let turn = 0; outcome === undefined && turn < 1000; turn++) { + jest.runOnlyPendingTimers() + await nextTurn() + } + + expect(outcome).toMatchObject({ name: 'TimeoutError' }) + } finally { + jest.useRealTimers() + configureEventLoopYield(threshold) + } + }) + it.each<[string, Record]>([ ['an input before the last', { next: { value: '{{ event.event }}', templating: 'liquid', order: 1 } }], ['the last input', {}], @@ -444,6 +496,44 @@ describe('Hog Inputs', () => { } }) + it('renders no further liquid leaf of an input once one leaf carries the clock past the deadline', async () => { + const deadlineMs = 50 + let now = performance.now() + const clock = jest.spyOn(performance, 'now').mockImplementation(() => now) + try { + const deadline = deadlineAfter(deadlineMs) + const leaf = '{{ event.properties.slow }}' + const manyLeaves = createHogFunction({ + ...hogFunction, + inputs: { leaves: { value: { first: leaf, second: leaf, third: leaf }, templating: 'liquid' } }, + inputs_schema: [], + }) + const renderedLeaves: number[] = [] + const properties = { ...globals.event.properties } + Object.defineProperty(properties, 'slow', { + enumerable: true, + get: () => { + renderedLeaves.push(now) + now += 2 * deadlineMs + return 'slow' + }, + }) + + await expect( + hogInputsService.buildInputs( + manyLeaves, + { ...globals, event: { ...globals.event, properties } }, + undefined, + undefined, + deadline + ) + ).rejects.toMatchObject({ name: 'TimeoutError' }) + expect(renderedLeaves).toHaveLength(1) + } finally { + clock.mockRestore() + } + }) + it.each<[string, number, () => Promise]>([ [ 'hog', diff --git a/nodejs/src/cdp/services/hog-inputs.service.ts b/nodejs/src/cdp/services/hog-inputs.service.ts index 928e51cb5eb9..49a50f7dc372 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.ts @@ -587,6 +587,7 @@ export const formatLiquidInput = ( } if (typeof value === 'string') { + budget.throwIfDeadlinePassed() try { return LiquidRenderer.renderWithHogFunctionGlobals(value, globals, budget, { escapeHtml: !isPlainTextKey(key, plainTextKeys), diff --git a/nodejs/src/cdp/utils/hog-exec.ts b/nodejs/src/cdp/utils/hog-exec.ts index 6faa3cffa1f3..11550c51276a 100644 --- a/nodejs/src/cdp/utils/hog-exec.ts +++ b/nodejs/src/cdp/utils/hog-exec.ts @@ -22,15 +22,15 @@ export async function execHog( }> { return await semaphore.run(async () => { return await instrumentFn(`hog-exec`, async () => { - return await yieldEventLoopIfNeeded('hog-exec', () => { + const result = await yieldEventLoopIfNeeded('hog-exec', () => { const passedBefore = deadlineError(signal) if (passedBefore !== undefined) { return { error: passedBefore, durationMs: 0 } } - const result = execHogImmediate(bytecode, withinDeadline(options, signal)) - const passedDuring = deadlineError(signal) - return passedDuring !== undefined ? { error: passedDuring, durationMs: result.durationMs } : result + return execHogImmediate(bytecode, withinDeadline(options, signal)) }) + const passed = deadlineError(signal) + return passed !== undefined ? { error: passed, durationMs: result.durationMs } : result }) }) } diff --git a/nodejs/src/cdp/utils/liquid.ts b/nodejs/src/cdp/utils/liquid.ts index ef895051ed40..7331ef12ef3e 100644 --- a/nodejs/src/cdp/utils/liquid.ts +++ b/nodejs/src/cdp/utils/liquid.ts @@ -2,7 +2,7 @@ import { Context, Emitter, Liquid, Tag, TagToken, Template, TopLevelToken } from import { rowScopedTriggerTypeForEvent } from '../schema/hogflow' import { HogFunctionInvocationGlobalsWithInputs } from '../types' -import { deadlineRemainingMs } from './deadline' +import { deadlineRemainingMs, throwIfDeadlinePassed } from './deadline' // Rendering is synchronous on shared multi-tenant workers, so an unbounded template would // stall or OOM the whole process. These budgets make an over-budget template fail only its own invocation. @@ -37,6 +37,10 @@ export class LiquidRenderBudget { return deadlineMs === undefined ? ownMs : Math.min(ownMs, Math.ceil(deadlineMs) + 1) } + throwIfDeadlinePassed(): void { + throwIfDeadlinePassed(this.deadline) + } + useTime(ms: number): void { this.usedMs += ms } From 5e7d68d49b224f02d419b2da7c7970d4bdf2f8a6 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 11:31:13 +0000 Subject: [PATCH 03/11] chore(cdp): assert the time a liquid input spends past its deadline The test now checks how far the clock moves past the deadline, which the caller waits on, rather than how often a property was read. Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- nodejs/src/cdp/services/hog-inputs.service.test.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/nodejs/src/cdp/services/hog-inputs.service.test.ts b/nodejs/src/cdp/services/hog-inputs.service.test.ts index 5f98e1f24c39..b03bd4e7d249 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.test.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.test.ts @@ -496,9 +496,11 @@ describe('Hog Inputs', () => { } }) - it('renders no further liquid leaf of an input once one leaf carries the clock past the deadline', async () => { + it('stops a liquid input of many leaves within one leaf of the deadline', async () => { const deadlineMs = 50 + const leafMs = 2 * deadlineMs let now = performance.now() + const startedAt = now const clock = jest.spyOn(performance, 'now').mockImplementation(() => now) try { const deadline = deadlineAfter(deadlineMs) @@ -508,13 +510,11 @@ describe('Hog Inputs', () => { inputs: { leaves: { value: { first: leaf, second: leaf, third: leaf }, templating: 'liquid' } }, inputs_schema: [], }) - const renderedLeaves: number[] = [] const properties = { ...globals.event.properties } Object.defineProperty(properties, 'slow', { enumerable: true, get: () => { - renderedLeaves.push(now) - now += 2 * deadlineMs + now += leafMs return 'slow' }, }) @@ -528,7 +528,7 @@ describe('Hog Inputs', () => { deadline ) ).rejects.toMatchObject({ name: 'TimeoutError' }) - expect(renderedLeaves).toHaveLength(1) + expect(now - startedAt).toBeLessThan(deadlineMs + leafMs) } finally { clock.mockRestore() } From a1a6e32c8523bcba31b77fd5c8f176508d6c9947 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:05:06 +0000 Subject: [PATCH 04/11] fix(cdp): stop a hog input that waits for the render semaphore once its deadline passes A render that queues behind another one now leaves the queue when its deadline passes, so buildInputs settles with the deadline error at once rather than after the earlier renders finish. Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- .../cdp/services/hog-inputs.service.test.ts | 60 +++++++++++++++++++ nodejs/src/cdp/utils/hog-exec.ts | 31 ++++++---- nodejs/src/cdp/utils/sempahore.ts | 20 +++++-- 3 files changed, 95 insertions(+), 16 deletions(-) diff --git a/nodejs/src/cdp/services/hog-inputs.service.test.ts b/nodejs/src/cdp/services/hog-inputs.service.test.ts index b03bd4e7d249..d2766dbcfeee 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.test.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.test.ts @@ -456,6 +456,66 @@ describe('Hog Inputs', () => { } }) + it('fails the inputs with the deadline error when the deadline passes while a hog input queues behind another render', async () => { + const onlyHog = createHogFunction({ + ...hogFunction, + inputs: { only: { value: 'x', templating: 'hog', bytecode: ['_H', 1, 32, 'x'] } }, + inputs_schema: [], + }) + const threshold = getEventLoopYieldThresholdMs() + await new Promise((resolve) => setTimeout(resolve, 5)) + configureEventLoopYield(0) + jest.useFakeTimers({ + doNotFake: [ + 'hrtime', + 'nextTick', + 'queueMicrotask', + 'setImmediate', + 'clearImmediate', + 'setInterval', + 'clearInterval', + 'Date', + ], + }) + const nextTurn = (): Promise => new Promise((resolve) => setImmediate(resolve)) + try { + let first: unknown + void hogInputsService.buildInputs(onlyHog, globals).then( + () => (first = 'rendered'), + (error: unknown) => (first = error) + ) + for (let turn = 0; jest.getTimerCount() === 0 && turn < 1000; turn++) { + await nextTurn() + } + const deadline = new AbortController() + let queued: unknown + void hogInputsService.buildInputs(onlyHog, globals, undefined, undefined, deadline.signal).then( + () => (queued = 'rendered'), + (error: unknown) => (queued = error) + ) + for (let turn = 0; turn < 20; turn++) { + await nextTurn() + } + expect(queued).toBeUndefined() + + deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) + for (let turn = 0; queued === undefined && turn < 1000; turn++) { + await nextTurn() + } + + expect(queued).toMatchObject({ name: 'TimeoutError' }) + expect(first).toBeUndefined() + for (let turn = 0; first === undefined && turn < 1000; turn++) { + jest.runOnlyPendingTimers() + await nextTurn() + } + expect(first).toBe('rendered') + } finally { + jest.useRealTimers() + configureEventLoopYield(threshold) + } + }) + it.each<[string, Record]>([ ['an input before the last', { next: { value: '{{ event.event }}', templating: 'liquid', order: 1 } }], ['the last input', {}], diff --git a/nodejs/src/cdp/utils/hog-exec.ts b/nodejs/src/cdp/utils/hog-exec.ts index 11550c51276a..d5f92351af7c 100644 --- a/nodejs/src/cdp/utils/hog-exec.ts +++ b/nodejs/src/cdp/utils/hog-exec.ts @@ -20,19 +20,26 @@ export async function execHog( error?: any durationMs: number }> { - return await semaphore.run(async () => { - return await instrumentFn(`hog-exec`, async () => { - const result = await yieldEventLoopIfNeeded('hog-exec', () => { - const passedBefore = deadlineError(signal) - if (passedBefore !== undefined) { - return { error: passedBefore, durationMs: 0 } - } - return execHogImmediate(bytecode, withinDeadline(options, signal)) + try { + return await semaphore.run(async () => { + return await instrumentFn(`hog-exec`, async () => { + const result = await yieldEventLoopIfNeeded('hog-exec', () => { + const passedBefore = deadlineError(signal) + if (passedBefore !== undefined) { + return { error: passedBefore, durationMs: 0 } + } + return execHogImmediate(bytecode, withinDeadline(options, signal)) + }) + const passed = deadlineError(signal) + return passed !== undefined ? { error: passed, durationMs: result.durationMs } : result }) - const passed = deadlineError(signal) - return passed !== undefined ? { error: passed, durationMs: result.durationMs } : result - }) - }) + }, signal) + } catch (error) { + if (signal?.aborted && error === signal.reason) { + return { error, durationMs: 0 } + } + throw error + } } function withinDeadline(options: ExecOptions | undefined, signal: AbortSignal | undefined): ExecOptions | undefined { diff --git a/nodejs/src/cdp/utils/sempahore.ts b/nodejs/src/cdp/utils/sempahore.ts index 09e196412436..a7ddf2225a21 100644 --- a/nodejs/src/cdp/utils/sempahore.ts +++ b/nodejs/src/cdp/utils/sempahore.ts @@ -3,12 +3,24 @@ export class Semaphore { constructor(private permits: number) {} - private async acquire(): Promise { + private async acquire(signal?: AbortSignal): Promise { if (this.permits > 0) { this.permits-- return } - return new Promise((resolve) => this.waiting.push(resolve)) + signal?.throwIfAborted() + return new Promise((resolve, reject) => { + const onAbort = (): void => { + this.waiting = this.waiting.filter((waiter) => waiter !== grant) + reject(signal!.reason) + } + const grant = (): void => { + signal?.removeEventListener('abort', onAbort) + resolve() + } + signal?.addEventListener('abort', onAbort, { once: true }) + this.waiting.push(grant) + }) } private release(): void { @@ -22,8 +34,8 @@ export class Semaphore { } } - async run(func: () => Promise): Promise { - await this.acquire() + async run(func: () => Promise, signal?: AbortSignal): Promise { + await this.acquire(signal) try { return await func() } finally { From 32d13cc4efecbab5d5652116dcd300624731bb24 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:10:16 +0000 Subject: [PATCH 05/11] fix(cdp): drop a cancelled semaphore waiter in constant time An input of many hog leaves queues one waiter per leaf under one deadline. Removing each with a filter made the abort quadratic and blocked the event loop; a set removes each waiter at once. Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- .../cdp/services/hog-inputs.service.test.ts | 128 ++++++++++-------- nodejs/src/cdp/utils/sempahore.ts | 17 ++- 2 files changed, 81 insertions(+), 64 deletions(-) diff --git a/nodejs/src/cdp/services/hog-inputs.service.test.ts b/nodejs/src/cdp/services/hog-inputs.service.test.ts index d2766dbcfeee..ad1628eaf0c6 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.test.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.test.ts @@ -456,65 +456,83 @@ describe('Hog Inputs', () => { } }) - it('fails the inputs with the deadline error when the deadline passes while a hog input queues behind another render', async () => { - const onlyHog = createHogFunction({ - ...hogFunction, - inputs: { only: { value: 'x', templating: 'hog', bytecode: ['_H', 1, 32, 'x'] } }, - inputs_schema: [], - }) - const threshold = getEventLoopYieldThresholdMs() - await new Promise((resolve) => setTimeout(resolve, 5)) - configureEventLoopYield(0) - jest.useFakeTimers({ - doNotFake: [ - 'hrtime', - 'nextTick', - 'queueMicrotask', - 'setImmediate', - 'clearImmediate', - 'setInterval', - 'clearInterval', - 'Date', - ], - }) - const nextTurn = (): Promise => new Promise((resolve) => setImmediate(resolve)) - try { - let first: unknown - void hogInputsService.buildInputs(onlyHog, globals).then( - () => (first = 'rendered'), - (error: unknown) => (first = error) - ) - for (let turn = 0; jest.getTimerCount() === 0 && turn < 1000; turn++) { - await nextTurn() - } - const deadline = new AbortController() - let queued: unknown - void hogInputsService.buildInputs(onlyHog, globals, undefined, undefined, deadline.signal).then( - () => (queued = 'rendered'), - (error: unknown) => (queued = error) - ) - for (let turn = 0; turn < 20; turn++) { - await nextTurn() - } - expect(queued).toBeUndefined() + it.each([1, 20_000])( + 'fails the inputs with the deadline error at once when the deadline passes while a hog input of %d leaves queues behind another render', + async (leaves) => { + const leaf = ['_H', 1, 32, 'x'] + const onlyHog = createHogFunction({ + ...hogFunction, + inputs: { only: { value: 'x', templating: 'hog', bytecode: leaf } }, + inputs_schema: [], + }) + const queuedHog = createHogFunction({ + ...hogFunction, + inputs: { + only: { + value: 'x', + templating: 'hog', + bytecode: leaves === 1 ? leaf : Array.from({ length: leaves }, () => leaf), + }, + }, + inputs_schema: [], + }) + const threshold = getEventLoopYieldThresholdMs() + await new Promise((resolve) => setTimeout(resolve, 5)) + configureEventLoopYield(0) + jest.useFakeTimers({ + doNotFake: [ + 'hrtime', + 'nextTick', + 'queueMicrotask', + 'setImmediate', + 'clearImmediate', + 'setInterval', + 'clearInterval', + 'Date', + ], + }) + const nextTurn = (): Promise => new Promise((resolve) => setImmediate(resolve)) + try { + let first: unknown + void hogInputsService.buildInputs(onlyHog, globals).then( + () => (first = 'rendered'), + (error: unknown) => (first = error) + ) + for (let turn = 0; jest.getTimerCount() === 0 && turn < 1000; turn++) { + await nextTurn() + } + const deadline = new AbortController() + let queued: unknown + void hogInputsService.buildInputs(queuedHog, globals, undefined, undefined, deadline.signal).then( + () => (queued = 'rendered'), + (error: unknown) => (queued = error) + ) + for (let turn = 0; turn < 20; turn++) { + await nextTurn() + } + expect(queued).toBeUndefined() - deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) - for (let turn = 0; queued === undefined && turn < 1000; turn++) { - await nextTurn() - } + const abortedAt = process.hrtime.bigint() + deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) + const abortMs = Number(process.hrtime.bigint() - abortedAt) / 1_000_000 + for (let turn = 0; queued === undefined && turn < 1000; turn++) { + await nextTurn() + } - expect(queued).toMatchObject({ name: 'TimeoutError' }) - expect(first).toBeUndefined() - for (let turn = 0; first === undefined && turn < 1000; turn++) { - jest.runOnlyPendingTimers() - await nextTurn() + expect(queued).toMatchObject({ name: 'TimeoutError' }) + expect(abortMs).toBeLessThan(250) + expect(first).toBeUndefined() + for (let turn = 0; first === undefined && turn < 1000; turn++) { + jest.runOnlyPendingTimers() + await nextTurn() + } + expect(first).toBe('rendered') + } finally { + jest.useRealTimers() + configureEventLoopYield(threshold) } - expect(first).toBe('rendered') - } finally { - jest.useRealTimers() - configureEventLoopYield(threshold) } - }) + ) it.each<[string, Record]>([ ['an input before the last', { next: { value: '{{ event.event }}', templating: 'liquid', order: 1 } }], diff --git a/nodejs/src/cdp/utils/sempahore.ts b/nodejs/src/cdp/utils/sempahore.ts index a7ddf2225a21..d6663ee509bf 100644 --- a/nodejs/src/cdp/utils/sempahore.ts +++ b/nodejs/src/cdp/utils/sempahore.ts @@ -1,5 +1,5 @@ export class Semaphore { - private waiting: Array<() => void> = [] + private waiting = new Set<() => void>() constructor(private permits: number) {} @@ -11,7 +11,7 @@ export class Semaphore { signal?.throwIfAborted() return new Promise((resolve, reject) => { const onAbort = (): void => { - this.waiting = this.waiting.filter((waiter) => waiter !== grant) + this.waiting.delete(grant) reject(signal!.reason) } const grant = (): void => { @@ -19,19 +19,18 @@ export class Semaphore { resolve() } signal?.addEventListener('abort', onAbort, { once: true }) - this.waiting.push(grant) + this.waiting.add(grant) }) } private release(): void { - if (this.waiting.length > 0) { - const next = this.waiting.shift() - if (next) { - next() - } - } else { + const next = this.waiting.values().next() + if (next.done) { this.permits++ + return } + this.waiting.delete(next.value) + next.value() } async run(func: () => Promise, signal?: AbortSignal): Promise { From 527fb8a9795b24234e57112a9ca0e4ee71e97f41 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:15:35 +0000 Subject: [PATCH 06/11] chore(cdp): bench the semaphore abort in the serial suite The wall-clock bound on dropping queued renders moves to an in-band serial bench, so runner contention in the parallel suite cannot fail it. Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- .../cdp/services/hog-inputs.service.test.ts | 128 ++++++++---------- .../cdp/utils/hog-exec-bench.serial.test.ts | 52 +++++++ 2 files changed, 107 insertions(+), 73 deletions(-) create mode 100644 nodejs/src/cdp/utils/hog-exec-bench.serial.test.ts diff --git a/nodejs/src/cdp/services/hog-inputs.service.test.ts b/nodejs/src/cdp/services/hog-inputs.service.test.ts index ad1628eaf0c6..d2766dbcfeee 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.test.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.test.ts @@ -456,83 +456,65 @@ describe('Hog Inputs', () => { } }) - it.each([1, 20_000])( - 'fails the inputs with the deadline error at once when the deadline passes while a hog input of %d leaves queues behind another render', - async (leaves) => { - const leaf = ['_H', 1, 32, 'x'] - const onlyHog = createHogFunction({ - ...hogFunction, - inputs: { only: { value: 'x', templating: 'hog', bytecode: leaf } }, - inputs_schema: [], - }) - const queuedHog = createHogFunction({ - ...hogFunction, - inputs: { - only: { - value: 'x', - templating: 'hog', - bytecode: leaves === 1 ? leaf : Array.from({ length: leaves }, () => leaf), - }, - }, - inputs_schema: [], - }) - const threshold = getEventLoopYieldThresholdMs() - await new Promise((resolve) => setTimeout(resolve, 5)) - configureEventLoopYield(0) - jest.useFakeTimers({ - doNotFake: [ - 'hrtime', - 'nextTick', - 'queueMicrotask', - 'setImmediate', - 'clearImmediate', - 'setInterval', - 'clearInterval', - 'Date', - ], - }) - const nextTurn = (): Promise => new Promise((resolve) => setImmediate(resolve)) - try { - let first: unknown - void hogInputsService.buildInputs(onlyHog, globals).then( - () => (first = 'rendered'), - (error: unknown) => (first = error) - ) - for (let turn = 0; jest.getTimerCount() === 0 && turn < 1000; turn++) { - await nextTurn() - } - const deadline = new AbortController() - let queued: unknown - void hogInputsService.buildInputs(queuedHog, globals, undefined, undefined, deadline.signal).then( - () => (queued = 'rendered'), - (error: unknown) => (queued = error) - ) - for (let turn = 0; turn < 20; turn++) { - await nextTurn() - } - expect(queued).toBeUndefined() + it('fails the inputs with the deadline error when the deadline passes while a hog input queues behind another render', async () => { + const onlyHog = createHogFunction({ + ...hogFunction, + inputs: { only: { value: 'x', templating: 'hog', bytecode: ['_H', 1, 32, 'x'] } }, + inputs_schema: [], + }) + const threshold = getEventLoopYieldThresholdMs() + await new Promise((resolve) => setTimeout(resolve, 5)) + configureEventLoopYield(0) + jest.useFakeTimers({ + doNotFake: [ + 'hrtime', + 'nextTick', + 'queueMicrotask', + 'setImmediate', + 'clearImmediate', + 'setInterval', + 'clearInterval', + 'Date', + ], + }) + const nextTurn = (): Promise => new Promise((resolve) => setImmediate(resolve)) + try { + let first: unknown + void hogInputsService.buildInputs(onlyHog, globals).then( + () => (first = 'rendered'), + (error: unknown) => (first = error) + ) + for (let turn = 0; jest.getTimerCount() === 0 && turn < 1000; turn++) { + await nextTurn() + } + const deadline = new AbortController() + let queued: unknown + void hogInputsService.buildInputs(onlyHog, globals, undefined, undefined, deadline.signal).then( + () => (queued = 'rendered'), + (error: unknown) => (queued = error) + ) + for (let turn = 0; turn < 20; turn++) { + await nextTurn() + } + expect(queued).toBeUndefined() - const abortedAt = process.hrtime.bigint() - deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) - const abortMs = Number(process.hrtime.bigint() - abortedAt) / 1_000_000 - for (let turn = 0; queued === undefined && turn < 1000; turn++) { - await nextTurn() - } + deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) + for (let turn = 0; queued === undefined && turn < 1000; turn++) { + await nextTurn() + } - expect(queued).toMatchObject({ name: 'TimeoutError' }) - expect(abortMs).toBeLessThan(250) - expect(first).toBeUndefined() - for (let turn = 0; first === undefined && turn < 1000; turn++) { - jest.runOnlyPendingTimers() - await nextTurn() - } - expect(first).toBe('rendered') - } finally { - jest.useRealTimers() - configureEventLoopYield(threshold) + expect(queued).toMatchObject({ name: 'TimeoutError' }) + expect(first).toBeUndefined() + for (let turn = 0; first === undefined && turn < 1000; turn++) { + jest.runOnlyPendingTimers() + await nextTurn() } + expect(first).toBe('rendered') + } finally { + jest.useRealTimers() + configureEventLoopYield(threshold) } - ) + }) it.each<[string, Record]>([ ['an input before the last', { next: { value: '{{ event.event }}', templating: 'liquid', order: 1 } }], diff --git a/nodejs/src/cdp/utils/hog-exec-bench.serial.test.ts b/nodejs/src/cdp/utils/hog-exec-bench.serial.test.ts new file mode 100644 index 000000000000..5e040695eb74 --- /dev/null +++ b/nodejs/src/cdp/utils/hog-exec-bench.serial.test.ts @@ -0,0 +1,52 @@ +import { configureEventLoopYield, getEventLoopYieldThresholdMs } from '~/common/utils/event-loop-yield' + +import { execHog } from './hog-exec' + +describe('hog render queue benchmark', () => { + const CEILING_MS = 1000 + const leaf = ['_H', 1, 32, 'x'] + + it('drops 50,000 queued renders under the ceiling once their shared deadline passes', async () => { + const threshold = getEventLoopYieldThresholdMs() + configureEventLoopYield(0) + jest.useFakeTimers({ + doNotFake: [ + 'hrtime', + 'nextTick', + 'queueMicrotask', + 'setImmediate', + 'clearImmediate', + 'setInterval', + 'clearInterval', + 'Date', + ], + }) + const nextTurn = (): Promise => new Promise((resolve) => setImmediate(resolve)) + try { + let first: Awaited> | undefined + void execHog(leaf).then((outcome) => (first = outcome)) + for (let turn = 0; jest.getTimerCount() === 0 && turn < 1000; turn++) { + await nextTurn() + } + const deadline = new AbortController() + const queued = Array.from({ length: 50_000 }, () => execHog(leaf, undefined, deadline.signal)) + await nextTurn() + + const started = process.hrtime.bigint() + deadline.abort(new DOMException('The deadline passed', 'TimeoutError')) + const elapsedMs = Number(process.hrtime.bigint() - started) / 1_000_000 + const outcomes = await Promise.all(queued) + + expect(outcomes.every((outcome) => outcome.error?.name === 'TimeoutError')).toBe(true) + expect(elapsedMs).toBeLessThan(CEILING_MS) + for (let turn = 0; first === undefined && turn < 1000; turn++) { + jest.runOnlyPendingTimers() + await nextTurn() + } + expect(first?.execResult?.result).toBe('x') + } finally { + jest.useRealTimers() + configureEventLoopYield(threshold) + } + }) +}) From daeea5e0dc82cda1f2816563db9d8ddbd209be6d Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:21:30 +0000 Subject: [PATCH 07/11] chore(cdp): move the semaphore abort bench under dev/ A wall-clock bound belongs in a benchmark, not in a suite CI runs. The bench now sits with the other dev-only runs, which Jest skips in CI. Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- .../hog-exec-queue-bench.test.ts} | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) rename nodejs/src/cdp/{utils/hog-exec-bench.serial.test.ts => dev/hog-exec-queue-bench.test.ts} (97%) diff --git a/nodejs/src/cdp/utils/hog-exec-bench.serial.test.ts b/nodejs/src/cdp/dev/hog-exec-queue-bench.test.ts similarity index 97% rename from nodejs/src/cdp/utils/hog-exec-bench.serial.test.ts rename to nodejs/src/cdp/dev/hog-exec-queue-bench.test.ts index 5e040695eb74..75b1cda8e6f5 100644 --- a/nodejs/src/cdp/utils/hog-exec-bench.serial.test.ts +++ b/nodejs/src/cdp/dev/hog-exec-queue-bench.test.ts @@ -1,6 +1,6 @@ import { configureEventLoopYield, getEventLoopYieldThresholdMs } from '~/common/utils/event-loop-yield' -import { execHog } from './hog-exec' +import { execHog } from '../utils/hog-exec' describe('hog render queue benchmark', () => { const CEILING_MS = 1000 From e9a810a29801de35aed026e0a6c9c05cec1e51aa Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:25:14 +0000 Subject: [PATCH 08/11] fix(cdp): register one abort listener per deadline for queued hog renders Each queued leaf added its own abort listener, and the signal scans its listeners on every add, so queuing thousands of leaves under one deadline blocked the event loop. Waiters on one signal now share a listener. Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- .../src/cdp/dev/hog-exec-queue-bench.test.ts | 11 ++-- nodejs/src/cdp/utils/sempahore.ts | 63 +++++++++++++++---- 2 files changed, 59 insertions(+), 15 deletions(-) diff --git a/nodejs/src/cdp/dev/hog-exec-queue-bench.test.ts b/nodejs/src/cdp/dev/hog-exec-queue-bench.test.ts index 75b1cda8e6f5..30f5999c05f3 100644 --- a/nodejs/src/cdp/dev/hog-exec-queue-bench.test.ts +++ b/nodejs/src/cdp/dev/hog-exec-queue-bench.test.ts @@ -6,7 +6,7 @@ describe('hog render queue benchmark', () => { const CEILING_MS = 1000 const leaf = ['_H', 1, 32, 'x'] - it('drops 50,000 queued renders under the ceiling once their shared deadline passes', async () => { + it('queues and drops 50,000 renders that share a deadline under the ceiling', async () => { const threshold = getEventLoopYieldThresholdMs() configureEventLoopYield(0) jest.useFakeTimers({ @@ -29,16 +29,19 @@ describe('hog render queue benchmark', () => { await nextTurn() } const deadline = new AbortController() + const queuedAt = process.hrtime.bigint() const queued = Array.from({ length: 50_000 }, () => execHog(leaf, undefined, deadline.signal)) + const queueMs = Number(process.hrtime.bigint() - queuedAt) / 1_000_000 await nextTurn() - const started = process.hrtime.bigint() + const abortedAt = process.hrtime.bigint() deadline.abort(new DOMException('The deadline passed', 'TimeoutError')) - const elapsedMs = Number(process.hrtime.bigint() - started) / 1_000_000 + const abortMs = Number(process.hrtime.bigint() - abortedAt) / 1_000_000 const outcomes = await Promise.all(queued) expect(outcomes.every((outcome) => outcome.error?.name === 'TimeoutError')).toBe(true) - expect(elapsedMs).toBeLessThan(CEILING_MS) + expect(queueMs).toBeLessThan(CEILING_MS) + expect(abortMs).toBeLessThan(CEILING_MS) for (let turn = 0; first === undefined && turn < 1000; turn++) { jest.runOnlyPendingTimers() await nextTurn() diff --git a/nodejs/src/cdp/utils/sempahore.ts b/nodejs/src/cdp/utils/sempahore.ts index d6663ee509bf..149381d12774 100644 --- a/nodejs/src/cdp/utils/sempahore.ts +++ b/nodejs/src/cdp/utils/sempahore.ts @@ -1,5 +1,17 @@ +interface Waiter { + resolve: () => void + reject: (reason: unknown) => void + signal?: AbortSignal +} + +interface SignalWaiters { + waiters: Set + onAbort: () => void +} + export class Semaphore { - private waiting = new Set<() => void>() + private waiting = new Set() + private waitingBySignal = new Map() constructor(private permits: number) {} @@ -10,19 +22,47 @@ export class Semaphore { } signal?.throwIfAborted() return new Promise((resolve, reject) => { - const onAbort = (): void => { - this.waiting.delete(grant) - reject(signal!.reason) + const waiter = { resolve, reject, signal } + this.waiting.add(waiter) + if (signal) { + this.watch(signal, waiter) } - const grant = (): void => { - signal?.removeEventListener('abort', onAbort) - resolve() - } - signal?.addEventListener('abort', onAbort, { once: true }) - this.waiting.add(grant) }) } + private watch(signal: AbortSignal, waiter: Waiter): void { + let group = this.waitingBySignal.get(signal) + if (!group) { + const waiters = new Set() + const onAbort = (): void => { + this.waitingBySignal.delete(signal) + for (const aborted of waiters) { + this.waiting.delete(aborted) + aborted.reject(signal.reason) + } + } + group = { waiters, onAbort } + this.waitingBySignal.set(signal, group) + signal.addEventListener('abort', onAbort, { once: true }) + } + group.waiters.add(waiter) + } + + private unwatch(waiter: Waiter): void { + if (!waiter.signal) { + return + } + const group = this.waitingBySignal.get(waiter.signal) + if (!group) { + return + } + group.waiters.delete(waiter) + if (group.waiters.size === 0) { + waiter.signal.removeEventListener('abort', group.onAbort) + this.waitingBySignal.delete(waiter.signal) + } + } + private release(): void { const next = this.waiting.values().next() if (next.done) { @@ -30,7 +70,8 @@ export class Semaphore { return } this.waiting.delete(next.value) - next.value() + this.unwatch(next.value) + next.value.resolve() } async run(func: () => Promise, signal?: AbortSignal): Promise { From cd5bb1a6c30dc6e100ddf5dbde674835b4ec30b3 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:33:20 +0000 Subject: [PATCH 09/11] chore(cdp): cover a deadline that passes after one leaf of an input takes the render queue Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- .../cdp/services/hog-inputs.service.test.ts | 68 +++++++++++++++++++ 1 file changed, 68 insertions(+) diff --git a/nodejs/src/cdp/services/hog-inputs.service.test.ts b/nodejs/src/cdp/services/hog-inputs.service.test.ts index d2766dbcfeee..d5312ec9f261 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.test.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.test.ts @@ -516,6 +516,74 @@ describe('Hog Inputs', () => { } }) + it('fails the inputs with the deadline error when the deadline passes while the rest of a hog input queues behind its own first leaf', async () => { + const leaf = ['_H', 1, 32, 'x'] + const onlyHog = createHogFunction({ + ...hogFunction, + inputs: { only: { value: 'x', templating: 'hog', bytecode: leaf } }, + inputs_schema: [], + }) + const manyLeaves = createHogFunction({ + ...hogFunction, + inputs: { only: { value: 'x', templating: 'hog', bytecode: Array.from({ length: 5 }, () => leaf) } }, + inputs_schema: [], + }) + const threshold = getEventLoopYieldThresholdMs() + await new Promise((resolve) => setTimeout(resolve, 5)) + configureEventLoopYield(0) + jest.useFakeTimers({ + doNotFake: [ + 'hrtime', + 'nextTick', + 'queueMicrotask', + 'setImmediate', + 'clearImmediate', + 'setInterval', + 'clearInterval', + 'Date', + ], + }) + const nextTurn = (): Promise => new Promise((resolve) => setImmediate(resolve)) + try { + let first: unknown + void hogInputsService.buildInputs(onlyHog, globals).then( + () => (first = 'rendered'), + (error: unknown) => (first = error) + ) + for (let turn = 0; jest.getTimerCount() === 0 && turn < 1000; turn++) { + await nextTurn() + } + const deadline = new AbortController() + let queued: unknown + void hogInputsService.buildInputs(manyLeaves, globals, undefined, undefined, deadline.signal).then( + () => (queued = 'rendered'), + (error: unknown) => (queued = error) + ) + for (let turn = 0; turn < 20; turn++) { + await nextTurn() + } + for (let turn = 0; first === undefined && turn < 1000; turn++) { + jest.runOnlyPendingTimers() + await nextTurn() + } + for (let turn = 0; jest.getTimerCount() === 0 && turn < 1000; turn++) { + await nextTurn() + } + expect(first).toBe('rendered') + expect(queued).toBeUndefined() + + deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) + for (let turn = 0; queued === undefined && turn < 1000; turn++) { + await nextTurn() + } + + expect(queued).toMatchObject({ name: 'TimeoutError' }) + } finally { + jest.useRealTimers() + configureEventLoopYield(threshold) + } + }) + it.each<[string, Record]>([ ['an input before the last', { next: { value: '{{ event.event }}', templating: 'liquid', order: 1 } }], ['the last input', {}], From 024fd7b54f4946dae8149ea87e49e02de70f3c1c Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:34:26 +0000 Subject: [PATCH 10/11] chore(cdp): let the queued leaf finish before the test restores real timers The leaf that held the render queue waited on a fake timer, so the tests after it found the queue taken and hung. Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- nodejs/src/cdp/services/hog-inputs.service.test.ts | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/nodejs/src/cdp/services/hog-inputs.service.test.ts b/nodejs/src/cdp/services/hog-inputs.service.test.ts index d5312ec9f261..15106e0a438d 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.test.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.test.ts @@ -579,6 +579,10 @@ describe('Hog Inputs', () => { expect(queued).toMatchObject({ name: 'TimeoutError' }) } finally { + for (let turn = 0; jest.getTimerCount() > 0 && turn < 1000; turn++) { + jest.runOnlyPendingTimers() + await nextTurn() + } jest.useRealTimers() configureEventLoopYield(threshold) } From 9d0d9e3285c38c8544eec41cf648d69acbd50a91 Mon Sep 17 00:00:00 2001 From: "Reichenbach, Michael" <755327+Silthus@users.noreply.github.com> Date: Wed, 11 Nov 2026 13:17:28 +0000 Subject: [PATCH 11/11] fix(cdp): keep the deadline error unstamped when builds share a deadline A hog input that fails with the deadline's own reason rethrows it unchanged, so a second build on the same deadline no longer fails with a TypeError while stamping a different bytecode contract. Refs PostHog/michaelr-context#77 Co-Authored-By: Claude Opus 5.5 --- .../cdp/services/hog-inputs.service.test.ts | 36 ++++++++++++++----- nodejs/src/cdp/services/hog-inputs.service.ts | 3 ++ 2 files changed, 30 insertions(+), 9 deletions(-) diff --git a/nodejs/src/cdp/services/hog-inputs.service.test.ts b/nodejs/src/cdp/services/hog-inputs.service.test.ts index 15106e0a438d..a6600f45547c 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.test.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.test.ts @@ -456,12 +456,25 @@ describe('Hog Inputs', () => { } }) - it('fails the inputs with the deadline error when the deadline passes while a hog input queues behind another render', async () => { + it('fails every build that shares a deadline with the deadline error when it passes while their hog inputs queue behind another render', async () => { const onlyHog = createHogFunction({ ...hogFunction, inputs: { only: { value: 'x', templating: 'hog', bytecode: ['_H', 1, 32, 'x'] } }, inputs_schema: [], }) + const compiledUnder = (contract: string): HogFunctionType => + createHogFunction({ + ...hogFunction, + inputs: { + only: { + value: 'x', + templating: 'hog', + bytecode: ['_H', 1, 32, 'x'], + bytecode_contract: contract, + }, + }, + inputs_schema: [], + }) const threshold = getEventLoopYieldThresholdMs() await new Promise((resolve) => setTimeout(resolve, 5)) configureEventLoopYield(0) @@ -488,22 +501,27 @@ describe('Hog Inputs', () => { await nextTurn() } const deadline = new AbortController() - let queued: unknown - void hogInputsService.buildInputs(onlyHog, globals, undefined, undefined, deadline.signal).then( - () => (queued = 'rendered'), - (error: unknown) => (queued = error) - ) + const contracts = ['older', 'newer'] + const queued: unknown[] = contracts.map(() => undefined) + contracts.forEach((contract, index) => { + void hogInputsService + .buildInputs(compiledUnder(contract), globals, undefined, undefined, deadline.signal) + .then( + () => (queued[index] = 'rendered'), + (error: unknown) => (queued[index] = error) + ) + }) for (let turn = 0; turn < 20; turn++) { await nextTurn() } - expect(queued).toBeUndefined() + expect(queued).toEqual([undefined, undefined]) deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) - for (let turn = 0; queued === undefined && turn < 1000; turn++) { + for (let turn = 0; queued.includes(undefined) && turn < 1000; turn++) { await nextTurn() } - expect(queued).toMatchObject({ name: 'TimeoutError' }) + expect(queued.map((outcome) => (outcome as Error).name)).toEqual(['TimeoutError', 'TimeoutError']) expect(first).toBeUndefined() for (let turn = 0; first === undefined && turn < 1000; turn++) { jest.runOnlyPendingTimers() diff --git a/nodejs/src/cdp/services/hog-inputs.service.ts b/nodejs/src/cdp/services/hog-inputs.service.ts index 49a50f7dc372..89c67719579c 100644 --- a/nodejs/src/cdp/services/hog-inputs.service.ts +++ b/nodejs/src/cdp/services/hog-inputs.service.ts @@ -124,6 +124,9 @@ export class HogInputsService { try { return await formatHogInput(input.bytecode, newGlobals, key, signal) } catch (error) { + if (error === signal?.reason) { + throw error + } throw withBytecodeContract(error, input.bytecode_contract) } }