diff --git a/nodejs/src/cdp/dev/hog-exec-queue-bench.test.ts b/nodejs/src/cdp/dev/hog-exec-queue-bench.test.ts new file mode 100644 index 000000000000..30f5999c05f3 --- /dev/null +++ b/nodejs/src/cdp/dev/hog-exec-queue-bench.test.ts @@ -0,0 +1,55 @@ +import { configureEventLoopYield, getEventLoopYieldThresholdMs } from '~/common/utils/event-loop-yield' + +import { execHog } from '../utils/hog-exec' + +describe('hog render queue benchmark', () => { + const CEILING_MS = 1000 + const leaf = ['_H', 1, 32, 'x'] + + it('queues and drops 50,000 renders that share a deadline under the ceiling', 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 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 abortedAt = process.hrtime.bigint() + deadline.abort(new DOMException('The deadline passed', 'TimeoutError')) + 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(queueMs).toBeLessThan(CEILING_MS) + expect(abortMs).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) + } + }) +}) diff --git a/nodejs/src/cdp/services/hog-inputs.service.test.ts b/nodejs/src/cdp/services/hog-inputs.service.test.ts index 50865198a04c..287aa35b6a93 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,11 +14,12 @@ 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 { EncryptedFields } from '../utils/encryption-utils' 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' @@ -306,6 +309,484 @@ 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('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('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) + 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() + 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).toEqual([undefined, undefined]) + + deadline.abort(new DOMException('The verification deadline passed', 'TimeoutError')) + for (let turn = 0; queued.includes(undefined) && turn < 1000; turn++) { + await nextTurn() + } + + 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() + await nextTurn() + } + expect(first).toBe('rendered') + } finally { + jest.useRealTimers() + configureEventLoopYield(threshold) + } + }) + + 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 { + for (let turn = 0; jest.getTimerCount() > 0 && turn < 1000; turn++) { + jest.runOnlyPendingTimers() + await nextTurn() + } + 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('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) + const leaf = '{{ event.properties.slow }}' + const manyLeaves = createHogFunction({ + ...hogFunction, + inputs: { leaves: { value: { first: leaf, second: leaf, third: leaf }, templating: 'liquid' } }, + inputs_schema: [], + }) + const properties = { ...globals.event.properties } + Object.defineProperty(properties, 'slow', { + enumerable: true, + get: () => { + now += leafMs + return 'slow' + }, + }) + + await expect( + hogInputsService.buildInputs( + manyLeaves, + { ...globals, event: { ...globals.event, properties } }, + undefined, + undefined, + deadline + ) + ).rejects.toMatchObject({ name: 'TimeoutError' }) + expect(now - startedAt).toBeLessThan(deadlineMs + leafMs) + } 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 be76e796c78f..301e6fdd8b1c 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,12 +111,22 @@ 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) { + if (error === signal?.reason) { + throw error + } throw withBytecodeContract(error, input.bytecode_contract) } } @@ -172,11 +185,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), } } @@ -470,7 +484,8 @@ const decryptIntegrationMessage = (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 @@ -481,7 +496,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 } @@ -502,12 +517,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, @@ -520,7 +535,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) }) ) @@ -592,6 +607,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/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..d5f92351af7c 100644 --- a/nodejs/src/cdp/utils/hog-exec.ts +++ b/nodejs/src/cdp/utils/hog-exec.ts @@ -6,23 +6,48 @@ 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 durationMs: number }> { - return await semaphore.run(async () => { - return await instrumentFn(`hog-exec`, async () => { - return await yieldEventLoopIfNeeded('hog-exec', () => execHogImmediate(bytecode, options)) - }) - }) + 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 + }) + }, 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 { + 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( diff --git a/nodejs/src/cdp/utils/liquid.ts b/nodejs/src/cdp/utils/liquid.ts index 6794e5689fb0..7331ef12ef3e 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, 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. @@ -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,16 @@ 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) + } + + throwIfDeadlinePassed(): void { + throwIfDeadlinePassed(this.deadline) } useTime(ms: number): void { diff --git a/nodejs/src/cdp/utils/sempahore.ts b/nodejs/src/cdp/utils/sempahore.ts index 09e196412436..149381d12774 100644 --- a/nodejs/src/cdp/utils/sempahore.ts +++ b/nodejs/src/cdp/utils/sempahore.ts @@ -1,29 +1,81 @@ +interface Waiter { + resolve: () => void + reject: (reason: unknown) => void + signal?: AbortSignal +} + +interface SignalWaiters { + waiters: Set + onAbort: () => void +} + export class Semaphore { - private waiting: Array<() => void> = [] + private waiting = new Set() + private waitingBySignal = new Map() 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 waiter = { resolve, reject, signal } + this.waiting.add(waiter) + if (signal) { + this.watch(signal, waiter) + } + }) } - private release(): void { - if (this.waiting.length > 0) { - const next = this.waiting.shift() - if (next) { - next() + 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) + } } - } else { + 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) { this.permits++ + return } + this.waiting.delete(next.value) + this.unwatch(next.value) + next.value.resolve() } - async run(func: () => Promise): Promise { - await this.acquire() + async run(func: () => Promise, signal?: AbortSignal): Promise { + await this.acquire(signal) try { return await func() } finally { 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]