migrate: move flock and hash utilities to shared package (#22640)

This commit is contained in:
Dax
2026-04-15 11:50:24 -04:00
committed by GitHub
parent f1751401aa
commit 4ae7c77f8a
23 changed files with 929 additions and 41 deletions

View File

@@ -5,7 +5,9 @@
"type": "module",
"license": "MIT",
"private": true,
"scripts": {},
"scripts": {
"test": "bun test"
},
"bin": {
"opencode": "./bin/opencode"
},
@@ -14,7 +16,8 @@
},
"imports": {},
"devDependencies": {
"@types/semver": "catalog:"
"@types/semver": "catalog:",
"@types/bun": "catalog:"
},
"dependencies": {
"@effect/platform-node": "catalog:",
@@ -23,6 +26,7 @@
"mime-types": "3.0.2",
"minimatch": "10.2.5",
"semver": "catalog:",
"xdg-basedir": "5.1.0",
"zod": "catalog:"
},
"overrides": {

View File

@@ -0,0 +1,42 @@
import path from "path"
import { xdgData, xdgCache, xdgConfig, xdgState } from "xdg-basedir"
import os from "os"
import { Context, Effect, Layer } from "effect"
export namespace Global {
export class Service extends Context.Service<Service, Interface>()("@opencode/Global") {}
export interface Interface {
readonly home: string
readonly data: string
readonly cache: string
readonly config: string
readonly state: string
readonly bin: string
readonly log: string
}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const app = "opencode"
const home = process.env.OPENCODE_TEST_HOME ?? os.homedir()
const data = path.join(xdgData!, app)
const cache = path.join(xdgCache!, app)
const cfg = path.join(xdgConfig!, app)
const state = path.join(xdgState!, app)
const bin = path.join(cache, "bin")
const log = path.join(data, "log")
return Service.of({
home,
data,
cache,
config: cfg,
state,
bin,
log,
})
}),
)
}

247
packages/shared/src/npm.ts Normal file
View File

