summaryrefslogtreecommitdiffhomepage
path: root/packages/shared/test/util/flock.test.ts
diff options
context:
space:
mode:
Diffstat (limited to 'packages/shared/test/util/flock.test.ts')
-rw-r--r--packages/shared/test/util/flock.test.ts426
1 files changed, 0 insertions, 426 deletions
diff --git a/packages/shared/test/util/flock.test.ts b/packages/shared/test/util/flock.test.ts
deleted file mode 100644
index f1053dfd2..000000000
--- a/packages/shared/test/util/flock.test.ts
+++ /dev/null
@@ -1,426 +0,0 @@
-import { describe, expect, test } from "bun:test"
-import fs from "fs/promises"
-import { spawn } from "child_process"
-import path from "path"
-import os from "os"
-import { Flock } from "@opencode-ai/shared/util/flock"
-import { Hash } from "@opencode-ai/shared/util/hash"
-
-type Msg = {
- key: string
- dir: string
- staleMs?: number
- timeoutMs?: number
- baseDelayMs?: number
- maxDelayMs?: number
- holdMs?: number
- ready?: string
- active?: string
- done?: string
-}
-
-const root = path.join(import.meta.dir, "../..")
-const worker = path.join(import.meta.dir, "../fixture/flock-worker.ts")
-
-async function tmpdir() {
- const dir = await fs.mkdtemp(path.join(os.tmpdir(), "flock-test-"))
- return {
- path: dir,
- async [Symbol.asyncDispose]() {
- await fs.rm(dir, { recursive: true, force: true })
- },
- }
-}
-
-function lock(dir: string, key: string) {
- return path.join(dir, Hash.fast(key) + ".lock")
-}
-
-function sleep(ms: number) {
- return new Promise<void>((resolve) => {
- setTimeout(resolve, ms)
- })
-}
-
-async function exists(file: string) {
- return fs
- .stat(file)
- .then(() => true)
- .catch(() => false)
-}
-
-async function wait(file: string, timeout = 3_000) {
- const stop = Date.now() + timeout
- while (Date.now() < stop) {
- if (await exists(file)) return
- await sleep(20)
- }
-
- throw new Error(`Timed out waiting for file: ${file}`)
-}
-
-function run(msg: Msg) {
- return new Promise<{ code: number; stdout: Buffer; stderr: Buffer }>((resolve) => {
- const proc = spawn(process.execPath, [worker, JSON.stringify(msg)], {
- cwd: root,
- })
-
- const stdout: Buffer[] = []
- const stderr: Buffer[] = []
-
- proc.stdout?.on("data", (data) => stdout.push(Buffer.from(data)))
- proc.stderr?.on("data", (data) => stderr.push(Buffer.from(data)))
-
- proc.on("close", (code) => {
- resolve({
- code: code ?? 1,
- stdout: Buffer.concat(stdout),
- stderr: Buffer.concat(stderr),
- })
- })
- })
-}
-
-function spawnWorker(msg: Msg) {
- return spawn(process.execPath, [worker, JSON.stringify(msg)], {
- cwd: root,
- stdio: ["ignore", "pipe", "pipe"],
- })
-}
-
-function stopWorker(proc: ReturnType<typeof spawnWorker>) {
- if (proc.exitCode !== null || proc.signalCode !== null) return Promise.resolve()
-
- if (process.platform !== "win32" || !proc.pid) {
- proc.kill()
- return Promise.resolve()
- }
-
- return new Promise<void>((resolve) => {
- const killProc = spawn("taskkill", ["/pid", String(proc.pid), "/T", "/F"])
- killProc.on("close", () => {
- proc.kill()
- resolve()
- })
- })
-}
-
-async function readJson<T>(p: string): Promise<T> {
- return JSON.parse(await fs.readFile(p, "utf8"))
-}
-
-describe("util.flock", () => {
- test("enforces mutual exclusion under process contention", async () => {
- await using tmp = await tmpdir()
- const dir = path.join(tmp.path, "locks")
- const done = path.join(tmp.path, "done.log")
- const active = path.join(tmp.path, "active")
- const key = "flock:stress"
- const n = 16
-
- const out = await Promise.all(
- Array.from({ length: n }, () =>
- run({
- key,
- dir,
- done,
- active,
- holdMs: 30,
- staleMs: 1_000,
- timeoutMs: 15_000,
- }),
- ),
- )
-
- expect(out.map((x) => x.code)).toEqual(Array.from({ length: n }, () => 0))
- expect(out.map((x) => x.stderr.toString()).filter(Boolean)).toEqual([])
-
- const lines = (await fs.readFile(done, "utf8"))
- .split("\n")
- .map((x) => x.trim())
- .filter(Boolean)
- expect(lines.length).toBe(n)
- }, 20_000)
-
- test("times out while waiting when lock is still healthy", async () => {
- await using tmp = await tmpdir()
- const dir = path.join(tmp.path, "locks")
- const key = "flock:timeout"
- const ready = path.join(tmp.path, "ready")
- const proc = spawnWorker({
- key,
- dir,
- ready,
- holdMs: 20_000,
- staleMs: 10_000,
- timeoutMs: 30_000,
- })
-
- try {
- await wait(ready, 5_000)
- const seen: string[] = []
- const err = await Flock.withLock(key, async () => {}, {
- dir,
- staleMs: 10_000,
- timeoutMs: 1_000,
- onWait: (tick) => {
- seen.push(tick.key)
- },
- }).catch((err) => err)
-
- expect(err).toBeInstanceOf(Error)
- if (!(err instanceof Error)) throw err
- expect(err.message).toContain("Timed out waiting for lock")
- expect(seen.length).toBeGreaterThan(0)
- expect(seen.every((x) => x === key)).toBe(true)
- } finally {
- await stopWorker(proc).catch(() => undefined)
- await new Promise((resolve) => proc.on("close", resolve))
- }
- }, 15_000)
-
- test("recovers after a crashed lock owner", async () => {
- await using tmp = await tmpdir()
- const dir = path.join(tmp.path, "locks")
- const key = "flock:crash"
- const ready = path.join(tmp.path, "ready")
- const proc = spawnWorker({
- key,
- dir,
- ready,
- holdMs: 20_000,
- staleMs: 500,
- timeoutMs: 30_000,
- })
-
- await wait(ready, 5_000)
- await stopWorker(proc)
- await new Promise((resolve) => proc.on("close", resolve))
-
- let hit = false
- await Flock.withLock(
- key,
- async () => {
- hit = true
- },
- {
- dir,
- staleMs: 500,
- timeoutMs: 8_000,
- },
- )
-
- expect(hit).toBe(true)
- }, 20_000)
-
- test("breaks stale lock dirs when heartbeat is missing", async () => {
- await using tmp = await tmpdir()
- const dir = path.join(tmp.path, "locks")
- const key = "flock:missing-heartbeat"
- const lockDir = lock(dir, key)
-
- await fs.mkdir(lockDir, { recursive: true })
- const old = new Date(Date.now() - 2_000)
- await fs.utimes(lockDir, old, old)
-
- let hit = false
- await Flock.withLock(
- key,
- async () => {
- hit = true
- },
- {
- dir,
- staleMs: 200,
- timeoutMs: 3_000,
- },
- )
-
- expect(hit).toBe(true)
- })
-
- test("recovers when a stale breaker claim was left behind", async () => {
- await using tmp = await tmpdir()
- const dir = path.join(tmp.path, "locks")
- const key = "flock:stale-breaker"
- const lockDir = lock(dir, key)
- const breaker = lockDir + ".breaker"
-
- await fs.mkdir(lockDir, { recursive: true })
- await fs.mkdir(breaker)
-
- const old = new Date(Date.now() - 2_000)
- await fs.utimes(lockDir, old, old)
- await fs.utimes(breaker, old, old)
-
- let hit = false
- await Flock.withLock(
- key,
- async () => {
- hit = true
- },
- {
- dir,
- staleMs: 200,
- timeoutMs: 3_000,
- },
- )
-
- expect(hit).toBe(true)
- expect(await exists(breaker)).toBe(false)
- })
-
- test("fails clearly if lock dir is removed while held", async () => {
- await using tmp = await tmpdir()
- const dir = path.join(tmp.path, "locks")
- const key = "flock:compromised"
- const lockDir = lock(dir, key)
-
- const err = await Flock.withLock(
- key,
- async () => {
- await fs.rm(lockDir, {
- recursive: true,
- force: true,
- })
- },
- {
- dir,
- staleMs: 1_000,
- timeoutMs: 3_000,
- },
- ).catch((err) => err)
-
- expect(err).toBeInstanceOf(Error)
- if (!(err instanceof Error)) throw err
- expect(err.message).toContain("compromised")
-
- let hit = false
- await Flock.withLock(
- key,
- async () => {
- hit = true
- },
- {
- dir,
- staleMs: 200,
- timeoutMs: 3_000,
- },
- )
- expect(hit).toBe(true)
- })
-
- test("writes owner metadata while lock is held", async () => {
- await using tmp = await tmpdir()
- const dir = path.join(tmp.path, "locks")
- const key = "flock:meta"
- const file = path.join(lock(dir, key), "meta.json")
-
- await Flock.withLock(
- key,
- async () => {
- const json = await readJson<{
- token?: unknown
- pid?: unknown
- hostname?: unknown
- createdAt?: unknown
- }>(file)
-
- expect(typeof json.token).toBe("string")
- expect(typeof json.pid).toBe("number")
- expect(typeof json.hostname).toBe("string")
- expect(typeof json.createdAt).toBe("string")
- },
- {
- dir,
- staleMs: 1_000,
- timeoutMs: 3_000,
- },
- )
- })
-
- test("supports acquire with await using", async () => {
- await using tmp = await tmpdir()
- const dir = path.join(tmp.path, "locks")
- const key = "flock:acquire"
- const lockDir = lock(dir, key)
-
- {
- await using _ = await Flock.acquire(key, {
- dir,
- staleMs: 1_000,
- timeoutMs: 3_000,
- })
- expect(await exists(lockDir)).toBe(true)
- }
-
- expect(await exists(lockDir)).toBe(false)
- })
-
- test("refuses token mismatch release and recovers from stale", async () => {
- await using tmp = await tmpdir()
- const dir = path.join(tmp.path, "locks")
- const key = "flock:token"
- const lockDir = lock(dir, key)
- const meta = path.join(lockDir, "meta.json")
-
- const err = await Flock.withLock(
- key,
- async () => {
- const json = await readJson<{ token?: string }>(meta)
- json.token = "tampered"
- await fs.writeFile(meta, JSON.stringify(json, null, 2))
- },
- {
- dir,
- staleMs: 500,
- timeoutMs: 3_000,
- },
- ).catch((err) => err)
-
- expect(err).toBeInstanceOf(Error)
- if (!(err instanceof Error)) throw err
- expect(err.message).toContain("token mismatch")
- expect(await exists(lockDir)).toBe(true)
-
- let hit = false
- await Flock.withLock(
- key,
- async () => {
- hit = true
- },
- {
- dir,
- staleMs: 500,
- timeoutMs: 6_000,
- },
- )
- expect(hit).toBe(true)
- })
-
- test("fails clearly on unwritable lock roots", async () => {
- if (process.platform === "win32") return
-
- await using tmp = await tmpdir()
- const dir = path.join(tmp.path, "locks")
- const key = "flock:perm"
-
- await fs.mkdir(dir, { recursive: true })
- await fs.chmod(dir, 0o500)
-
- try {
- const err = await Flock.withLock(key, async () => {}, {
- dir,
- staleMs: 100,
- timeoutMs: 500,
- }).catch((err) => err)
-
- expect(err).toBeInstanceOf(Error)
- if (!(err instanceof Error)) throw err
- const text = err.message
- expect(text.includes("EACCES") || text.includes("EPERM")).toBe(true)
- } finally {
- await fs.chmod(dir, 0o700)
- }
- })
-})