Release the timeline writer lock when a trace open refuses

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)).

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))

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 }))