@@ -0,0 +1,247 @@
import path from "path"
import semver from "semver"
import { Arborist } from "@npmcli/arborist"
import { Effect, Schema, Context, Layer, Option, FileSystem } from "effect"
import { NodeFileSystem } from "@effect/platform-node"
import { AppFileSystem } from "./filesystem"
import { Global } from "./global"
import { Flock } from "./util/flock"
export namespace Npm {
export class InstallFailedError extends Schema.TaggedErrorClass<InstallFailedError>()("NpmInstallFailedError", {
pkg: Schema.String,
cause: Schema.optional(Schema.Defect),
}) {}
export interface EntryPoint {
readonly directory: string
readonly entrypoint: Option.Option<string>
}
export interface Interface {
readonly add: (pkg: string) => Effect.Effect<EntryPoint, InstallFailedError>
readonly install: (dir: string) => Effect.Effect<void>
readonly outdated: (pkg: string, cachedVersion: string) => Effect.Effect<boolean>
readonly which: (pkg: string) => Effect.Effect<Option.Option<string>>
}
export class Service extends Context.Service<Service, Interface>()("@opencode/Npm") {}
const illegal = process.platform === "win32" ? new Set(["<", ">", ":", '"', "|", "?", "*"]) : undefined
export function sanitize(pkg: string) {
if (!illegal) return pkg
return Array.from(pkg, (char) => (illegal.has(char) || char.charCodeAt(0) < 32 ? "_" : char)).join("")
}
const resolveEntryPoint = (name: string, dir: string): EntryPoint => {
let entrypoint: Option.Option<string>
try {
const resolved = typeof Bun !== "undefined" ? import.meta.resolve(name, dir) : import.meta.resolve(dir)
entrypoint = Option.some(resolved)
} catch {
entrypoint = Option.none()
}
return {
directory: dir,
entrypoint,
}
}
interface ArboristNode {
name: string
path: string
}
interface ArboristTree {
edgesOut: Map<string, { to?: ArboristNode }>
}
export const layer = Layer.effect(
Service,
Effect.gen(function* () {
const afs = yield* AppFileSystem.Service
const global = yield* Global.Service
const fs = yield* FileSystem.FileSystem
const directory = (pkg: string) => path.join(global.cache, "packages", sanitize(pkg))
const outdated = Effect.fn("Npm.outdated")(function* (pkg: string, cachedVersion: string) {
const response = yield* Effect.tryPromise({
try: () => fetch(`https://registry.npmjs.org/${pkg}`),
catch: () => undefined,
}).pipe(Effect.orElseSucceed(() => undefined))
if (!response || !response.ok) {
return false
}
const data = yield* Effect.tryPromise({
try: () => response.json() as Promise<{ "dist-tags"?: { latest?: string } }>,
catch: () => undefined,
}).pipe(Effect.orElseSucceed(() => undefined))
const latestVersion = data?.["dist-tags"]?.latest
if (!latestVersion) {
return false
}
const range = /[\s^~*xX<>|=]/.test(cachedVersion)
if (range) return !semver.satisfies(latestVersion, cachedVersion)
return semver.lt(cachedVersion, latestVersion)
})
const add = Effect.fn("Npm.add")(function* (pkg: string) {
const dir = directory(pkg)
yield* Flock.effect(`npm-install:${dir}`)
const arborist = new Arborist({
path: dir,
binLinks: true,
progress: false,
savePrefix: "",
ignoreScripts: true,
})
const tree = yield* Effect.tryPromise({
try: () => arborist.loadVirtual().catch(() => undefined),
catch: () => undefined,
}).pipe(Effect.orElseSucceed(() => undefined)) as Effect.Effect<ArboristTree | undefined>
if (tree) {
const first = tree.edgesOut.values().next().value?.to
if (first) {
return resolveEntryPoint(first.name, first.path)
}
}
const result = yield* Effect.tryPromise({
try: () =>
arborist.reify({
add: [pkg],
save: true,
saveType: "prod",
}),
catch: (cause) => new InstallFailedError({ pkg, cause }),
}) as Effect.Effect<ArboristTree, InstallFailedError>
const first = result.edgesOut.values().next().value?.to
if (!first) {
return yield* new InstallFailedError({ pkg })
}
return resolveEntryPoint(first.name, first.path)
}, Effect.scoped)
const install = Effect.fn("Npm.install")(function* (dir: string) {
yield* Flock.effect(`npm-install:${dir}`)
const reify = Effect.fnUntraced(function* () {
const arb = new Arborist({
path: dir,
binLinks: true,
progress: false,
savePrefix: "",
ignoreScripts: true,
})
yield* Effect.tryPromise({
try: () => arb.reify().catch(() => {}),
catch: () => {},
}).pipe(Effect.orElseSucceed(() => {}))
})
const nodeModulesExists = yield* afs.existsSafe(path.join(dir, "node_modules"))
if (!nodeModulesExists) {
yield* reify()
return
}
const pkg = yield* afs.readJson(path.join(dir, "package.json")).pipe(Effect.orElseSucceed(() => ({})))
const lock = yield* afs.readJson(path.join(dir, "package-lock.json")).pipe(Effect.orElseSucceed(() => ({})))
const pkgAny = pkg as any
const lockAny = lock as any
const declared = new Set([
...Object.keys(pkgAny?.dependencies || {}),
...Object.keys(pkgAny?.devDependencies || {}),
...Object.keys(pkgAny?.peerDependencies || {}),
...Object.keys(pkgAny?.optionalDependencies || {}),
])
const root = lockAny?.packages?.[""] || {}
const locked = new Set([
...Object.keys(root?.dependencies || {}),
...Object.keys(root?.devDependencies || {}),
...Object.keys(root?.peerDependencies || {}),
...Object.keys(root?.optionalDependencies || {}),
])
for (const name of declared) {
if (!locked.has(name)) {
yield* reify()
return
}
}
}, Effect.scoped)
const which = Effect.fn("Npm.which")(function* (pkg: string) {
const dir = directory(pkg)
const binDir = path.join(dir, "node_modules", ".bin")
const pick = Effect.fnUntraced(function* () {
const files = yield* fs.readDirectory(binDir).pipe(Effect.catch(() => Effect.succeed([] as string[])))
if (files.length === 0) return Option.none<string>()
if (files.length === 1) return Option.some(files[0])
const pkgJson = yield* afs.readJson(path.join(dir, "node_modules", pkg, "package.json")).pipe(Effect.option)
if (Option.isSome(pkgJson)) {
const parsed = pkgJson.value as { bin?: string | Record<string, string> }
if (parsed?.bin) {
const unscoped = pkg.startsWith("@") ? pkg.split("/")[1] : pkg
const bin = parsed.bin
if (typeof bin === "string") return Option.some(unscoped)
const keys = Object.keys(bin)
if (keys.length === 1) return Option.some(keys[0])
return bin[unscoped] ? Option.some(unscoped) : Option.some(keys[0])
}
}
return Option.some(files[0])
})
return yield* Effect.gen(function* () {
const bin = yield* pick()
if (Option.isSome(bin)) {
return Option.some(path.join(binDir, bin.value))
}
yield* fs.remove(path.join(dir, "package-lock.json")).pipe(Effect.orElseSucceed(() => {}))
yield* add(pkg)
const resolved = yield* pick()
if (Option.isNone(resolved)) return Option.none<string>()
return Option.some(path.join(binDir, resolved.value))
}).pipe(
Effect.scoped,
Effect.orElseSucceed(() => Option.none<string>()),
)
})
return Service.of({
add,
install,
outdated,
which,
})
}),
)
export const defaultLayer = layer.pipe(
Layer.provide(AppFileSystem.layer),
Layer.provide(Global.layer),
Layer.provide(NodeFileSystem.layer),
)
}

