From ddba790de13a1ee10a9456236b5413dd457cc717 Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Mon, 9 Nov 2026 05:22:51 +0800 Subject: [PATCH] Fix the timeline trace lock so a window behind another writer persists A trace opened while another process held writer.lock stayed memory-only for the rest of its life: the lock was taken once at open, and flush() answered true although the appended spans never reached the disk. The store can now ask for the lock again when it has something to write, adopts the segments already on disk when it gets it, and answers false (naming the reason once) while it still cannot write. Tested with two real child processes on one trace directory: the window behind the writer takes the lock after the writer exits and its span lands on disk, while a blocked flush returns false and warns once. Fixes #2706 Docs-not-needed: the change stays inside the timeline store's existing single-writer design; no subsystem contract, owned path or public architecture changed. --- changelog.d/2706-trace-lock.fixed.md | 1 + plugins/timeline/src/store.js | 122 +++++++++--- plugins/timeline/src/timeline.js | 29 +-- .../timeline/test/trace-lock-retry.test.js | 186 ++++++++++++++++++ 4 files changed, 298 insertions(+), 40 deletions(-) create mode 100644 changelog.d/2706-trace-lock.fixed.md create mode 100644 plugins/timeline/test/trace-lock-retry.test.js diff --git a/changelog.d/2706-trace-lock.fixed.md b/changelog.d/2706-trace-lock.fixed.md new file mode 100644 index 000000000..d2ca1d75f --- /dev/null +++ b/changelog.d/2706-trace-lock.fixed.md @@ -0,0 +1 @@ +A timeline window that opens while another window is writing the trace no longer loses every span for the rest of its life: it asks for the writer lock again when it has something to write, begins persisting as soon as the other window closes, and keeps the segments already on disk. Until the lock is free a flush that wrote nothing answers `false` and names the reason once, instead of reporting success for a dropped span. See [#2706](https://github.com/ThinkFlowLab/FlyBot/issues/2706). diff --git a/plugins/timeline/src/store.js b/plugins/timeline/src/store.js index c7feaa494..b574ada41 100644 --- a/plugins/timeline/src/store.js +++ b/plugins/timeline/src/store.js @@ -82,7 +82,12 @@ function writeWhole(dir, path, value) { const segmentName = n => `trace-${String(n).padStart(6, '0')}.json` // Opens (and creates) the trace in `dir`. `persist: false` keeps everything in memory. -export function openTrace({ dir, persist = true, now = Date.now, warn = () => {}, ...limits } = {}) { +// +// `retryLock: true` answers a memory-only trace instead of throwing when another live process +// holds the writer lock; that trace asks for the lock again on every flush (and on its flush +// timer) and adopts everything already on disk when it gets it. Without the option a held lock +// still refuses the open, which is what a caller that wants one writer at a time relies on. +export function openTrace({ dir, persist = true, retryLock = false, now = Date.now, warn = () => {}, ...limits } = {}) { const { segmentRecords, segments, retentionMs, flushMs } = { ...DEFAULTS, ...limits } if (persist && typeof dir !== 'string') throw new Error('openTrace needs a directory') /** @type {{ n: number, startedAt: number, names: Record>, records: object[] }[]} */ @@ -92,6 +97,35 @@ export function openTrace({ dir, persist = true, now = Date.now, warn = () => {} let dirty = false let closed = false let failures = 0 + let persisting = persist + let deferred = false + let lockFailure + let lockWarned = false + + const release = () => { + if (!lock) return + held.delete(dir) + try { unlinkSync(lock) } catch { /* already gone */ } + lock = undefined + } + + // Reads the segments already in `dir` into `kept`. A segment that does not parse is set aside + // and never read again; a directory that cannot be listed throws to the caller. + const load = () => { + const files = readdirSync(dir).filter(name => SEGMENT.test(name)).sort() + for (const name of files) { + try { + const data = JSON.parse(readFileSync(join(dir, name), 'utf8')) + if (data?.version !== TRACE_VERSION || !Array.isArray(data.records)) throw new Error('not a trace segment') + const names = data.names && typeof data.names === 'object' ? data.names : {} + kept.push({ n: Number(SEGMENT.exec(name)[1]), startedAt: Number(data.startedAt) || 0, names, records: data.records, saved: data.records.length, savedNames: JSON.stringify(names) }) + } catch (error) { + // Never crash the host over the trace, and never read the same bad file every start. + warn('flybot-timeline: trace segment %s is unreadable (%s); it is set aside', name, error?.message ?? error) + try { renameSync(join(dir, name), join(dir, `${name}.corrupt`)) } catch { /* it is ignored either way */ } + } + } + } // Everything between taking the lock and answering the trace can throw: a directory that // cannot be listed, a segment read, a prune. A refused open has to give the lock back the way @@ -100,20 +134,16 @@ export function openTrace({ dir, persist = true, now = Date.now, warn = () => {} try { if (persist) { mkdirSync(dir, { recursive: true, mode: 0o700 }) - lock = acquire(dir) - const files = readdirSync(dir).filter(name => SEGMENT.test(name)).sort() - for (const name of files) { - try { - const data = JSON.parse(readFileSync(join(dir, name), 'utf8')) - if (data?.version !== TRACE_VERSION || !Array.isArray(data.records)) throw new Error('not a trace segment') - const names = data.names && typeof data.names === 'object' ? data.names : {} - kept.push({ n: Number(SEGMENT.exec(name)[1]), startedAt: Number(data.startedAt) || 0, names, records: data.records, saved: data.records.length, savedNames: JSON.stringify(names) }) - } catch (error) { - // Never crash the host over the trace, and never read the same bad file every start. - warn('flybot-timeline: trace segment %s is unreadable (%s); it is set aside', name, error?.message ?? error) - try { renameSync(join(dir, name), join(dir, `${name}.corrupt`)) } catch { /* it is ignored either way */ } - } + try { + lock = acquire(dir) + } catch (error) { + if (!(retryLock && error instanceof TraceLocked)) throw error + // Another writer is live: keep in memory and ask again when there is something to write. + lockFailure = error + persisting = false + deferred = true } + if (!deferred) load() } const fresh = () => ({ n: (kept.at(-1)?.n ?? 0) + 1, startedAt: now(), names: {}, records: [] }) const newest = () => { @@ -122,10 +152,54 @@ export function openTrace({ dir, persist = true, now = Date.now, warn = () => {} } newest() + // Takes the lock that was held at open, and continues after the newest segment already on + // disk: the records collected while deferred are newer than every one of those, so they go + // into a segment of their own and the older segments are read, never overwritten. + const adopt = () => { + if (!deferred) return true + try { + lock = acquire(dir) + } catch (error) { + lockFailure = error + return false + } + const pending = kept + try { + kept = [] + load() + if (pending.some(segment => segment.records.length > 0)) kept.push(fresh()) + const target = newest() + for (const segment of pending) { + target.records.push(...segment.records) + for (const [group, table] of Object.entries(segment.names)) { + target.names[group] ??= {} + Object.assign(target.names[group], table) + } + } + persisting = true + prune() + } catch (error) { + // A directory that can be written but not read: give the lock back and stay in memory. + release() + kept = pending + persisting = false + lockFailure = error + return false + } + deferred = false + return true + } const flush = () => { clearTimeout(timer) timer = undefined - if (!persist || !dirty) return true + if (deferred && dirty && !adopt()) { + if (!lockWarned) { + lockWarned = true + warn('flybot-timeline: the trace in %s is not writable yet (%s); keeping %d record(s) in memory and asking for the writer lock again', dir, lockFailure?.message ?? lockFailure, kept.reduce((sum, segment) => sum + segment.records.length, 0)) + } + return false + } + if (!persisting || !dirty) return true try { for (const segment of kept) { if (segment.saved === segment.records.length && segment.savedNames === JSON.stringify(segment.names)) continue @@ -143,7 +217,7 @@ export function openTrace({ dir, persist = true, now = Date.now, warn = () => {} } } const schedule = () => { - if (!persist || timer !== undefined || closed) return + if (timer !== undefined || closed || (!persisting && !deferred)) return timer = setTimeout(flush, flushMs) timer.unref?.() } @@ -154,9 +228,9 @@ export function openTrace({ dir, persist = true, now = Date.now, warn = () => {} const expired = segment => segment.records.length > 0 && segment.records.every(record => record.t < floor) while (kept.length > 1 && (kept.length > segments || expired(kept[0]))) { const [old] = kept.splice(0, 1) - if (persist) { try { unlinkSync(join(dir, segmentName(old.n))) } catch { /* already gone */ } } + if (persisting) { try { unlinkSync(join(dir, segmentName(old.n))) } catch { /* already gone */ } } } - if (!persist) return + if (!persisting) return for (const name of readdirSync(dir)) { const match = /^trace-(\d{6})\.json\.corrupt$/.exec(name) if (match && Number(match[1]) < kept[0].n) { try { unlinkSync(join(dir, name)) } catch { /* already gone */ } } @@ -199,25 +273,19 @@ export function openTrace({ dir, persist = true, now = Date.now, warn = () => {} return out }, flush, - stats: () => ({ segments: kept.length, records: kept.reduce((sum, segment) => sum + segment.records.length, 0), segmentRecords, maxSegments: segments, retentionMs, persist, failing: failures > 0 }), + stats: () => ({ segments: kept.length, records: kept.reduce((sum, segment) => sum + segment.records.length, 0), segmentRecords, maxSegments: segments, retentionMs, persist: persisting, failing: failures > 0, ...(deferred ? { deferred: true } : {}) }), close() { if (closed) return flush() closed = true clearTimeout(timer) - if (lock) { - held.delete(dir) - try { unlinkSync(lock) } catch { /* already gone */ } - } + release() }, } } catch (error) { // A refused open leaves nothing behind: the directory leaves `held` and the lock file goes, // exactly as `close()` would have done it. - if (lock) { - held.delete(dir) - try { unlinkSync(lock) } catch { /* already gone */ } - } + release() throw error } } diff --git a/plugins/timeline/src/timeline.js b/plugins/timeline/src/timeline.js index 941522e9f..d44b38479 100644 --- a/plugins/timeline/src/timeline.js +++ b/plugins/timeline/src/timeline.js @@ -15,14 +15,14 @@ export const DIGEST_ENDPOINT = 'timeline/digest' export function createTimeline(adapter, { dir, persist = true, now = Date.now, exporter, ...limits } = {}) { let trace - let persisting = persist try { - trace = openTrace({ dir, persist, now, warn: adapter.warn, ...limits }) + // A writer that is live now is asked for the lock again when there is something to write: + // two windows on one home is normal, and the second one's spans are the ones that explain + // what the first did. Only a directory that cannot be opened at all stays memory-only. + trace = openTrace({ dir, persist, retryLock: true, now, warn: adapter.warn, ...limits }) } catch (error) { - // Another live writer, or an unwritable directory: the timeline still works from memory. adapter.warn('flybot-timeline: the trace in %s is not writable (%s); recording in memory only', dir, error instanceof TraceLocked ? error.message : error?.message ?? error) trace = openTrace({ persist: false, now, warn: adapter.warn, ...limits }) - persisting = false } const recorder = createRecorder({ adapter, trace, now }) const off = adapter.subscribe(recorder.handlers) @@ -40,15 +40,18 @@ export function createTimeline(adapter, { dir, persist = true, now = Date.now, e } // The panel polls this every 2 s. Reads only; a host that another dsh holds still answers. - const window = (body = {}) => buildWindow(trace.records(), { - now: now(), - to: Number.isFinite(body?.to) ? body.to : undefined, - windowMs: clampWindow(body?.windowMs), - bots: adapter.bots(), - rooms: adapter.rooms(), - names: trace.names(), - stats: { ...trace.stats(), persisting }, - }) + const window = (body = {}) => { + const stats = trace.stats() + return buildWindow(trace.records(), { + now: now(), + to: Number.isFinite(body?.to) ? body.to : undefined, + windowMs: clampWindow(body?.windowMs), + bots: adapter.bots(), + rooms: adapter.rooms(), + names: trace.names(), + stats: { ...stats, persisting: stats.persist }, + }) + } const offEndpoint = adapter.handle(ENDPOINT, window) // On-demand OTLP export (#357): the same mapping the timer exporter uses, built from the trace // as it stands. Ids only unless the caller asks for names (#483), so a one-off export never diff --git a/plugins/timeline/test/trace-lock-retry.test.js b/plugins/timeline/test/trace-lock-retry.test.js new file mode 100644 index 000000000..bdf5ed2c9 --- /dev/null +++ b/plugins/timeline/test/trace-lock-retry.test.js @@ -0,0 +1,186 @@ +// #2706: a window that opens while another process writes the trace starts in memory. +// The lock is asked for again when there is something to write, so the window begins +// persisting as soon as the other writer leaves; until then a flush that cannot write +// answers `false` and names the reason once instead of reporting success for a lost span. +// +// Real modules and real child processes, one temporary trace directory per test. +import assert from 'node:assert/strict' +import { spawn } from 'node:child_process' +import { mkdtempSync, readdirSync, readFileSync, rmSync } from 'node:fs' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { test } from 'node:test' + +const TIMELINE = new URL('../src/timeline.js', import.meta.url).href +const NOW = 1_700_000_000_000 +const J = JSON.stringify + +const home = () => mkdtempSync(join(tmpdir(), 'flybot-timeline-lock-')) +const traceText = dir => readdirSync(dir) + .filter(name => /^trace-\d+\.json$/.test(name)) + .map(name => readFileSync(join(dir, name), 'utf8')) + .join('') +const warns = text => text.split('\n').filter(line => /not writable/i.test(line)).length +const payload = line => JSON.parse(line.slice(line.indexOf(' ') + 1)) + +const ADAPTER = `const adapter = { + warn: (...args) => process.stderr.write('warn: ' + args.map(String).join(' ') + '\\n'), + subscribe: () => () => {}, bots: () => [], rooms: () => [], handle: () => () => {}, readOnly: () => false, +}` + +const live = [] +// A real node child running the module graph under test; argv[1] is its trace directory. +// `waitFor` answers each marker once, from the output produced after the previous answer. +function start(code, dir) { + const proc = spawn(process.execPath, ['--input-type=module', '-e', code, dir], { stdio: ['pipe', 'pipe', 'pipe'] }) + live.push(proc) + let out = '' + let err = '' + let cursor = 0 + const waiters = [] + const take = (waiter) => { + const match = out.slice(cursor).match(waiter.re) + if (!match) return false + cursor += match.index + match[0].length + waiter.resolve(match[0]) + return true + } + const sweep = () => { + for (const waiter of waiters.slice()) { + if (!take(waiter)) continue + clearTimeout(waiter.timer) + waiters.splice(waiters.indexOf(waiter), 1) + } + } + proc.stdout.setEncoding('utf8') + proc.stderr.setEncoding('utf8') + proc.stdout.on('data', (chunk) => { out += chunk; sweep() }) + proc.stderr.on('data', (chunk) => { err += chunk }) + const exited = new Promise(resolve => proc.on('exit', (exitCode, signal) => resolve({ exitCode, signal }))) + return { + proc, + get output() { return out }, + get errors() { return err }, + exited, + waitFor(re, timeoutMs = 15_000) { + const found = out.slice(cursor).match(re) + if (found) { cursor += found.index + found[0].length; return Promise.resolve(found[0]) } + return new Promise((resolve, reject) => { + const waiter = { re, resolve, timer: undefined } + waiter.timer = setTimeout(() => reject(new Error(`timeout waiting for ${re}\nstdout: ${out}\nstderr: ${err}`)), timeoutMs) + waiters.push(waiter) + }) + }, + send(line) { proc.stdin.write(`${line}\n`) }, + } +} + +const writerCode = ` + import { createTimeline } from ${J(TIMELINE)} + ${ADAPTER} + const tl = createTimeline(adapter, { dir: process.argv[1], now: () => ${NOW}, flushMs: 60_000 }) + tl.trace.append({ t: ${NOW}, k: 'ts', id: 'first-writer' }) + process.stdout.write('@@flushed ' + JSON.stringify({ ok: tl.trace.flush(), stats: tl.trace.stats() }) + '\\n') + process.stdin.once('data', () => { tl.dispose(); process.exit(0) }) +` + +// The window behind the writer appends on demand and reports what flush() answered. +const behindCode = ` + import { createTimeline } from ${J(TIMELINE)} + ${ADAPTER} + const tl = createTimeline(adapter, { dir: process.argv[1], now: () => ${NOW}, flushMs: 60_000 }) + process.stdout.write('@@open ' + JSON.stringify(tl.trace.stats()) + '\\n') + process.stdin.setEncoding('utf8') + process.stdin.on('data', (chunk) => { + for (const line of chunk.split('\\n')) { + if (line.startsWith('append:')) { + tl.trace.append({ t: ${NOW}, k: 'ts', id: line.slice(7) }) + const ok = tl.trace.flush() + process.stdout.write('@@tried ' + JSON.stringify({ ok, stats: tl.trace.stats() }) + '\\n') + } + if (line === 'exit') { tl.dispose(); process.exit(0) } + } + }) +` + +test('a window that started behind another writer persists once that writer leaves', { timeout: 30_000 }, async (t) => { + const dir = home() + t.after(() => rmSync(dir, { recursive: true, force: true })) + + const first = start(writerCode, dir) + await first.waitFor(/@@flushed/) + assert.ok(traceText(dir).includes('first-writer'), 'the first writer is on disk') + + const behind = start(behindCode, dir) + const opened = payload(await behind.waitFor(/@@open (.*)/)) + assert.equal(opened.persist, false, 'the second window starts without the lock') + + first.send('exit') + await first.exited + + behind.send('append:after-the-writer-left') + const tried = payload(await behind.waitFor(/@@tried (.*)/)) + assert.equal(tried.ok, true, 'flush succeeds once the lock is free') + assert.equal(tried.stats.persist, true, 'the window is persisting now') + assert.ok(traceText(dir).includes('after-the-writer-left'), 'the span written after the lock was taken reaches the disk') + assert.ok(traceText(dir).includes('first-writer'), 'the segments already on disk are kept, not overwritten') + + behind.send('exit') + await behind.exited +}) + +test('a flush that still cannot take the lock answers false and names it once', { timeout: 30_000 }, async (t) => { + const dir = home() + t.after(() => rmSync(dir, { recursive: true, force: true })) + + const first = start(writerCode, dir) + await first.waitFor(/@@flushed/) + const behind = start(behindCode, dir) + await behind.waitFor(/@@open/) + + behind.send('append:blocked-one') + const blocked = payload(await behind.waitFor(/@@tried (.*)/)) + assert.equal(blocked.ok, false, 'a span that was not written is not a successful flush') + assert.equal(blocked.stats.persist, false) + assert.ok(blocked.stats.records >= 1, 'the record is kept in memory') + assert.equal(warns(behind.errors), 1, 'the reason is said once') + + behind.send('append:blocked-two') + const second = payload(await behind.waitFor(/@@tried (.*)/)) + await new Promise(resolve => setTimeout(resolve, 100)) + assert.equal(second.ok, false) + assert.equal(warns(behind.errors), 1, 'it is not repeated on the next failed flush') + + first.send('exit') + await first.exited + // Once the writer is gone the same window takes the lock and the held spans land on disk. + behind.send('append:after-the-writer-left') + const recovered = payload(await behind.waitFor(/@@tried (.*)/)) + assert.equal(recovered.ok, true, `stderr: ${behind.errors}`) + assert.equal(recovered.stats.persist, true) + const text = traceText(dir) + for (const id of ['blocked-one', 'blocked-two', 'after-the-writer-left']) assert.ok(text.includes(id), `${id} reaches the disk`) + + behind.send('exit') + await behind.exited +}) + +test('a window opened after the writer leaves takes the lock at once', { timeout: 30_000 }, async (t) => { + const dir = home() + t.after(() => rmSync(dir, { recursive: true, force: true })) + const first = start(writerCode, dir) + await first.waitFor(/@@flushed/) + first.send('exit') + await first.exited + + const fresh = start(writerCode, dir) + const flushed = payload(await fresh.waitFor(/@@flushed (.*)/)) + assert.equal(flushed.stats.persist, true) + assert.ok(traceText(dir).includes('first-writer')) + fresh.send('exit') + await fresh.exited +}) + +test.after(() => { + for (const proc of live) if (proc.exitCode === null && proc.signalCode === null) proc.kill('SIGKILL') +})