diff --git a/src/daemon.ts b/src/daemon.ts index a19de0c..26d97e0 100644 --- a/src/daemon.ts +++ b/src/daemon.ts @@ -458,35 +458,11 @@ export class GlobalDaemon { this.log('INFO', 'No weave_project / API key configured — tracing disabled'); } - // Herd prevention: probe the socket before removing it. Concurrent hook - // invocations can race the probe→spawn window and each start a daemon. If - // a live listener already owns the socket, yield — otherwise the late spawn - // would unlink the winner's socket and split the in-memory teamMembers map - // across processes, breaking cross-session nesting. - if (fs.existsSync(this.socketPath)) { - const alive = await new Promise((resolve) => { - const probe = net.createConnection(this.socketPath); - probe.once('connect', () => { probe.destroy(); resolve(true); }); - probe.once('error', () => resolve(false)); - }); - if (alive) { - this.log('INFO', 'Another daemon already owns the socket — exiting to avoid a herd'); - process.exit(0); - } - fs.unlinkSync(this.socketPath); - } - - // Restrict socket to owner-only access - const prevUmask = process.umask(0o077); - // allowHalfOpen lets handleConnection write a reply after the client - // half-closes (the `config-hash` query). Every branch closes the socket - // explicitly so the high-frequency hook-event path still ends promptly. - this.server = net.createServer({ allowHalfOpen: true }, (socket) => this.handleConnection(socket)); - await new Promise((resolve, reject) => { - this.server!.listen(this.socketPath, resolve); - this.server!.once('error', reject); - }); - process.umask(prevUmask); + // Bind the socket, exiting cleanly if another daemon already owns it. + // Concurrent hook invocations can each cold-start a daemon, but only one + // can bind; the losers exit (process.exit(0)) and their hook still reaches + // the winner over the socket. See bindSocketWithHerdProtection. + await this.bindSocketWithHerdProtection(); this.running = true; this.log('INFO', `Daemon started — socket: ${this.socketPath}`); @@ -510,6 +486,65 @@ export class GlobalDaemon { setInterval(() => this.checkInactivity(), checkEveryMs).unref(); } + /** Probe whether a live daemon is accepting connections on the socket. Uses a + * real connect() attempt — the inode existing is not proof of a listener + * (an ungraceful exit leaves a stale inode behind). */ + private socketHasLiveListener(): Promise { + if (!fs.existsSync(this.socketPath)) return Promise.resolve(false); + return new Promise((resolve) => { + const probe = net.createConnection(this.socketPath); + probe.once('connect', () => { probe.destroy(); resolve(true); }); + probe.once('error', () => { probe.destroy(); resolve(false); }); + }); + } + + /** Create a fresh server and listen once, resolving on success and rejecting + * on the first listen error. A new server per attempt — one that errored on + * listen cannot be reused. Socket is owner-only (umask 0o077). */ + private listenOnce(): Promise { + return new Promise((resolve, reject) => { + const prevUmask = process.umask(0o077); + // allowHalfOpen lets handleConnection write a reply after the client + // half-closes (the `config-hash` query). Every branch closes the socket + // explicitly so the high-frequency hook-event path still ends promptly. + const server = net.createServer({ allowHalfOpen: true }, (socket) => this.handleConnection(socket)); + const onError = (err: Error) => { process.umask(prevUmask); reject(err); }; + server.once('error', onError); + server.listen(this.socketPath, () => { + process.umask(prevUmask); + server.removeListener('error', onError); + this.server = server; + resolve(); + }); + }); + } + + /** + * Bind the daemon socket, tolerant of a herd of concurrent starts. Listen; on + * EADDRINUSE/EEXIST, re-probe: a live listener means another daemon won → exit + * 0; a stale inode is unlinked and retried. Only a confirmed-stale socket is + * ever unlinked, so a late starter can't delete the winner's live socket. + */ + private async bindSocketWithHerdProtection(): Promise { + const MAX_RECLAIM_ATTEMPTS = 5; + for (let attempt = 0; ; attempt++) { + try { + await this.listenOnce(); + return; + } catch (err) { + const code = (err as NodeJS.ErrnoException).code; + if (code !== 'EADDRINUSE' && code !== 'EEXIST') throw err; + if (await this.socketHasLiveListener()) { + this.log('INFO', 'Another daemon already owns the socket — exiting to avoid a herd'); + process.exit(0); + } + // Stale inode from an ungraceful exit — reclaim it and retry. + if (attempt >= MAX_RECLAIM_ATTEMPTS) throw err; + try { fs.unlinkSync(this.socketPath); } catch { /* already cleaned; retry */ } + } + } + } + // ── tracer initialization ─────────────────────────────────────────────── private initTracer(): void { diff --git a/tests/daemon-startup-race.test.ts b/tests/daemon-startup-race.test.ts new file mode 100644 index 0000000..5d415fd --- /dev/null +++ b/tests/daemon-startup-race.test.ts @@ -0,0 +1,69 @@ +// SPDX-FileCopyrightText: 2026 CoreWeave, Inc. +// SPDX-License-Identifier: MIT +// SPDX-PackageName: weave-claude-code + +// Herd safety. When several hooks fire at once and each cold-starts a daemon, +// only one can bind the socket; the losers must yield cleanly. The old start() +// guarded with existsSync -> probe -> unlink, then listen() and threw on error. +// Two daemons that both found no socket raced listen(): the loser crashed with +// EEXIST/EADDRINUSE ("Daemon failed to start", exit 1). Seven such crashes +// appeared in one local log over 14 days. +// +// The race is inherent, so a single run is probabilistic (measured on the old +// code: ~2 of 3 herds crash at least one daemon, the rest happen to serialize). +// The assertion here is therefore the POST-FIX invariant, which is +// deterministic once listen() errors are handled by re-probing instead of +// throwing: a herd crashes nobody and leaves exactly one listener. + +import { test } from 'node:test'; +import assert from 'node:assert/strict'; +import { spawn } from 'node:child_process'; +import * as fs from 'node:fs'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import { fileURLToPath } from 'node:url'; + +import { probeUnixSocket } from '../src/utils.ts'; +import { waitUntil } from './helpers.ts'; + +const HERE = path.dirname(fileURLToPath(import.meta.url)); +const REPO_ROOT = path.resolve(HERE, '..'); +const CLI = path.join(REPO_ROOT, 'src', 'cli.ts'); + +test('a herd of concurrent daemon starts crashes nobody and leaves exactly one listener', async () => { + const home = fs.mkdtempSync(path.join(os.homedir(), '.weave-herd-')); + const configDir = path.join(home, '.weave-claude-code'); + const socketPath = path.join(configDir, 'daemon.sock'); + const logPath = path.join(configDir, 'logs', 'daemon.log'); + fs.mkdirSync(path.join(configDir, 'logs'), { recursive: true }); + fs.writeFileSync( + path.join(configDir, 'settings.json'), + JSON.stringify({ + weave_project: 'test/test', wandb_api_key: 'fake-key', + daemon_socket: socketPath, log_file: logPath, debug: true, + }), + ); + + const N = 12; + const procs = Array.from({ length: N }, () => + spawn(process.execPath, ['--import', 'tsx', CLI, 'daemon'], { + env: { ...process.env, HOME: home, WANDB_BASE_URL: 'http://127.0.0.1:1' }, + stdio: 'ignore', + }), + ); + try { + await waitUntil(() => fs.existsSync(socketPath), 5000); + await new Promise((r) => setTimeout(r, 2000)); // let every daemon resolve bind/yield + + const log = fs.existsSync(logPath) ? fs.readFileSync(logPath, 'utf8') : ''; + const failures = (log.match(/Daemon failed to start/g) ?? []).length; + const started = (log.match(/Daemon started/g) ?? []).length; + + assert.equal(failures, 0, `herd must not crash any daemon; log:\n${log}`); + assert.equal(started, 1, `exactly one daemon should bind, got ${started}; log:\n${log}`); + assert.equal(await probeUnixSocket(socketPath), 'alive', 'a live listener should own the socket'); + } finally { + for (const p of procs) { try { p.kill('SIGKILL'); } catch { /* already gone */ } } + fs.rmSync(home, { recursive: true, force: true }); + } +});