44
packages/shared/src/types.d.ts vendored Normal file
View File

@@ -0,0 +1,44 @@
declare module "@npmcli/arborist" {
export interface ArboristOptions {
path: string
binLinks?: boolean
progress?: boolean
savePrefix?: string
ignoreScripts?: boolean
}
export interface ArboristNode {
name: string
path: string
}
export interface ArboristEdge {
to?: ArboristNode
}
export interface ArboristTree {
edgesOut: Map<string, ArboristEdge>
}
export interface ReifyOptions {
add?: string[]
save?: boolean
saveType?: "prod" | "dev" | "optional" | "peer"
}
export class Arborist {
constructor(options: ArboristOptions)
loadVirtual(): Promise<ArboristTree | undefined>
reify(options?: ReifyOptions): Promise<ArboristTree>
}
}
declare var Bun:
| {
file(path: string): {
text(): Promise<string>
json(): Promise<unknown>
}
write(path: string, content: string | Uint8Array): Promise<void>
}
| undefined

View File

@@ -0,0 +1,354 @@
import path from "path"
import os from "os"
import { randomBytes, randomUUID } from "crypto"
import { mkdir, readFile, rm, stat, utimes, writeFile } from "fs/promises"
import { Hash } from "./hash"
import { Effect } from "effect"
export type FlockGlobal = {
state: string
}
export namespace Flock {
let global: FlockGlobal | undefined
export function setGlobal(g: FlockGlobal) {
global = g
}
const root = () => {
if (!global) throw new Error("Flock global not set")
return path.join(global.state, "locks")
}
// Defaults for callers that do not provide timing options.
const defaultOpts = {
staleMs: 60_000,
timeoutMs: 5 * 60_000,
baseDelayMs: 100,
maxDelayMs: 2_000,
}
export interface WaitEvent {
key: string
attempt: number
delay: number
waited: number
}
export type Wait = (input: WaitEvent) => void | Promise<void>
export interface Options {
dir?: string
signal?: AbortSignal
staleMs?: number
timeoutMs?: number
baseDelayMs?: number
maxDelayMs?: number
onWait?: Wait
}
type Opts = {
staleMs: number
timeoutMs: number
baseDelayMs: number
maxDelayMs: number
}
type Owned = {
acquired: true
startHeartbeat: (intervalMs?: number) => void
release: () => Promise<void>
}
export interface Lease {
release: () => Promise<void>
[Symbol.asyncDispose]: () => Promise<void>
}
function code(err: unknown) {
if (typeof err !== "object" || err === null || !("code" in err)) return
const value = err.code
if (typeof value !== "string") return
return value
}
function sleep(ms: number, signal?: AbortSignal) {
return new Promise<void>((resolve, reject) => {
if (signal?.aborted) {
reject(signal.reason ?? new Error("Aborted"))
return
}
let timer: NodeJS.Timeout | undefined
const done = () => {
signal?.removeEventListener("abort", abort)
resolve()
}
const abort = () => {
if (timer) {
clearTimeout(timer)
}
signal?.removeEventListener("abort", abort)
reject(signal?.reason ?? new Error("Aborted"))
}
signal?.addEventListener("abort", abort, { once: true })
timer = setTimeout(done, ms)
})
}
function jitter(ms: number) {
const j = Math.floor(ms * 0.3)
const d = Math.floor(Math.random() * (2 * j + 1)) - j
return Math.max(0, ms + d)
}
function mono() {
return performance.now()
}
function wall() {
return performance.timeOrigin + mono()
}
async function stats(file: string) {
try {
return await stat(file)
} catch (err) {
const errCode = code(err)
if (errCode === "ENOENT" || errCode === "ENOTDIR") return
throw err
}
}
async function stale(lockDir: string, heartbeatPath: string, metaPath: string, staleMs: number) {
// Stale detection allows automatic recovery after crashed owners.
const now = wall()
const heartbeat = await stats(heartbeatPath)
if (heartbeat) {
return now - heartbeat.mtimeMs > staleMs
}
const meta = await stats(metaPath)
if (meta) {
return now - meta.mtimeMs > staleMs
}
const dir = await stats(lockDir)
if (!dir) {
return false
}
return now - dir.mtimeMs > staleMs
}
async function tryAcquireLockDir(lockDir: string, opts: Opts): Promise<Owned | { acquired: false }> {
const token = randomUUID?.() ?? randomBytes(16).toString("hex")
const metaPath = path.join(lockDir, "meta.json")
const heartbeatPath = path.join(lockDir, "heartbeat")
try {
await mkdir(lockDir, { mode: 0o700 })
} catch (err) {
if (code(err) !== "EEXIST") {
throw err
}
if (!(await stale(lockDir, heartbeatPath, metaPath, opts.staleMs))) {
return { acquired: false }
}
const breakerPath = lockDir + ".breaker"
try {
await mkdir(breakerPath, { mode: 0o700 })
} catch (claimErr) {
const errCode = code(claimErr)
if (errCode === "EEXIST") {
const breaker = await stats(breakerPath)
if (breaker && wall() - breaker.mtimeMs > opts.staleMs) {
await rm(breakerPath, { recursive: true, force: true }).catch(() => undefined)
}
return { acquired: false }
}
if (errCode === "ENOENT" || errCode === "ENOTDIR") {
return { acquired: false }
}
throw claimErr
}
try {
// Breaker ownership ensures only one contender performs stale cleanup.
if (!(await stale(lockDir, heartbeatPath, metaPath, opts.staleMs))) {
return { acquired: false }
}
await rm(lockDir, { recursive: true, force: true })
try {
await mkdir(lockDir, { mode: 0o700 })
} catch (retryErr) {
const errCode = code(retryErr)
if (errCode === "EEXIST" || errCode === "ENOTEMPTY") {
return { acquired: false }
}
throw retryErr
}
} finally {
await rm(breakerPath, { recursive: true, force: true }).catch(() => undefined)
}
}
const meta = {
token,
pid: process.pid,
hostname: os.hostname(),
createdAt: new Date().toISOString(),
}
await writeFile(heartbeatPath, "", { flag: "wx" }).catch(async () => {
await rm(lockDir, { recursive: true, force: true })
throw new Error("Lock acquired but heartbeat already existed (possible compromise).")
})
await writeFile(metaPath, JSON.stringify(meta, null, 2), { flag: "wx" }).catch(async () => {
await rm(lockDir, { recursive: true, force: true })
throw new Error("Lock acquired but meta.json already existed (possible compromise).")
})
let timer: NodeJS.Timeout | undefined
const startHeartbeat = (intervalMs = Math.max(100, Math.floor(opts.staleMs / 3))) => {
if (timer) return
// Heartbeat prevents long critical sections from being evicted as stale.
timer = setInterval(() => {
const t = new Date()
void utimes(heartbeatPath, t, t).catch(() => undefined)
}, intervalMs)
timer.unref?.()
}
const release = async () => {
if (timer) {
clearInterval(timer)
timer = undefined
}
const current = await readFile(metaPath, "utf8")
.then((raw) => {
const parsed = JSON.parse(raw)
if (!parsed || typeof parsed !== "object") return {}
return {
token: "token" in parsed && typeof parsed.token === "string" ? parsed.token : undefined,
}
})
.catch((err) => {
const errCode = code(err)
if (errCode === "ENOENT" || errCode === "ENOTDIR") {
throw new Error("Refusing to release: lock is compromised (metadata missing).")
}
if (err instanceof SyntaxError) {
throw new Error("Refusing to release: lock is compromised (metadata invalid).")
}
throw err
})
// Token check prevents deleting a lock that was re-acquired by another process.
if (current.token !== token) {
throw new Error("Refusing to release: lock token mismatch (not the owner).")
}
await rm(lockDir, { recursive: true, force: true })
}
return {
acquired: true,
startHeartbeat,
release,
}
}
async function acquireLockDir(
lockDir: string,
input: { key: string; onWait?: Wait; signal?: AbortSignal },
opts: Opts,
) {
const stop = mono() + opts.timeoutMs
let attempt = 0
let waited = 0
let delay = opts.baseDelayMs
while (true) {
input.signal?.throwIfAborted()
const res = await tryAcquireLockDir(lockDir, opts)
if (res.acquired) {
return res
}
if (mono() > stop) {
throw new Error(`Timed out waiting for lock: ${input.key}`)
}
attempt += 1
const ms = jitter(delay)
await input.onWait?.({
key: input.key,
attempt,
delay: ms,
waited,
})
await sleep(ms, input.signal)
waited += ms
delay = Math.min(opts.maxDelayMs, Math.floor(delay * 1.7))
}
}
export async function acquire(key: string, input: Options = {}): Promise<Lease> {
input.signal?.throwIfAborted()
const cfg: Opts = {
staleMs: input.staleMs ?? defaultOpts.staleMs,
timeoutMs: input.timeoutMs ?? defaultOpts.timeoutMs,
baseDelayMs: input.baseDelayMs ?? defaultOpts.baseDelayMs,
maxDelayMs: input.maxDelayMs ?? defaultOpts.maxDelayMs,
}
const dir = input.dir ?? root()
await mkdir(dir, { recursive: true })
const lockfile = path.join(dir, Hash.fast(key) + ".lock")
const lock = await acquireLockDir(
lockfile,
{
key,
onWait: input.onWait,
signal: input.signal,
},
cfg,
)
lock.startHeartbeat()
const release = () => lock.release()
return {
release,
[Symbol.asyncDispose]() {
return release()
},
}
}
export async function withLock<T>(key: string, fn: () => Promise<T>, input: Options = {}) {
await using _ = await acquire(key, input)
input.signal?.throwIfAborted()
return await fn()
}
export const effect = Effect.fn("Flock.effect")(function* (key: string) {
return yield* Effect.acquireRelease(
Effect.promise((signal) => Flock.acquire(key, { signal })),
(foo) => Effect.promise(() => foo.release()),
).pipe(Effect.asVoid)
})
}

