orca/src/shared/runtime-rpc-call-queue.test.ts

import { describe, expect, it, vi } from 'vitest'
import {
  isBackgroundRuntimeMethod,
  RuntimeRpcCallQueueOverloadError,
  RuntimeRpcCallQueuePool
} from './runtime-rpc-call-queue'

describe('runtime RPC call queue', () => {
  it('classifies per-worktree decoration lookups as background work', () => {
    expect(isBackgroundRuntimeMethod('github.prForBranch')).toBe(true)
    expect(isBackgroundRuntimeMethod('hostedReview.forBranch')).toBe(true)
    expect(isBackgroundRuntimeMethod('worktree.prefetchCreateBase')).toBe(true)
    expect(isBackgroundRuntimeMethod('terminal.send')).toBe(false)
    expect(isBackgroundRuntimeMethod('terminal.agentStatus')).toBe(false)
    expect(isBackgroundRuntimeMethod('worktree.create')).toBe(false)
  })

  it('limits background calls while allowing foreground runtime work through', async () => {
    const queue = new RuntimeRpcCallQueuePool(2, 1)
    const started: string[] = []
    const pending: ((value: string) => void)[] = []
    const enqueuePending = (method: string, label: string): Promise<string> =>
      queue.enqueue('web-runtime', method, async () => {
        started.push(label)
        return await new Promise<string>((resolve) => pending.push(resolve))
      })

    const background1 = enqueuePending('github.prForBranch', 'background-1')
    const background2 = enqueuePending('hostedReview.forBranch', 'background-2')
    await vi.waitFor(() => expect(started).toEqual(['background-1']))

    const foreground = queue.enqueue('web-runtime', 'terminal.send', async () => {
      started.push('foreground')
      return 'foreground'
    })
    await expect(foreground).resolves.toBe('foreground')
    expect(started).toEqual(['background-1', 'foreground'])

    pending.shift()?.('background-1')
    await expect(background1).resolves.toBe('background-1')
    await vi.waitFor(() => expect(started).toEqual(['background-1', 'foreground', 'background-2']))

    pending.shift()?.('background-2')
    await expect(background2).resolves.toBe('background-2')
  })

  it('frees the queue slot when a runtime call throws synchronously', async () => {
    const queue = new RuntimeRpcCallQueuePool(1, 1)
    const first = queue.enqueue('web-runtime', 'status.get', () => {
      throw new Error('invalid stored runtime pairing')
    })

    await expect(first).rejects.toThrow('invalid stored runtime pairing')

    const second = queue.enqueue('web-runtime', 'status.get', async () => 'second')
    await expect(second).resolves.toBe('second')
  })

  it('removes an aborted call before it starts', async () => {
    const queue = new RuntimeRpcCallQueuePool(1, 1)
    let releaseFirst: () => void = () => {}
    const first = queue.enqueue('runtime-a', 'status.get', async () => {
      await new Promise<void>((resolve) => {
        releaseFirst = resolve
      })
    })
    const controller = new AbortController()
    const run = vi.fn(async () => 'cancelled')
    const cancelled = queue.enqueue('runtime-a', 'status.get', run, 1, controller.signal)

    controller.abort()
    await expect(cancelled).rejects.toMatchObject({ name: 'AbortError' })
    expect(run).not.toHaveBeenCalled()

    releaseFirst()
    await expect(first).resolves.toBeUndefined()
    await expect(
      queue.enqueue('runtime-a', 'status.get', async () => 'recovered', 1)
    ).resolves.toBe('recovered')
  })

  it('preserves queued background ordering across large bursts', async () => {
    const queue = new RuntimeRpcCallQueuePool(1, 1)
    const started: number[] = []
    let releaseFirst: () => void = () => {}
    const first = queue.enqueue('web-runtime', 'github.prForBranch', async () => {
      started.push(0)
      await new Promise<void>((resolve) => {
        releaseFirst = resolve
      })
      return 0
    })
    const rest = Array.from({ length: 70 }, (_, index) =>
      queue.enqueue('web-runtime', 'github.prForBranch', async () => {
        const value = index + 1
        started.push(value)
        return value
      })
    )

    await vi.waitFor(() => expect(started).toEqual([0]))
    releaseFirst()

    await expect(Promise.all([first, ...rest])).resolves.toEqual(
      Array.from({ length: 71 }, (_, index) => index)
    )
    expect(started).toEqual(Array.from({ length: 71 }, (_, index) => index))
  })

  it('rejects per-selector overload and accepts work after the queue drains', async () => {
    const queue = new RuntimeRpcCallQueuePool(1, 1, 2, 10)
    let releaseFirst: () => void = () => {}
    const first = queue.enqueue('runtime-a', 'status.get', async () => {
      await new Promise<void>((resolve) => {
        releaseFirst = resolve
      })
      return 'first'
    })
    const second = queue.enqueue('runtime-a', 'status.get', async () => 'second')
    const third = queue.enqueue('runtime-a', 'status.get', async () => 'third')

    await expect(queue.enqueue('runtime-a', 'status.get', async () => 'overflow')).rejects.toEqual(
      expect.objectContaining({
        code: 'runtime_rpc_queue_overloaded',
        scope: 'selector'
      })
    )

    releaseFirst()
    await expect(Promise.all([first, second, third])).resolves.toEqual(['first', 'second', 'third'])
    await expect(queue.enqueue('runtime-a', 'status.get', async () => 'recovered')).resolves.toBe(
      'recovered'
    )
  })

  it('caps queued calls across selectors and recovers after draining', async () => {
    const queue = new RuntimeRpcCallQueuePool(1, 1, 10, 2)
    const releases: (() => void)[] = []
    const blockers = ['runtime-a', 'runtime-b'].map((selector) =>
      queue.enqueue(selector, 'status.get', async () => {
        await new Promise<void>((resolve) => releases.push(resolve))
      })
    )
    const queuedA = queue.enqueue('runtime-a', 'status.get', async () => 'queued-a')
    const queuedB = queue.enqueue('runtime-b', 'status.get', async () => 'queued-b')

    const overload = queue.enqueue('runtime-c', 'status.get', async () => 'overflow')
    await expect(overload).rejects.toBeInstanceOf(RuntimeRpcCallQueueOverloadError)
    await expect(overload).rejects.toMatchObject({ scope: 'global' })

    releases.splice(0).forEach((release) => release())
    await expect(Promise.all([...blockers, queuedA, queuedB])).resolves.toEqual([
      undefined,
      undefined,
      'queued-a',
      'queued-b'
    ])
    await expect(queue.enqueue('runtime-c', 'status.get', async () => 'recovered')).resolves.toBe(
      'recovered'
    )
  })

  it('caps retained call bytes across active and queued work, then recovers', async () => {
    const queue = new RuntimeRpcCallQueuePool(1, 1, 10, 10, 10)
    let releaseFirst: () => void = () => {}
    let firstStarted = false
    const first = queue.enqueue(
      'runtime-a',
      'status.get',
      async () => {
        firstStarted = true
        await new Promise<void>((resolve) => {
          releaseFirst = resolve
        })
        return 'first'
      },
      10
    )

    await vi.waitFor(() => expect(firstStarted).toBe(true))
    await expect(
      queue.enqueue('runtime-b', 'status.get', async () => 'overflow', 1)
    ).rejects.toMatchObject({ scope: 'memory' })

    releaseFirst()
    await expect(first).resolves.toBe('first')
    await expect(
      queue.enqueue('runtime-b', 'status.get', async () => 'recovered', 10)
    ).resolves.toBe('recovered')
  })
})