Skip to content
This repository was archived by the owner on May 15, 2026. It is now read-only.

Commit d35178a

Browse files
hannesrudolphhannesrudolph
committed
fix: address race condition in task purge by processing deletions sequentially
1 parent 1c0bc9c commit d35178a

1 file changed

Lines changed: 87 additions & 84 deletions

File tree

‎src/utils/task-history-retention.ts‎

Lines changed: 87 additions & 84 deletions
Original file line numberDiff line numberDiff line change
@@ -142,107 +142,110 @@ export async function purgeOldTasks(
142142
return !(await pathExists(dir))
143143
}
144144

145-
const results = await Promise.all(
146-
taskDirs.map(async (d) => {
147-
const taskDir = path.join(tasksDir, d.name)
148-
const metadataPath = path.join(taskDir, GlobalFileNames.taskMetadata)
145+
const results: number[] = []
149146

150-
let ts: number | null = null
147+
for (const d of taskDirs) {
148+
const taskDir = path.join(tasksDir, d.name)
149+
const metadataPath = path.join(taskDir, GlobalFileNames.taskMetadata)
151150

152-
// First try to get a timestamp from task_metadata.json (if present)
153-
try {
154-
const raw = await fs.readFile(metadataPath, "utf8")
155-
const meta = JSON.parse(raw)
156-
const maybeTs = Number((meta as any)?.ts)
157-
if (Number.isFinite(maybeTs)) {
158-
ts = maybeTs
159-
}
160-
} catch {
161-
// Missing or invalid metadata; we'll fall back to directory mtime.
151+
let ts: number | null = null
152+
153+
// First try to get a timestamp from task_metadata.json (if present)
154+
try {
155+
const raw = await fs.readFile(metadataPath, "utf8")
156+
const meta = JSON.parse(raw)
157+
const maybeTs = Number((meta as any)?.ts)
158+
if (Number.isFinite(maybeTs)) {
159+
ts = maybeTs
162160
}
161+
} catch {
162+
// Missing or invalid metadata; we'll fall back to directory mtime.
163+
}
163164

164-
let shouldDelete = false
165-
let reason = ""
165+
let shouldDelete = false
166+
let reason = ""
166167

167-
// Check for checkpoint-only orphan directories (delete regardless of age)
168+
// Check for checkpoint-only orphan directories (delete regardless of age)
169+
try {
170+
const childEntries = await fs.readdir(taskDir, { withFileTypes: true })
171+
const visibleNames = childEntries.map((e) => e.name).filter((n) => !n.startsWith("."))
172+
const hasCheckpointsDir = childEntries.some((e) => e.isDirectory() && e.name === "checkpoints")
173+
const nonCheckpointVisible = visibleNames.filter((n) => n !== "checkpoints")
174+
const hasMetadataFile = visibleNames.includes(GlobalFileNames.taskMetadata)
175+
if (hasCheckpointsDir && nonCheckpointVisible.length === 0 && !hasMetadataFile) {
176+
shouldDelete = true
177+
reason = "orphan checkpoints_only"
178+
}
179+
} catch {
180+
// Ignore errors while scanning children; proceed with normal logic
181+
}
182+
183+
if (!shouldDelete && ts !== null && ts < cutoff) {
184+
// Normal case: metadata has a valid ts older than cutoff
185+
shouldDelete = true
186+
reason = `ts=${ts}`
187+
} else if (!shouldDelete) {
188+
// Orphan/legacy case: no valid ts; fall back to directory mtime
168189
try {
169-
const childEntries = await fs.readdir(taskDir, { withFileTypes: true })
170-
const visibleNames = childEntries.map((e) => e.name).filter((n) => !n.startsWith("."))
171-
const hasCheckpointsDir = childEntries.some((e) => e.isDirectory() && e.name === "checkpoints")
172-
const nonCheckpointVisible = visibleNames.filter((n) => n !== "checkpoints")
173-
const hasMetadataFile = visibleNames.includes(GlobalFileNames.taskMetadata)
174-
if (hasCheckpointsDir && nonCheckpointVisible.length === 0 && !hasMetadataFile) {
190+
const stat = await fs.stat(taskDir)
191+
const mtimeMs = stat.mtime.getTime()
192+
if (mtimeMs < cutoff) {
175193
shouldDelete = true
176-
reason = "orphan checkpoints_only"
194+
reason = `no valid ts, mtime=${stat.mtime.toISOString()}`
177195
}
178196
} catch {
179-
// Ignore errors while scanning children; proceed with normal logic
180-
}
181-
182-
if (!shouldDelete && ts !== null && ts < cutoff) {
183-
// Normal case: metadata has a valid ts older than cutoff
184-
shouldDelete = true
185-
reason = `ts=${ts}`
186-
} else if (!shouldDelete) {
187-
// Orphan/legacy case: no valid ts; fall back to directory mtime
188-
try {
189-
const stat = await fs.stat(taskDir)
190-
const mtimeMs = stat.mtime.getTime()
191-
if (mtimeMs < cutoff) {
192-
shouldDelete = true
193-
reason = `no valid ts, mtime=${stat.mtime.toISOString()}`
194-
}
195-
} catch {
196-
// If we can't stat the directory, skip it.
197-
}
197+
// If we can't stat the directory, skip it.
198198
}
199+
}
199200

200-
if (!shouldDelete) {
201-
return 0
202-
}
201+
if (!shouldDelete) {
202+
results.push(0)
203+
continue
204+
}
203205

204-
if (dryRun) {
205-
logv(`[Retention][DRY RUN] Would delete task ${d.name} (${reason}) @ ${taskDir}`)
206-
return 1
207-
}
206+
if (dryRun) {
207+
logv(`[Retention][DRY RUN] Would delete task ${d.name} (${reason}) @ ${taskDir}`)
208+
results.push(1)
209+
continue
210+
}
208211

209-
// Attempt deletion using provider callback (for full cleanup) or direct rm
210-
let deletionError: unknown | null = null
211-
try {
212-
if (deleteTaskById) {
213-
logv(`[Retention] Deleting task ${d.name} via provider @ ${taskDir} (${reason})`)
214-
await deleteTaskById(d.name, taskDir)
215-
} else {
216-
logv(`[Retention] Deleting task ${d.name} via fs.rm @ ${taskDir} (${reason})`)
217-
await fs.rm(taskDir, { recursive: true, force: true })
218-
}
219-
} catch (e) {
220-
deletionError = e
212+
// Attempt deletion using provider callback (for full cleanup) or direct rm
213+
let deletionError: unknown | null = null
214+
try {
215+
if (deleteTaskById) {
216+
logv(`[Retention] Deleting task ${d.name} via provider @ ${taskDir} (${reason})`)
217+
await deleteTaskById(d.name, taskDir)
218+
} else {
219+
logv(`[Retention] Deleting task ${d.name} via fs.rm @ ${taskDir} (${reason})`)
220+
await fs.rm(taskDir, { recursive: true, force: true })
221221
}
222+
} catch (e) {
223+
deletionError = e
224+
}
222225

223-
// Verify deletion; if still exists, attempt aggressive cleanup with retries
224-
let deleted = await removeDirAggressive(taskDir)
225-
226-
if (!deleted) {
227-
// Did not actually remove; report the most relevant error
228-
if (deletionError) {
229-
log?.(
230-
`[Retention] Failed to delete task ${d.name} @ ${taskDir}: ${
231-
deletionError instanceof Error ? deletionError.message : String(deletionError)
232-
} (directory still present)`,
233-
)
234-
} else {
235-
log?.(
236-
`[Retention] Failed to delete task ${d.name} @ ${taskDir}: directory still present after cleanup attempts`,
237-
)
238-
}
239-
return 0
226+
// Verify deletion; if still exists, attempt aggressive cleanup with retries
227+
let deleted = await removeDirAggressive(taskDir)
228+
229+
if (!deleted) {
230+
// Did not actually remove; report the most relevant error
231+
if (deletionError) {
232+
log?.(
233+
`[Retention] Failed to delete task ${d.name} @ ${taskDir}: ${
234+
deletionError instanceof Error ? deletionError.message : String(deletionError)
235+
} (directory still present)`,
236+
)
237+
} else {
238+
log?.(
239+
`[Retention] Failed to delete task ${d.name} @ ${taskDir}: directory still present after cleanup attempts`,
240+
)
240241
}
242+
results.push(0)
243+
continue
244+
}
241245

242-
logv(`[Retention] Deleted task ${d.name} (${reason}) @ ${taskDir}`)
243-
return 1
244-
}),
245-
)
246+
logv(`[Retention] Deleted task ${d.name} (${reason}) @ ${taskDir}`)
247+
results.push(1)
248+
}
246249

247250
const purged = results.reduce<number>((sum, n) => sum + n, 0)
248251

0 commit comments

Comments
 (0)