View File

@@ -0,0 +1,7 @@
import { createHash } from "crypto"
export namespace Hash {
export function fast(input: string | Buffer): string {
return createHash("sha1").update(input).digest("hex")
}
}

View File

@@ -0,0 +1,338 @@
import { describe, test, expect } from "bun:test"
import { Effect, Layer, FileSystem } from "effect"
import { NodeFileSystem } from "@effect/platform-node"
import { AppFileSystem } from "@opencode-ai/shared/filesystem"
import { testEffect } from "../lib/effect"
import path from "path"
const live = AppFileSystem.layer.pipe(Layer.provideMerge(NodeFileSystem.layer))
const { effect: it } = testEffect(live)
describe("AppFileSystem", () => {
describe("isDir", () => {
it(
"returns true for directories",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
expect(yield* fs.isDir(tmp)).toBe(true)
}),
)
it(
"returns false for files",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const file = path.join(tmp, "test.txt")
yield* filesys.writeFileString(file, "hello")
expect(yield* fs.isDir(file)).toBe(false)
}),
)
it(
"returns false for non-existent paths",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
expect(yield* fs.isDir("/tmp/nonexistent-" + Math.random())).toBe(false)
}),
)
})
describe("isFile", () => {
it(
"returns true for files",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const file = path.join(tmp, "test.txt")
yield* filesys.writeFileString(file, "hello")
expect(yield* fs.isFile(file)).toBe(true)
}),
)
it(
"returns false for directories",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
expect(yield* fs.isFile(tmp)).toBe(false)
}),
)
})
describe("readJson / writeJson", () => {
it(
"round-trips JSON data",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const file = path.join(tmp, "data.json")
const data = { name: "test", count: 42, nested: { ok: true } }
yield* fs.writeJson(file, data)
const result = yield* fs.readJson(file)
expect(result).toEqual(data)
}),
)
})
describe("ensureDir", () => {
it(
"creates nested directories",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const nested = path.join(tmp, "a", "b", "c")
yield* fs.ensureDir(nested)
const info = yield* filesys.stat(nested)
expect(info.type).toBe("Directory")
}),
)
it(
"is idempotent",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const dir = path.join(tmp, "existing")
yield* filesys.makeDirectory(dir)
yield* fs.ensureDir(dir)
const info = yield* filesys.stat(dir)
expect(info.type).toBe("Directory")
}),
)
})
describe("writeWithDirs", () => {
it(
"creates parent directories if missing",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const file = path.join(tmp, "deep", "nested", "file.txt")
yield* fs.writeWithDirs(file, "hello")
expect(yield* filesys.readFileString(file)).toBe("hello")
}),
)
it(
"writes directly when parent exists",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const file = path.join(tmp, "direct.txt")
yield* fs.writeWithDirs(file, "world")
expect(yield* filesys.readFileString(file)).toBe("world")
}),
)
it(
"writes Uint8Array content",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const file = path.join(tmp, "binary.bin")
const content = new Uint8Array([0x00, 0x01, 0x02, 0x03])
yield* fs.writeWithDirs(file, content)
const result = yield* filesys.readFile(file)
expect(new Uint8Array(result)).toEqual(content)
}),
)
})
describe("findUp", () => {
it(
"finds target in start directory",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
yield* filesys.writeFileString(path.join(tmp, "target.txt"), "found")
const result = yield* fs.findUp("target.txt", tmp)
expect(result).toEqual([path.join(tmp, "target.txt")])
}),
)
it(
"finds target in parent directories",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
yield* filesys.writeFileString(path.join(tmp, "marker"), "root")
const child = path.join(tmp, "a", "b")
yield* filesys.makeDirectory(child, { recursive: true })
const result = yield* fs.findUp("marker", child, tmp)
expect(result).toEqual([path.join(tmp, "marker")])
}),
)
it(
"returns empty array when not found",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const result = yield* fs.findUp("nonexistent", tmp, tmp)
expect(result).toEqual([])
}),
)
})
describe("up", () => {
it(
"finds multiple targets walking up",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
yield* filesys.writeFileString(path.join(tmp, "a.txt"), "a")
yield* filesys.writeFileString(path.join(tmp, "b.txt"), "b")
const child = path.join(tmp, "sub")
yield* filesys.makeDirectory(child)
yield* filesys.writeFileString(path.join(child, "a.txt"), "a-child")
const result = yield* fs.up({ targets: ["a.txt", "b.txt"], start: child, stop: tmp })
expect(result).toContain(path.join(child, "a.txt"))
expect(result).toContain(path.join(tmp, "a.txt"))
expect(result).toContain(path.join(tmp, "b.txt"))
}),
)
})
describe("glob", () => {
it(
"finds files matching pattern",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
yield* filesys.writeFileString(path.join(tmp, "a.ts"), "a")
yield* filesys.writeFileString(path.join(tmp, "b.ts"), "b")
yield* filesys.writeFileString(path.join(tmp, "c.json"), "c")
const result = yield* fs.glob("*.ts", { cwd: tmp })
expect(result.sort()).toEqual(["a.ts", "b.ts"])
}),
)
it(
"supports absolute paths",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
yield* filesys.writeFileString(path.join(tmp, "file.txt"), "hello")
const result = yield* fs.glob("*.txt", { cwd: tmp, absolute: true })
expect(result).toEqual([path.join(tmp, "file.txt")])
}),
)
})
describe("globMatch", () => {
it(
"matches patterns",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
expect(fs.globMatch("*.ts", "foo.ts")).toBe(true)
expect(fs.globMatch("*.ts", "foo.json")).toBe(false)
expect(fs.globMatch("src/**", "src/a/b.ts")).toBe(true)
}),
)
})
describe("globUp", () => {
it(
"finds files walking up directories",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
yield* filesys.writeFileString(path.join(tmp, "root.md"), "root")
const child = path.join(tmp, "a", "b")
yield* filesys.makeDirectory(child, { recursive: true })
yield* filesys.writeFileString(path.join(child, "leaf.md"), "leaf")
const result = yield* fs.globUp("*.md", child, tmp)
expect(result).toContain(path.join(child, "leaf.md"))
expect(result).toContain(path.join(tmp, "root.md"))
}),
)
})
describe("built-in passthrough", () => {
it(
"exists works",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const file = path.join(tmp, "exists.txt")
yield* filesys.writeFileString(file, "yes")
expect(yield* filesys.exists(file)).toBe(true)
expect(yield* filesys.exists(file + ".nope")).toBe(false)
}),
)
it(
"remove works",
Effect.gen(function* () {
const fs = yield* AppFileSystem.Service
const filesys = yield* FileSystem.FileSystem
const tmp = yield* filesys.makeTempDirectoryScoped()
const file = path.join(tmp, "delete-me.txt")
yield* filesys.writeFileString(file, "bye")
yield* filesys.remove(file)
expect(yield* filesys.exists(file)).toBe(false)
}),
)
})
describe("pure helpers", () => {
test("mimeType returns correct types", () => {
expect(AppFileSystem.mimeType("file.json")).toBe("application/json")
expect(AppFileSystem.mimeType("image.png")).toBe("image/png")
expect(AppFileSystem.mimeType("unknown.qzx")).toBe("application/octet-stream")
})
test("contains checks path containment", () => {
expect(AppFileSystem.contains("/a/b", "/a/b/c")).toBe(true)
expect(AppFileSystem.contains("/a/b", "/a/c")).toBe(false)
})
test("overlaps detects overlapping paths", () => {
expect(AppFileSystem.overlaps("/a/b", "/a/b/c")).toBe(true)
expect(AppFileSystem.overlaps("/a/b/c", "/a/b")).toBe(true)
expect(AppFileSystem.overlaps("/a", "/b")).toBe(false)
})
})
})

