iamtoruk
272da806e4
The 0.9.20 menubar never completed a fetch on a large corpus: the cache version bump forced a full rehydration, DataClient's fixed 45s kill ended it mid-transaction, cache-refresh-lock had no signal cleanup so the dead holder's lock survived, and DEFAULT_WAIT_MS (30s) < DEFAULT_STALE_MS (90s) meant no waiter could ever recover it. A. Swift port of #1096's watchdog. CLIWatchdog holds the constants and the pure verdict; spawns and the resident serve child set CODEBURN_PROGRESS=1; the window restarts on any stdout/stderr byte; 45s silence, 10min cold floor until the first payload, 15min ceiling, SIGTERM then SIGKILL after 5s. ServeConnection's fixed 60s warm cap becomes the same silence window, re-armed by each progress frame, and a spent death budget is now a 5-minute cooldown instead of disabling the resident for the app run. B. cache-refresh-lock arms SIGINT/SIGTERM cleanup the way session-cache does for hydrating.lock, so a SIGTERMed holder unlinks its own lock. C. Staleness also opens on a dead holder pid, and the waiter budget derives from staleMs so it can never expire before the gate it waits for. A live holder - fresh heartbeat, pid answers signal 0 - is never taken from. parser.ts heartbeats through the lock wait, the one silent stretch left. D. The app closes its end of a retired child's stdin (dropping the handle left the pipe alive inside the Process), reaps every serve child synchronously at quit, and records pid+argv so a crash-orphaned child is reaped next launch. serve's final exit no longer runs through a monkeypatched process.exit, and its post-drain cleanup is bounded. Closes #1117
242 lines
11 KiB
TypeScript
242 lines
11 KiB
TypeScript
import { afterEach, describe, expect, it } from 'vitest'
|
|
import { spawn, type ChildProcess } from 'child_process'
|
|
import { existsSync } from 'fs'
|
|
import { mkdir, mkdtemp, readdir, rm, stat, utimes, writeFile } from 'fs/promises'
|
|
import { tmpdir } from 'os'
|
|
import { join } from 'path'
|
|
|
|
import { emptyCache, loadCache, saveCache } from '../src/session-cache.js'
|
|
|
|
const roots: string[] = []
|
|
|
|
async function waitFor(path: string, timeoutMs = 5_000): Promise<void> {
|
|
const deadline = Date.now() + timeoutMs
|
|
while (!existsSync(path)) {
|
|
if (Date.now() >= deadline) throw new Error(`timed out waiting for ${path}`)
|
|
await new Promise(resolve => { setTimeout(resolve, 5) })
|
|
}
|
|
}
|
|
|
|
async function waitForAny(dir: string, names: string[], timeoutMs = 5_000): Promise<string> {
|
|
const deadline = Date.now() + timeoutMs
|
|
while (Date.now() < deadline) {
|
|
for (const name of names) if (existsSync(join(dir, name))) return name
|
|
await new Promise(resolve => { setTimeout(resolve, 5) })
|
|
}
|
|
throw new Error(`timed out waiting for ${names.join(', ')}; saw ${(await readdir(dir)).join(', ')}`)
|
|
}
|
|
|
|
function waitForExit(child: ChildProcess): Promise<void> {
|
|
if (child.exitCode !== null) return child.exitCode === 0 ? Promise.resolve() : Promise.reject(new Error(`worker exited ${child.exitCode}`))
|
|
return new Promise((resolve, reject) => {
|
|
let stderr = ''
|
|
child.stderr?.on('data', chunk => { stderr += String(chunk) })
|
|
child.once('error', reject)
|
|
child.once('exit', code => code === 0 ? resolve() : reject(new Error(`worker exited ${code}: ${stderr}`)))
|
|
})
|
|
}
|
|
|
|
function worker(cacheDir: string, barriers: string, id: string, source: string, bypass = false): ChildProcess {
|
|
return spawn(process.execPath, ['--import', 'tsx', join(process.cwd(), 'tests/fixtures/cache-refresh-worker.ts'), cacheDir, barriers, id, source, String(bypass)], {
|
|
cwd: process.cwd(),
|
|
stdio: ['ignore', 'ignore', 'pipe'],
|
|
})
|
|
}
|
|
|
|
afterEach(async () => {
|
|
delete process.env['CODEBURN_CACHE_DIR']
|
|
await Promise.all(roots.splice(0).map(root => rm(root, { recursive: true, force: true })))
|
|
})
|
|
|
|
describe('warm refresh child-process regression', () => {
|
|
it('gives exactly one contender ownership of a stale lock', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'cb-refresh-stale-'))
|
|
roots.push(root)
|
|
const cacheDir = join(root, 'cache')
|
|
const barriers = join(root, 'barriers')
|
|
await mkdir(cacheDir, { recursive: true })
|
|
await mkdir(barriers, { recursive: true })
|
|
process.env['CODEBURN_CACHE_DIR'] = cacheDir
|
|
const initial = emptyCache()
|
|
initial.complete = true
|
|
await saveCache(initial)
|
|
const stalePath = join(cacheDir, 'session-refresh.lock')
|
|
await writeFile(stalePath, JSON.stringify({ pid: 1, token: 'abandoned', at: 1 }))
|
|
await utimes(stalePath, new Date(1), new Date(1))
|
|
const source = join(root, 'changed.json')
|
|
await writeFile(source, JSON.stringify({ output: 303 }))
|
|
|
|
const a = worker(cacheDir, barriers, 'a', source)
|
|
const b = worker(cacheDir, barriers, 'b', source)
|
|
const winner = await Promise.race([
|
|
waitFor(join(barriers, 'a.parsed')).then(() => 'a'),
|
|
waitFor(join(barriers, 'b.parsed')).then(() => 'b'),
|
|
])
|
|
const loser = winner === 'a' ? 'b' : 'a'
|
|
expect(Number(existsSync(join(barriers, 'a.parsed'))) + Number(existsSync(join(barriers, 'b.parsed')))).toBe(1)
|
|
// Keep the winner alive through the loser's full waiter budget: a second
|
|
// stale contender must never publish or steal a heartbeating successor.
|
|
const loserOutcome = await waitForAny(barriers, [
|
|
`${loser}.timed-out`, `${loser}.parsed`, `${loser}.completed-by-other`, `${loser}.unavailable`,
|
|
])
|
|
// Exactly one owner publishes. The loser of a stale-lock contest is not
|
|
// the owner: `timed-out` is the usual wait-out, but missing+missing during
|
|
// the winner's unlink-guard/create-successor gap is honestly
|
|
// `completed-by-other` (#904). Do not require timed-out.
|
|
expect(loserOutcome, (await readdir(barriers)).join(',')).not.toBe(`${loser}.parsed`)
|
|
expect([`${loser}.timed-out`, `${loser}.completed-by-other`, `${loser}.unavailable`]).toContain(loserOutcome)
|
|
await writeFile(join(barriers, `${winner}.save`), '')
|
|
await Promise.all([waitForExit(a), waitForExit(b)])
|
|
await expect(stat(join(cacheDir, 'session-refresh.lock.takeover'))).rejects.toMatchObject({ code: 'ENOENT' })
|
|
})
|
|
|
|
it('gives exactly one contender ownership of a stale zero-byte lock', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'cb-refresh-corrupt-'))
|
|
roots.push(root)
|
|
const cacheDir = join(root, 'cache')
|
|
const barriers = join(root, 'barriers')
|
|
await mkdir(cacheDir, { recursive: true })
|
|
await mkdir(barriers, { recursive: true })
|
|
process.env['CODEBURN_CACHE_DIR'] = cacheDir
|
|
const initial = emptyCache()
|
|
initial.complete = true
|
|
await saveCache(initial)
|
|
const corruptPath = join(cacheDir, 'session-refresh.lock')
|
|
await writeFile(corruptPath, '')
|
|
await utimes(corruptPath, new Date(1), new Date(1))
|
|
const source = join(root, 'changed.json')
|
|
await writeFile(source, JSON.stringify({ output: 404 }))
|
|
|
|
const a = worker(cacheDir, barriers, 'a', source)
|
|
const b = worker(cacheDir, barriers, 'b', source)
|
|
const winner = await Promise.race([
|
|
waitFor(join(barriers, 'a.parsed')).then(() => 'a'),
|
|
waitFor(join(barriers, 'b.parsed')).then(() => 'b'),
|
|
])
|
|
const loser = winner === 'a' ? 'b' : 'a'
|
|
expect(Number(existsSync(join(barriers, 'a.parsed'))) + Number(existsSync(join(barriers, 'b.parsed')))).toBe(1)
|
|
const loserOutcome = await waitForAny(barriers, [
|
|
`${loser}.timed-out`, `${loser}.parsed`, `${loser}.completed-by-other`, `${loser}.unavailable`,
|
|
])
|
|
// Exactly one owner publishes. The loser of a stale-lock contest is not
|
|
// the owner: `timed-out` is the usual wait-out, but missing+missing during
|
|
// the winner's unlink-guard/create-successor gap is honestly
|
|
// `completed-by-other` (#904). Do not require timed-out.
|
|
expect(loserOutcome, (await readdir(barriers)).join(',')).not.toBe(`${loser}.parsed`)
|
|
expect([`${loser}.timed-out`, `${loser}.completed-by-other`, `${loser}.unavailable`]).toContain(loserOutcome)
|
|
await writeFile(join(barriers, `${winner}.save`), '')
|
|
await Promise.all([waitForExit(a), waitForExit(b)])
|
|
await expect(stat(join(cacheDir, 'session-refresh.lock.takeover'))).rejects.toMatchObject({ code: 'ENOENT' })
|
|
expect(Object.keys((await loadCache()).providers['regression']?.files ?? {})).toEqual([source])
|
|
})
|
|
|
|
// #1117: the menubar/desktop watchdog kills a CLI child with SIGTERM first,
|
|
// precisely so the holder can unlink its own lock instead of leaving one for
|
|
// every later spawn to wait out.
|
|
it('unlinks its own lock when the holder is SIGTERMed mid-transaction', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'cb-refresh-sigterm-'))
|
|
roots.push(root)
|
|
const cacheDir = join(root, 'cache')
|
|
const barriers = join(root, 'barriers')
|
|
await mkdir(cacheDir, { recursive: true })
|
|
await mkdir(barriers, { recursive: true })
|
|
process.env['CODEBURN_CACHE_DIR'] = cacheDir
|
|
const initial = emptyCache()
|
|
initial.complete = true
|
|
await saveCache(initial)
|
|
const source = join(root, 'changed.json')
|
|
await writeFile(source, JSON.stringify({ output: 505 }))
|
|
|
|
const holder = worker(cacheDir, barriers, 'a', source)
|
|
// The worker blocks on its save barrier while holding the lock.
|
|
await waitFor(join(barriers, 'a.parsed'))
|
|
const lock = join(cacheDir, 'session-refresh.lock')
|
|
expect(existsSync(lock)).toBe(true)
|
|
|
|
const exited = new Promise<void>(resolve => { holder.once('exit', () => resolve()) })
|
|
holder.kill('SIGTERM')
|
|
await exited
|
|
expect(existsSync(lock)).toBe(false)
|
|
})
|
|
|
|
it('serializes disjoint parsed updates so the later publication cannot drop the first', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'cb-refresh-process-'))
|
|
roots.push(root)
|
|
const cacheDir = join(root, 'cache')
|
|
const barriers = join(root, 'barriers')
|
|
await mkdir(cacheDir, { recursive: true })
|
|
await mkdir(barriers, { recursive: true })
|
|
process.env['CODEBURN_CACHE_DIR'] = cacheDir
|
|
const initial = emptyCache()
|
|
initial.complete = true
|
|
await saveCache(initial)
|
|
|
|
const sourceA = join(root, 'changed-a.json')
|
|
const sourceB = join(root, 'changed-b.json')
|
|
await writeFile(sourceA, JSON.stringify({ output: 101 }))
|
|
await writeFile(sourceB, JSON.stringify({ output: 202 }))
|
|
|
|
const a = worker(cacheDir, barriers, 'a', sourceA)
|
|
const b = worker(cacheDir, barriers, 'b', sourceB)
|
|
|
|
// Exactly one child can cross the acquisition barrier. Let it publish and
|
|
// release; only then can the other reload the first child's update.
|
|
let retry: ChildProcess | undefined
|
|
await Promise.race([
|
|
waitFor(join(barriers, 'a.parsed')).then(() => 'a'),
|
|
waitFor(join(barriers, 'b.parsed')).then(() => 'b'),
|
|
]).then(async first => {
|
|
const second = first === 'a' ? 'b' : 'a'
|
|
const secondSource = second === 'a' ? sourceA : sourceB
|
|
await writeFile(join(barriers, `${first}.save`), '')
|
|
await waitFor(join(barriers, `${first}.published`))
|
|
// A waiter that observes the clean release correctly serves the holder's
|
|
// fresh snapshot instead of mutating. A subsequent refresh of that
|
|
// process then acquires normally and applies its independently visible
|
|
// source change on top of the holder's publication.
|
|
const outcome = await waitForAny(barriers, [`${second}.completed-by-other`, `${second}.parsed`])
|
|
if (outcome === `${second}.parsed`) {
|
|
await writeFile(join(barriers, `${second}.save`), '')
|
|
} else {
|
|
await waitForExit(second === 'a' ? a : b)
|
|
retry = worker(cacheDir, barriers, `${second}-retry`, secondSource)
|
|
await waitFor(join(barriers, `${second}-retry.parsed`))
|
|
await writeFile(join(barriers, `${second}-retry.save`), '')
|
|
}
|
|
})
|
|
|
|
await Promise.all([waitForExit(a), waitForExit(b), ...(retry ? [waitForExit(retry)] : [])])
|
|
const files = (await loadCache()).providers['regression']?.files ?? {}
|
|
expect(Object.keys(files).sort()).toEqual([sourceA, sourceB].sort())
|
|
})
|
|
|
|
it('proves the barrier reproducer loses one update when the transaction gate is bypassed', async () => {
|
|
const root = await mkdtemp(join(tmpdir(), 'cb-refresh-control-'))
|
|
roots.push(root)
|
|
const cacheDir = join(root, 'cache')
|
|
const barriers = join(root, 'barriers')
|
|
await mkdir(cacheDir, { recursive: true })
|
|
await mkdir(barriers, { recursive: true })
|
|
process.env['CODEBURN_CACHE_DIR'] = cacheDir
|
|
const initial = emptyCache()
|
|
initial.complete = true
|
|
await saveCache(initial)
|
|
|
|
const sourceA = join(root, 'changed-a.json')
|
|
const sourceB = join(root, 'changed-b.json')
|
|
await writeFile(sourceA, JSON.stringify({ output: 101 }))
|
|
await writeFile(sourceB, JSON.stringify({ output: 202 }))
|
|
const a = worker(cacheDir, barriers, 'a', sourceA, true)
|
|
const b = worker(cacheDir, barriers, 'b', sourceB, true)
|
|
await Promise.all([waitFor(join(barriers, 'a.parsed')), waitFor(join(barriers, 'b.parsed'))])
|
|
|
|
await writeFile(join(barriers, 'a.save'), '')
|
|
await waitFor(join(barriers, 'a.published'))
|
|
await writeFile(join(barriers, 'b.save'), '')
|
|
await Promise.all([waitForExit(a), waitForExit(b)])
|
|
|
|
const files = (await loadCache()).providers['regression']?.files ?? {}
|
|
expect(Object.keys(files)).toEqual([sourceB])
|
|
})
|
|
})
|