mirror of
https://github.com/anomalyco/opencode.git
synced 2026-10-24 10:50:04 +00:00
240 lines
11 KiB
TypeScript
240 lines
11 KiB
TypeScript
import { expect, test } from "bun:test"
|
|
import { Cause, Effect, Fiber, Queue } from "effect"
|
|
import { Socket } from "effect/socket"
|
|
import FastCheck from "fast-check"
|
|
import { type Outbound, runPtySocket } from "../src/handlers/pty-socket"
|
|
|
|
// Latencies are mostly scheduler yields so interleavings stay dense and fast; a few are real sleeps.
|
|
type Latency = { readonly yields: number } | { readonly millis: number }
|
|
|
|
type Step =
|
|
| { readonly type: "inbound"; readonly message: string | Uint8Array }
|
|
| { readonly type: "outbound"; readonly frame: string | Uint8Array }
|
|
| { readonly type: "serverClose"; readonly code: number }
|
|
| { readonly type: "clientClose"; readonly code: number }
|
|
| { readonly type: "readError" }
|
|
| { readonly type: "writeError" }
|
|
| { readonly type: "interrupt" }
|
|
|
|
type Scenario = {
|
|
readonly replay: ReadonlyArray<string | Uint8Array>
|
|
readonly steps: ReadonlyArray<{ readonly step: Step; readonly delay: Latency }>
|
|
readonly reader: Latency
|
|
readonly open: Latency | undefined
|
|
readonly message: ReadonlyArray<Latency>
|
|
readonly write: ReadonlyArray<Latency>
|
|
readonly batch: number
|
|
}
|
|
|
|
type Terminal = "close" | "readError" | "writeError" | "interrupt"
|
|
|
|
type Entry =
|
|
| { readonly type: "reader" | "openStart" | "openEnd" | "detach" }
|
|
| { readonly type: "messageStart" | "messageEnd"; readonly message: string | Uint8Array }
|
|
| { readonly type: "write"; readonly chunk: Outbound }
|
|
| { readonly type: "terminal"; readonly terminal: Terminal }
|
|
|
|
const latency: FastCheck.Arbitrary<Latency> = FastCheck.oneof(
|
|
{ weight: 12, arbitrary: FastCheck.record({ yields: FastCheck.nat(6) }) },
|
|
{ weight: 1, arbitrary: FastCheck.record({ millis: FastCheck.integer({ min: 1, max: 3 }) }) },
|
|
)
|
|
const payload = FastCheck.oneof(FastCheck.string({ maxLength: 8 }), FastCheck.uint8Array({ maxLength: 8 }))
|
|
const code = FastCheck.constantFrom(1000, 1001, 1006, 4404)
|
|
const step: FastCheck.Arbitrary<Step> = FastCheck.oneof(
|
|
{ weight: 8, arbitrary: FastCheck.record({ type: FastCheck.constant("inbound"), message: payload }) },
|
|
{ weight: 8, arbitrary: FastCheck.record({ type: FastCheck.constant("outbound"), frame: payload }) },
|
|
{ weight: 1, arbitrary: FastCheck.record({ type: FastCheck.constant("serverClose"), code }) },
|
|
{ weight: 1, arbitrary: FastCheck.record({ type: FastCheck.constant("clientClose"), code }) },
|
|
{ weight: 1, arbitrary: FastCheck.record({ type: FastCheck.constant("readError") }) },
|
|
{ weight: 1, arbitrary: FastCheck.record({ type: FastCheck.constant("writeError") }) },
|
|
{ weight: 1, arbitrary: FastCheck.record({ type: FastCheck.constant("interrupt") }) },
|
|
)
|
|
const scenario: FastCheck.Arbitrary<Scenario> = FastCheck.record({
|
|
replay: FastCheck.array(payload, { maxLength: 4 }),
|
|
steps: FastCheck.array(FastCheck.record({ step, delay: latency }), { maxLength: 24 }),
|
|
reader: latency,
|
|
open: FastCheck.option(latency, { nil: undefined }),
|
|
message: FastCheck.array(latency, { minLength: 1, maxLength: 4 }),
|
|
write: FastCheck.array(latency, { minLength: 1, maxLength: 4 }),
|
|
batch: FastCheck.integer({ min: 1, max: 3 }),
|
|
})
|
|
|
|
const wait = (value: Latency) =>
|
|
Effect.gen(function* () {
|
|
if ("millis" in value) return yield* Effect.sleep(value.millis)
|
|
for (let i = 0; i < value.yields; i++) yield* Effect.yieldNow
|
|
})
|
|
|
|
const nonEmpty = <A>(values: ReadonlyArray<A>): values is readonly [A, ...Array<A>] => values.length > 0
|
|
|
|
const socketError = (reason: Socket.SocketError["reason"]) => new Socket.SocketError({ reason })
|
|
|
|
// Drives one connection through the scenario against an in-memory socket and records every observable event.
|
|
const simulate = (input: Scenario) =>
|
|
Effect.gen(function* () {
|
|
const inbound = yield* Queue.unbounded<string | Uint8Array, Socket.SocketError>()
|
|
const outbox = yield* Queue.unbounded<Outbound>()
|
|
const log: Entry[] = []
|
|
const sent: Array<string | Uint8Array> = []
|
|
const offered: Outbound[] = [...input.replay]
|
|
const state = { failNextWrite: false, writes: 0, messages: 0, detached: 0 }
|
|
yield* Queue.offerAll(outbox, input.replay)
|
|
|
|
const socket = Socket.make({
|
|
reader: wait(input.reader).pipe(
|
|
Effect.tap(() => Effect.sync(() => log.push({ type: "reader" }))),
|
|
Effect.as({
|
|
pull: Queue.takeBetween(inbound, 1, input.batch).pipe(
|
|
Effect.tapError((error) =>
|
|
Effect.sync(() =>
|
|
log.push({
|
|
type: "terminal",
|
|
terminal: error.reason._tag === "SocketCloseError" ? "close" : "readError",
|
|
}),
|
|
),
|
|
),
|
|
Effect.flatMap((batch) =>
|
|
nonEmpty(batch) ? Effect.succeed(batch) : Effect.die(new Error("empty batch")),
|
|
),
|
|
),
|
|
upgrade: Socket.SocketUpgradeError.unsupported,
|
|
}),
|
|
),
|
|
writer: Effect.succeed({
|
|
write: (chunk: Outbound) =>
|
|
Effect.gen(function* () {
|
|
yield* wait(input.write[state.writes++ % input.write.length]!)
|
|
if (state.failNextWrite) {
|
|
state.failNextWrite = false
|
|
log.push({ type: "terminal", terminal: "writeError" })
|
|
return yield* socketError(new Socket.SocketWriteError({ cause: new Error("write failed") }))
|
|
}
|
|
log.push({ type: "write", chunk })
|
|
if (chunk instanceof Socket.CloseEvent) log.push({ type: "terminal", terminal: "close" })
|
|
}),
|
|
writeAll: () => Effect.die(new Error("runPtySocket writes frames one at a time")),
|
|
}),
|
|
})
|
|
|
|
const open = input.open
|
|
const run = yield* runPtySocket({
|
|
socket,
|
|
outbox,
|
|
onOpen:
|
|
open &&
|
|
Effect.gen(function* () {
|
|
log.push({ type: "openStart" })
|
|
yield* wait(open)
|
|
log.push({ type: "openEnd" })
|
|
}),
|
|
onMessage: (message) =>
|
|
Effect.gen(function* () {
|
|
log.push({ type: "messageStart", message })
|
|
yield* wait(input.message[state.messages++ % input.message.length]!)
|
|
log.push({ type: "messageEnd", message })
|
|
}),
|
|
detach: () => {
|
|
state.detached++
|
|
log.push({ type: "detach" })
|
|
},
|
|
}).pipe(Effect.scoped, Effect.forkChild({ startImmediately: true }))
|
|
|
|
for (const { step, delay } of input.steps) {
|
|
yield* wait(delay)
|
|
if (step.type === "inbound" && (yield* Queue.offer(inbound, step.message))) sent.push(step.message)
|
|
if (step.type === "outbound") {
|
|
offered.push(step.frame)
|
|
yield* Queue.offer(outbox, step.frame)
|
|
}
|
|
if (step.type === "serverClose") {
|
|
const close = new Socket.CloseEvent(step.code)
|
|
offered.push(close)
|
|
yield* Queue.offer(outbox, close)
|
|
}
|
|
if (step.type === "clientClose")
|
|
yield* Queue.fail(inbound, socketError(new Socket.SocketCloseError({ code: step.code })))
|
|
if (step.type === "readError")
|
|
yield* Queue.fail(inbound, socketError(new Socket.SocketReadError({ cause: new Error("reset") })))
|
|
if (step.type === "writeError") state.failNextWrite = true
|
|
if (step.type === "interrupt") {
|
|
log.push({ type: "terminal", terminal: "interrupt" })
|
|
// Fiber.interrupt waits for the run to finish, so a run that ignores interruption would stall here.
|
|
if ((yield* Fiber.interrupt(run).pipe(Effect.timeoutOption("2 seconds")))._tag === "None")
|
|
return yield* Effect.die(new Error("runPtySocket ignored interruption"))
|
|
}
|
|
}
|
|
// Every real connection eventually closes; this bounds runs whose scenario never ended them.
|
|
yield* Queue.fail(inbound, socketError(new Socket.SocketCloseError({ code: 1000 })))
|
|
const exit = yield* Fiber.await(run).pipe(Effect.timeoutOption("5 seconds"))
|
|
return { exit, log, sent, offered, state, hasOpen: open !== undefined }
|
|
})
|
|
|
|
function verify(result: Effect.Success<ReturnType<typeof simulate>>) {
|
|
// The run terminates within the scheduler bound.
|
|
if (result.exit._tag === "None") throw new Error("runPtySocket did not terminate")
|
|
const exit = result.exit.value
|
|
const log = result.log
|
|
const index = (predicate: (entry: Entry) => boolean) => log.findIndex(predicate)
|
|
const detach = index((entry) => entry.type === "detach")
|
|
|
|
// detach runs exactly once, whatever ended the connection.
|
|
expect(result.state.detached).toBe(1)
|
|
|
|
// Outbound frames are written in queue order and nothing follows a close frame or detach.
|
|
const written = log.flatMap((entry) => (entry.type === "write" ? [entry.chunk] : []))
|
|
expect(written.every((chunk, position) => chunk === result.offered[position])).toBeTrue()
|
|
const close = written.findIndex((chunk) => chunk instanceof Socket.CloseEvent)
|
|
if (close !== -1) expect(close).toBe(written.length - 1)
|
|
expect(log.slice(detach).some((entry) => entry.type === "write")).toBeFalse()
|
|
|
|
// Inbound messages reach onMessage one at a time, in order, after onOpen and before detach.
|
|
const started = log.flatMap((entry) => (entry.type === "messageStart" ? [entry.message] : []))
|
|
expect(started).toEqual(result.sent.slice(0, started.length))
|
|
const reader = index((entry) => entry.type === "reader")
|
|
const openStart = index((entry) => entry.type === "openStart")
|
|
const openEnd = index((entry) => entry.type === "openEnd")
|
|
if (openStart !== -1) expect(openStart).toBeGreaterThan(reader)
|
|
const firstMessage = index((entry) => entry.type === "messageStart")
|
|
if (firstMessage !== -1) {
|
|
expect(firstMessage).toBeGreaterThan(reader)
|
|
if (result.hasOpen) expect(firstMessage).toBeGreaterThan(openEnd)
|
|
expect(firstMessage).toBeLessThan(detach)
|
|
}
|
|
expect(log.slice(detach).some((entry) => entry.type === "messageStart")).toBeFalse()
|
|
let inFlight = 0
|
|
for (const entry of log) {
|
|
if (entry.type === "messageStart") inFlight++
|
|
if (entry.type === "messageEnd") inFlight--
|
|
expect(inFlight).toBeLessThanOrEqual(1)
|
|
}
|
|
|
|
// Clean closes succeed, socket failures die, and interruption is honored. The outcome follows the first
|
|
// close or failure the run observed; an interrupt that lands while the race settles may still win, but
|
|
// nothing observed after an interrupt may decide the outcome.
|
|
const terminals = log.flatMap((entry) => (entry.type === "terminal" ? [entry.terminal] : []))
|
|
if (exit._tag === "Failure" && Cause.hasInterrupts(exit.cause)) {
|
|
expect(terminals).toContain("interrupt")
|
|
return
|
|
}
|
|
const first = terminals[0]
|
|
if (first === "close") {
|
|
expect(exit._tag).toBe("Success")
|
|
return
|
|
}
|
|
if (first === "interrupt" || first === undefined) throw new Error(`Interrupt was not honored: ${exit._tag}`)
|
|
if (exit._tag !== "Failure" || !Cause.hasDies(exit.cause))
|
|
throw new Error(`Expected the run to die after ${first}, got ${exit._tag}`)
|
|
const defect = Cause.squash(exit.cause)
|
|
if (!Socket.isSocketError(defect)) throw new Error(`Expected a socket error defect, got ${String(defect)}`)
|
|
expect(defect.reason._tag).toBe(first === "readError" ? "SocketReadError" : "SocketWriteError")
|
|
}
|
|
|
|
test(
|
|
"runPtySocket preserves ordering, detaches once, and ends correctly under random interleavings",
|
|
() =>
|
|
FastCheck.assert(
|
|
FastCheck.asyncProperty(scenario, async (input) => verify(await Effect.runPromise(simulate(input)))),
|
|
{ numRuns: 300 },
|
|
),
|
|
60_000,
|
|
)
|