View File

@@ -0,0 +1,72 @@
import fs from "fs/promises"
import { Flock } from "@opencode-ai/shared/util/flock"
type Msg = {
key: string
dir: string
staleMs?: number
timeoutMs?: number
baseDelayMs?: number
maxDelayMs?: number
holdMs?: number
ready?: string
active?: string
done?: string
}
function sleep(ms: number) {
return new Promise<void>((resolve) => {
setTimeout(resolve, ms)
})
}
function input() {
const raw = process.argv[2]
if (!raw) {
throw new Error("Missing flock worker input")
}
return JSON.parse(raw) as Msg
}
async function job(input: Msg) {
if (input.ready) {
await fs.writeFile(input.ready, String(process.pid))
}
if (input.active) {
await fs.writeFile(input.active, String(process.pid), { flag: "wx" })
}
try {
if (input.holdMs && input.holdMs > 0) {
await sleep(input.holdMs)
}
if (input.done) {
await fs.appendFile(input.done, "1\n")
}
} finally {
if (input.active) {
await fs.rm(input.active, { force: true })
}
}
}
async function main() {
const msg = input()
await Flock.withLock(msg.key, () => job(msg), {
dir: msg.dir,
staleMs: msg.staleMs,
timeoutMs: msg.timeoutMs,
baseDelayMs: msg.baseDelayMs,
maxDelayMs: msg.maxDelayMs,
})
}
await main().catch((err) => {
const text = err instanceof Error ? (err.stack ?? err.message) : String(err)
process.stderr.write(text)
process.exit(1)
})

