From ec7e28beded03527a73cd5202ac3a510e006abb0 Mon Sep 17 00:00:00 2001 From: Danyang Chen Date: Sun, 8 Nov 2026 02:33:22 +0800 Subject: [PATCH] Release the timeline writer lock when a trace open refuses openTrace took writer.lock and then listed the directory, read the segments and pruned, all unguarded. A throw on that path left close(), the only release, unreached: the directory stayed in the in-process `held` set and writer.lock stayed on disk for every later process, so one transient read failure fell back to the memory trace for the life of the host (timeline.js, no retry). Everything after acquire(dir) now runs in one try whose catch drops the directory from `held` and unlinks the lock, exactly as close() does, then rethrows. A successful open still holds the lock until close(). Fixes #2629 Docs-not-needed: #2629 only changes when the writer lock is released; the single-writer discipline, lock takeover and lock content documented in docs/subsystems/timeline.md are unchanged. --- changelog.d/2629.fixed.md | 1 + plugins/timeline/src/store.js | 220 +++++++++++++++------------- plugins/timeline/test/store.test.js | 29 +++- 3 files changed, 146 insertions(+), 104 deletions(-) create mode 100644 changelog.d/2629.fixed.md diff --git a/changelog.d/2629.fixed.md b/changelog.d/2629.fixed.md new file mode 100644 index 000000000..b82640ea2 --- /dev/null +++ b/changelog.d/2629.fixed.md @@ -0,0 +1 @@ +A timeline trace whose open is refused — a directory that cannot be listed, one removed underneath, or an I/O error — now gives its writer lock back instead of keeping it for the life of the process, so the next open in the same process and every later process succeeds once the problem is gone, and one transient read failure no longer leaves the on-disk trace switched off until a restart ([#2629](https://github.com/ThinkFlowLab/FlyBot/issues/2629)). diff --git a/plugins/timeline/src/store.js b/plugins/timeline/src/store.js index 275cd278d..c7feaa494 100644 --- a/plugins/timeline/src/store.js +++ b/plugins/timeline/src/store.js @@ -93,118 +93,132 @@ export function openTrace({ dir, persist = true, now = Date.now, warn = () => {} let closed = false let failures = 0 - 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 */ } + // 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 + // `close()` does, or this process holds the directory for good and `writer.lock` stays on disk + // for every later process. + 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 */ } + } } } - } - const fresh = () => ({ n: (kept.at(-1)?.n ?? 0) + 1, startedAt: now(), names: {}, records: [] }) - const newest = () => { - if (kept.length === 0) kept.push(fresh()) - return kept.at(-1) - } - newest() + const fresh = () => ({ n: (kept.at(-1)?.n ?? 0) + 1, startedAt: now(), names: {}, records: [] }) + const newest = () => { + if (kept.length === 0) kept.push(fresh()) + return kept.at(-1) + } + newest() - const flush = () => { - clearTimeout(timer) - timer = undefined - if (!persist || !dirty) return true - try { - for (const segment of kept) { - if (segment.saved === segment.records.length && segment.savedNames === JSON.stringify(segment.names)) continue - writeWhole(dir, join(dir, segmentName(segment.n)), { version: TRACE_VERSION, segment: segment.n, startedAt: segment.startedAt, names: segment.names, records: segment.records }) - segment.saved = segment.records.length - segment.savedNames = JSON.stringify(segment.names) + const flush = () => { + clearTimeout(timer) + timer = undefined + if (!persist || !dirty) return true + try { + for (const segment of kept) { + if (segment.saved === segment.records.length && segment.savedNames === JSON.stringify(segment.names)) continue + writeWhole(dir, join(dir, segmentName(segment.n)), { version: TRACE_VERSION, segment: segment.n, startedAt: segment.startedAt, names: segment.names, records: segment.records }) + segment.saved = segment.records.length + segment.savedNames = JSON.stringify(segment.names) + } + dirty = false + failures = 0 + return true + } catch (error) { + // Disk full, directory removed: record in memory and say so once per streak. + if (failures++ === 0) warn('flybot-timeline: trace write failed: %s', error?.message ?? error) + return false } - dirty = false - failures = 0 - return true - } catch (error) { - // Disk full, directory removed: record in memory and say so once per streak. - if (failures++ === 0) warn('flybot-timeline: trace write failed: %s', error?.message ?? error) - return false } - } - const schedule = () => { - if (!persist || timer !== undefined || closed) return - timer = setTimeout(flush, flushMs) - timer.unref?.() - } - // Files beyond the newest `segments`, segments that hold only expired records, and - // corrupt leftovers older than every kept segment go. - const prune = () => { - const floor = now() - retentionMs - 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 */ } } + const schedule = () => { + if (!persist || timer !== undefined || closed) return + timer = setTimeout(flush, flushMs) + timer.unref?.() } - if (!persist) 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 */ } } + // Files beyond the newest `segments`, segments that hold only expired records, and + // corrupt leftovers older than every kept segment go. + const prune = () => { + const floor = now() - retentionMs + 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 (!persist) 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 */ } } + } } - } - prune() + prune() - return { - dir, - // Appends one record; `names` (optional) refreshes the id -> label table kept with the segment. - append(record, names) { - if (closed) return - let segment = newest() - if (segment.records.length >= segmentRecords) { - // A refused write leaves the newest segments undurable: do not delete the last good one. - const saved = flush() - segment = fresh() - kept.push(segment) - if (saved) prune() - } - segment.records.push(record) - if (names) { - for (const [group, table] of Object.entries(names)) { - segment.names[group] ??= {} - Object.assign(segment.names[group], table) + return { + dir, + // Appends one record; `names` (optional) refreshes the id -> label table kept with the segment. + append(record, names) { + if (closed) return + let segment = newest() + if (segment.records.length >= segmentRecords) { + // A refused write leaves the newest segments undurable: do not delete the last good one. + const saved = flush() + segment = fresh() + kept.push(segment) + if (saved) prune() } - } - dirty = true - schedule() - }, - // The retained records, oldest first, without those past the retention age. - records() { - const floor = now() - retentionMs - return kept.flatMap(segment => segment.records).filter(record => !(record.t < floor)) - }, - // Newest-wins merge of the label tables of the retained segments. - names() { - const out = { bots: {}, rooms: {} } - for (const segment of kept) for (const [group, table] of Object.entries(segment.names)) Object.assign(out[group] ??= {}, table) - 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 }), - close() { - if (closed) return - flush() - closed = true - clearTimeout(timer) - if (lock) { - held.delete(dir) - try { unlinkSync(lock) } catch { /* already gone */ } - } - }, + segment.records.push(record) + if (names) { + for (const [group, table] of Object.entries(names)) { + segment.names[group] ??= {} + Object.assign(segment.names[group], table) + } + } + dirty = true + schedule() + }, + // The retained records, oldest first, without those past the retention age. + records() { + const floor = now() - retentionMs + return kept.flatMap(segment => segment.records).filter(record => !(record.t < floor)) + }, + // Newest-wins merge of the label tables of the retained segments. + names() { + const out = { bots: {}, rooms: {} } + for (const segment of kept) for (const [group, table] of Object.entries(segment.names)) Object.assign(out[group] ??= {}, table) + 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 }), + close() { + if (closed) return + flush() + closed = true + clearTimeout(timer) + if (lock) { + held.delete(dir) + try { unlinkSync(lock) } catch { /* already gone */ } + } + }, + } + } 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 */ } + } + throw error } } export const traceExists = dir => existsSync(dir) && readdirSync(dir).some(name => SEGMENT.test(name)) diff --git a/plugins/timeline/test/store.test.js b/plugins/timeline/test/store.test.js index 9d9a491c9..385b623a0 100644 --- a/plugins/timeline/test/store.test.js +++ b/plugins/timeline/test/store.test.js @@ -1,5 +1,5 @@ import assert from 'node:assert/strict' -import { existsSync, linkSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, symlinkSync, writeFileSync } from 'node:fs' +import { chmodSync, existsSync, linkSync, lstatSync, mkdirSync, mkdtempSync, readdirSync, readFileSync, rmSync, statSync, symlinkSync, writeFileSync } from 'node:fs' import { tmpdir } from 'node:os' import { join } from 'node:path' import { test } from 'node:test' @@ -89,6 +89,33 @@ test('one writer: a live holder is respected, a dead one is taken over, a second again.close() }) +// #2629: an open that refuses after it took the writer lock must give the lock back. A directory +// that exists but cannot be listed takes the lock (creating `writer.lock` needs only `+wx`) and +// then throws on the listing; the only release used to be `close()`, which that path never reached, +// so the process held the directory for good and the lock file stayed for every later process. +test('a refused open releases the writer lock for the next open in this process', (t) => { + const root = mkdtempSync(join(tmpdir(), 'flybot-timeline-store-')) + const dir = join(root, 'trace') + mkdirSync(dir, { recursive: true }) + t.after(() => { + chmodSync(dir, 0o700) + rmSync(root, { recursive: true, force: true }) + }) + chmodSync(dir, 0o300) + // (a) the open throws instead of answering a trace that cannot read its own directory. + assert.throws(() => openTrace({ dir }), (error) => error?.code === 'EACCES', 'the open fails on the unlistable directory') + chmodSync(dir, 0o700) + const lockLeft = existsSync(join(dir, 'writer.lock')) + const reopened = (() => { try { return openTrace({ dir, now: () => 10_000 }) } catch (error) { return error } })() + // (b) the directory is not still held in this process: the same open succeeds once it is listable. + assert.ok(!(reopened instanceof Error), `the same process opens the directory again: ${reopened?.message}`) + // (c) and the failed open left no lock file behind for the next process. + assert.equal(lockLeft, false, 'the refused open left writer.lock on disk') + reopened.append(rec(9_000)) + assert.equal(reopened.flush(), true) + reopened.close() +}) + test('a corrupt or foreign segment is set aside, never crashes the start, and is not read again', (t) => { const dir = tmp() t.after(() => rmSync(dir, { recursive: true, force: true }))