feat(cdp): let a caller bound hog and liquid rendering with a deadline

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<void> => new Promise((resolve) => setImmediate(resolve))
+        try {
+            let first: Awaited<ReturnType<typeof execHog>> | 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)
+        }
+    })
+})

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<string, any>,
-        preferencesTokenSource?: PreferencesTokenSource
+        preferencesTokenSource?: PreferencesTokenSource,
+        signal?: AbortSignal
     ): Promise<Record<string, any>> {
         // 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<string, string> = {}
@@ -92,6 +94,7 @@ export class HogInputsService {
         }
 
         const _formatInput = async (input: CyclotronInputType, key: string): Promise<any> => {
+            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<string, any>,
-        preferencesTokenSource?: PreferencesTokenSource
+        preferencesTokenSource?: PreferencesTokenSource,
+        signal?: AbortSignal
     ): Promise<HogFunctionInvocationGlobalsWithInputs> {
         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<any> => {
     // 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<string, any> = {}
 
         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),

nodejs/src/cdp/utils/deadline.ts

@@ -0,0 +1,32 @@
+const expiresAt = new WeakMap<AbortSignal, number>()
+
+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
+    }
+}

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(

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<string, string> = { '{': '}}', '%': '%}' }
@@ -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 {

nodejs/src/cdp/utils/sempahore.ts

@@ -1,29 +1,81 @@
+interface Waiter {
+    resolve: () => void
+    reject: (reason: unknown) => void
+    signal?: AbortSignal
+}
+
+interface SignalWaiters {
+    waiters: Set<Waiter>
+    onAbort: () => void
+}
+
 export class Semaphore {
-    private waiting: Array<() => void> = []
+    private waiting = new Set<Waiter>()
+    private waitingBySignal = new Map<AbortSignal, SignalWaiters>()
 
     constructor(private permits: number) {}
 
-    private async acquire(): Promise<void> {
+    private async acquire(signal?: AbortSignal): Promise<void> {
         if (this.permits > 0) {
             this.permits--
             return
         }
-        return new Promise<void>((resolve) => this.waiting.push(resolve))
+        signal?.throwIfAborted()
+        return new Promise<void>((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<Waiter>()
+            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<T>(func: () => Promise<T>): Promise<T> {
-        await this.acquire()
+    async run<T>(func: () => Promise<T>, signal?: AbortSignal): Promise<T> {
+        await this.acquire(signal)
         try {
             return await func()
         } finally {

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]