View File

@@ -0,0 +1,53 @@
import { test, type TestOptions } from "bun:test"
import { Cause, Effect, Exit, Layer } from "effect"
import type * as Scope from "effect/Scope"
import * as TestClock from "effect/testing/TestClock"
import * as TestConsole from "effect/testing/TestConsole"
type Body<A, E, R> = Effect.Effect<A, E, R> | (() => Effect.Effect<A, E, R>)
const body = <A, E, R>(value: Body<A, E, R>) => Effect.suspend(() => (typeof value === "function" ? value() : value))
const run = <A, E, R, E2>(value: Body<A, E, R | Scope.Scope>, layer: Layer.Layer<R, E2>) =>
Effect.gen(function* () {
const exit = yield* body(value).pipe(Effect.scoped, Effect.provide(layer), Effect.exit)
if (Exit.isFailure(exit)) {
for (const err of Cause.prettyErrors(exit.cause)) {
yield* Effect.logError(err)
}
}
return yield* exit
}).pipe(Effect.runPromise)
const make = <R, E>(testLayer: Layer.Layer<R, E>, liveLayer: Layer.Layer<R, E>) => {
const effect = <A, E2>(name: string, value: Body<A, E2, R | Scope.Scope>, opts?: number | TestOptions) =>
test(name, () => run(value, testLayer), opts)
effect.only = <A, E2>(name: string, value: Body<A, E2, R | Scope.Scope>, opts?: number | TestOptions) =>
test.only(name, () => run(value, testLayer), opts)
effect.skip = <A, E2>(name: string, value: Body<A, E2, R | Scope.Scope>, opts?: number | TestOptions) =>
test.skip(name, () => run(value, testLayer), opts)
const live = <A, E2>(name: string, value: Body<A, E2, R | Scope.Scope>, opts?: number | TestOptions) =>
test(name, () => run(value, liveLayer), opts)
live.only = <A, E2>(name: string, value: Body<A, E2, R | Scope.Scope>, opts?: number | TestOptions) =>
test.only(name, () => run(value, liveLayer), opts)
live.skip = <A, E2>(name: string, value: Body<A, E2, R | Scope.Scope>, opts?: number | TestOptions) =>
test.skip(name, () => run(value, liveLayer), opts)
return { effect, live }
}
// Test environment with TestClock and TestConsole
const testEnv = Layer.mergeAll(TestConsole.layer, TestClock.layer())
// Live environment - uses real clock, but keeps TestConsole for output capture
const liveEnv = TestConsole.layer
export const it = make(testEnv, liveEnv)
export const testEffect = <R, E>(layer: Layer.Layer<R, E>) =>
make(Layer.provideMerge(layer, testEnv), Layer.provideMerge(layer, liveEnv))

View File

@@ -0,0 +1,18 @@
import { describe, expect, test } from "bun:test"
import { Npm } from "@opencode-ai/shared/npm"
const win = process.platform === "win32"
describe("Npm.sanitize", () => {
test("keeps normal scoped package specs unchanged", () => {
expect(Npm.sanitize("@opencode/acme")).toBe("@opencode/acme")
expect(Npm.sanitize("@opencode/acme@1.0.0")).toBe("@opencode/acme@1.0.0")
expect(Npm.sanitize("prettier")).toBe("prettier")
})
test("handles git https specs", () => {
const spec = "acme@git+https://github.com/opencode/acme.git"
const expected = win ? "acme@git+https_//github.com/opencode/acme.git" : spec
expect(Npm.sanitize(spec)).toBe(expected)
})
})

View File

@@ -0,0 +1,426 @@
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)
}
})
})