fix(state-store): retry Windows contention and repair fixtures

This commit is contained in:
Yann Roberto
2026-10-01 00:25:22 +08:00
parent ed4b856bfe
commit 94cd7db7c3
4 changed files with 92 additions and 8 deletions
+4 -1
View File
@@ -73,7 +73,10 @@ function acquireLock(dbPath, timeoutMs) {
try {
descriptor = fs.openSync(lockPath, 'wx', 0o600);
} catch (error) {
if (!hasCode(error, 'EEXIST')) throw error;
// Windows may report EPERM while the previous owner's unlinked file is
// pending deletion. Retry exclusive creation within the same deadline.
const pendingDelete = process.platform === 'win32' && hasCode(error, 'EPERM');
if (!hasCode(error, 'EEXIST') && !pendingDelete) throw error;
const remaining = deadline - performance.now();
if (remaining <= 0) {
const busy = new Error(`State store is busy: ${dbPath}. Retry after the other ECC operation finishes. `
+18 -6
View File
@@ -30,11 +30,12 @@ function startOwnedChild(script, args = [], dependencies = {}) {
let child;
try {
fs.mkdirSync(home); fs.mkdirSync(tmp);
const dependencyRoot = path.resolve(path.dirname(require.resolve('sql.js/package.json')), '..');
// Absolute fixture paths resolve dependencies from the checkout, even with
// an isolated cwd. No package-private metadata or NODE_PATH is needed.
child = forkChild(script, args, {
execPath: process.execPath, execArgv: [], shell: false, detached: platform !== 'win32',
cwd: root, stdio: ['ignore', 'pipe', 'pipe', 'ipc'],
env: { PATH: path.dirname(process.execPath), NODE_PATH: dependencyRoot,
env: { PATH: path.dirname(process.execPath),
HOME: home, USERPROFILE: home, APPDATA: home, LOCALAPPDATA: home,
TMPDIR: tmp, TMP: tmp, TEMP: tmp, NODE_USE_ENV_PROXY: '0' },
});
@@ -47,6 +48,8 @@ function startOwnedChild(script, args = [], dependencies = {}) {
const closeWaiters = new Set();
let ownedPid = null;
let closed = false;
let ipcClosed = false;
const closedStreams = new Set();
let reaped = false;
let exitCode = null;
let exitSignal = null;
@@ -59,17 +62,25 @@ function startOwnedChild(script, args = [], dependencies = {}) {
if (!failed) { failed = true; failure = error; }
for (const waiter of [...waiters]) waiter.reject(error);
};
const markClosed = () => {
closed = true;
for (const done of [...closeWaiters]) done(true);
};
// Some Node versions omit the aggregate close event after a parent-initiated
// IPC disconnect. Require public confirmation of every owned resource instead.
const checkClosed = () => {
if (reaped && ipcClosed && closedStreams.size === 2) markClosed();
};
child.once('disconnect', () => { ipcClosed = true; checkClosed(); });
child.once('spawn', () => { if (Number.isInteger(child.pid) && child.pid > 0) ownedPid = child.pid; });
child.on('error', fail);
child.once('exit', (code, signal) => {
reaped = true; exitCode = code; exitSignal = signal;
const error = new Error(`Fixture exited (${code}, ${signal || 'no signal'}): ${stderr}`);
for (const waiter of [...waiters]) waiter.reject(error);
checkClosed();
});
child.once('close', () => {
closed = true;
for (const done of [...closeWaiters]) done(true);
});
child.once('close', markClosed);
child.on('message', message => {
for (const observer of [...observers]) {
try { observer(message); } catch (error) { fail(error); }
@@ -77,6 +88,7 @@ function startOwnedChild(script, args = [], dependencies = {}) {
for (const waiter of [...waiters]) waiter.check(message);
});
for (const stream of [child.stdout, child.stderr]) {
stream.once('close', () => { closedStreams.add(stream); checkClosed(); });
stream.on('data', data => {
captured += data.length;
if (captured > 128 * 1024) { fail(new Error('Fixture output exceeded 128 KiB')); return; }
+32
View File
@@ -4,6 +4,7 @@ const assert = require('assert');
const fs = require('fs');
const os = require('os');
const path = require('path');
const { EventEmitter } = require('events');
const { startOwnedChild, withOwnedChildren, withCleanup } = require('./helpers/state-store-worker');
const { createStateStore } = require('../../scripts/lib/state-store');
@@ -61,6 +62,37 @@ async function run() {
console.log('\n=== Testing state-store concurrency ===\n');
await test('owned fixtures wait for exit, both pipes, and IPC when aggregate close is absent', async () => {
const events = ['exit', 'stdout', 'stderr', 'disconnect'];
for (const last of events) {
const child = new EventEmitter();
child.stdout = new EventEmitter();
child.stderr = new EventEmitter();
child.stdout.destroy = () => child.stdout.emit('close');
child.stderr.destroy = () => child.stderr.emit('close');
child.connected = true;
child.pid = 12345;
child.disconnect = () => { child.connected = false; };
const signals = [];
const client = startOwnedChild(WORKER, [], {
fork: () => child, platform: 'linux', kill: (...args) => signals.push(args)
});
child.emit('spawn');
const stopping = client.stop();
const emit = event => {
if (event === 'exit') child.emit('exit', 0, null);
else if (event === 'disconnect') child.emit('disconnect');
else child[event].emit('close');
};
for (const event of events.filter(event => event !== last)) emit(event);
assert.strictEqual(client.diagnostics.closed, false, `Must wait for ${last}`);
emit(last);
await stopping;
assert.strictEqual(client.diagnostics.closed, true);
assert.deepStrictEqual(signals, [], 'Fully closed fixtures need no forced termination');
}
});
await test('closing an old reader preserves a task saved by the real CLI', async dbPath => {
const reader = await createStateStore({ dbPath });
await withCleanup(async () => {
+38 -1
View File
@@ -48,10 +48,47 @@ function lockFixture(overrides = {}, platform = 'linux') {
}
const lock = loadPrivate(LOCK, { fs: facade, os: { hostname: () => 'fixture' } }, undefined,
{ platform, pid: 7 });
return { ...state, calls, run: callback => lock.withStateStoreLock('synthetic.db', callback, { timeoutMs: 0 }) };
return { ...state, calls, run: (callback, timeoutMs = 0) => lock.withStateStoreLock('synthetic.db', callback, { timeoutMs }) };
}
async function runSynthetic(test) {
await test('Windows pending-delete EPERM retries until exclusive creation succeeds', () => {
const denied = Object.assign(new Error('pending deletion'), { code: 'EPERM' });
let attempts = 0;
let callbacks = 0;
const fixture = lockFixture({ openSync(state, open, file, flags, mode) {
assert.strictEqual(file, 'synthetic.db.ecc-state.lock');
assert.strictEqual(flags, 'wx');
assert.strictEqual(mode, 0o600);
attempts += 1;
if (attempts === 1) { state.calls.push('open'); throw denied; }
return open();
} }, 'win32');
assert.strictEqual(fixture.run(() => { callbacks += 1; return 'done'; }, 1000), 'done');
assert.strictEqual(attempts, 2);
assert.strictEqual(callbacks, 1);
assert.deepStrictEqual(fixture.calls, ['open', 'open', 'fstat', 'write', 'lstat', 'unlink', 'close']);
});
await test('sustained Windows open EPERM times out as busy without touching the lock', () => {
const denied = Object.assign(new Error('pending deletion'), { code: 'EPERM' });
const fixture = lockFixture({ openSync(state) { state.calls.push('open'); throw denied; } }, 'win32');
const result = thrownBy(() => fixture.run(() => assert.fail('callback must not run'), 20));
assert.ok(result.failed);
assert.strictEqual(result.error.code, 'STATE_STORE_BUSY');
assert.match(result.error.message, /synthetic\.db\.ecc-state\.lock/);
assert.ok(fixture.calls.length > 0);
assert.ok(fixture.calls.every(call => call === 'open'));
});
for (const [platform, code] of [['linux', 'EPERM'], ['darwin', 'EPERM'], ['win32', 'EACCES'], ['win32', 'EIO']]) {
await test(`unrelated lock-open errors propagate immediately (${platform}, ${code})`, () => {
const denied = Object.assign(new Error('open denied'), { code });
const fixture = lockFixture({ openSync(state) { state.calls.push('open'); throw denied; } }, platform);
const result = thrownBy(() => fixture.run(() => assert.fail('callback must not run'), 1000));
assert.ok(result.failed);
assert.strictEqual(result.error, denied);
assert.deepStrictEqual(fixture.calls, ['open']);
});
}
await test('owned lock unlinks before its only close', () => {
const fixture = lockFixture();
assert.strictEqual(fixture.run(() => 'done'), 'done');