Skip to content

Commit 64ba1be

Browse files
fix: a message write no longer waits for the metadata stage of the save before it (two ordered queues); the listener test tracks which callback is removed
1 parent 4731548 commit 64ba1be

3 files changed

Lines changed: 67 additions & 27 deletions

File tree

‎src/core/task/Task.ts‎

Lines changed: 29 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -1786,39 +1786,56 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
17861786
* to replace the stored messages outright.
17871787
*/
17881788
private async saveClineMessages(merge = true): Promise<boolean> {
1789-
// One save at a time per task, from the message write to the task-history write: callers that do not wait for it
1790-
// (the fire-and-forget ones) would otherwise let an older save's metadata land after a newer save's and replace
1791-
// it, an explicit clear of the usage breakdown included. A save that fails does not hold up the next one.
1792-
const run = this.saveClineMessagesQueue.then(
1793-
() => this.persistClineMessages(merge),
1794-
() => this.persistClineMessages(merge),
1789+
// Two queues. The message writes go one at a time, and a write never waits for the metadata of the save before
1790+
// it: the messages are what must reach the disk, so a slow history write cannot delay them. The metadata stage
1791+
// (the history entry, derived from the messages that save persisted) also goes one at a time, in the order the
1792+
// message writes finished, which is the order the saves were asked for: callers that do not wait for a save (the
1793+
// fire-and-forget ones) would otherwise let an older save's metadata land after a newer save's and replace it,
1794+
// an explicit clear of the usage breakdown included. A save that fails does not hold up the next one.
1795+
const written = this.saveClineMessagesQueue.then(
1796+
() => this.writeClineMessages(merge),
1797+
() => this.writeClineMessages(merge),
17951798
)
1796-
this.saveClineMessagesQueue = run.then(
1799+
this.saveClineMessagesQueue = written.then(
17971800
() => undefined,
17981801
() => undefined,
17991802
)
1800-
return run
1803+
1804+
return written.then(async (result) => {
1805+
if (!result.saved) return false
1806+
const metadata = this.saveTaskMetadataQueue.then(() => this.writeTaskMetadata(result.persistedMessages))
1807+
this.saveTaskMetadataQueue = metadata
1808+
await metadata
1809+
return true
1810+
})
18011811
}
18021812

18031813
private saveClineMessagesQueue: Promise<void> = Promise.resolve()
1814+
private saveTaskMetadataQueue: Promise<void> = Promise.resolve()
18041815

1805-
private async persistClineMessages(merge: boolean): Promise<boolean> {
1816+
/** The message write: `saved` is whether it reached the disk, and `persistedMessages` what was actually stored. */
1817+
private async writeClineMessages(
1818+
merge: boolean,
1819+
): Promise<{ saved: false } | { saved: true; persistedMessages: ClineMessage[] | undefined }> {
18061820
// What was actually persisted: with `merge`, newer messages another host wrote are folded in. The metadata
18071821
// (including the usage breakdown) is computed from it, so a stale local snapshot cannot clear usage that
18081822
// the stored messages still contain.
1809-
let persistedMessages: ClineMessage[] | undefined
18101823
try {
1811-
persistedMessages = await saveTaskMessages({
1824+
const persistedMessages = await saveTaskMessages({
18121825
messages: structuredClone(this.clineMessages),
18131826
taskId: this.taskId,
18141827
globalStoragePath: this.globalStoragePath,
18151828
merge,
18161829
})
1830+
return { saved: true, persistedMessages }
18171831
} catch (error) {
18181832
console.error("Failed to save Roo messages:", error)
1819-
return false
1833+
return { saved: false }
18201834
}
1835+
}
18211836

1837+
/** The metadata stage of a save: it never fails the save, whose messages are already persisted. */
1838+
private async writeTaskMetadata(persistedMessages: ClineMessage[] | undefined): Promise<void> {
18221839
try {
18231840
if (this._taskApiConfigName === undefined) {
18241841
await this.taskApiConfigReady
@@ -1853,8 +1870,6 @@ export class Task extends EventEmitter<TaskEvents> implements TaskLike {
18531870
// recomputes and re-emits the metadata update.
18541871
console.error("Failed to save task metadata:", error)
18551872
}
1856-
1857-
return true
18581873
}
18591874

18601875
private findMessageByTimestamp(ts: number): ClineMessage | undefined {

‎src/core/task/__tests__/Task.persistence.spec.ts‎

Lines changed: 27 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1117,7 +1117,7 @@ describe("Task persistence", () => {
11171117
tokenUsage: { totalTokensIn: 0, totalTokensOut: 0, totalCost: 0, contextTokens: 0 },
11181118
})
11191119

1120-
it("does not start a save before the previous save has written its task-history entry", async () => {
1120+
it("does not let the metadata of an older save land after the metadata of a newer one, whatever the timing", async () => {
11211121
const events: string[] = []
11221122
const firstMetadata = createDeferred<ReturnType<typeof metadataFor>>()
11231123
const firstHistoryWrite = createDeferred<undefined>()
@@ -1150,20 +1150,42 @@ describe("Task persistence", () => {
11501150
const newer = save()
11511151
await vi.waitFor(() => expect(mockTaskMetadata).toHaveBeenCalledTimes(1))
11521152

1153-
// The older save is between its two writes; the newer one has not written anything yet.
1154-
expect(events).toEqual(["messages"])
1153+
// The older save is in its metadata stage; the newer one has written its messages but not its metadata.
1154+
expect(events).toEqual(["messages", "messages"])
11551155

11561156
firstMetadata.resolve(metadataFor(false))
11571157
await vi.waitFor(() => expect(updateTaskHistory).toHaveBeenCalledTimes(1))
1158-
expect(events).toEqual(["messages"])
1158+
expect(events).toEqual(["messages", "messages"]) // the newer metadata waits for the older one
11591159
firstHistoryWrite.resolve(undefined)
11601160
await Promise.all([older, newer])
11611161

11621162
// The newer save's entry is the last one written, whatever the timing of the older one.
1163-
expect(events).toEqual(["messages", "history 1", "messages", "history 2"])
1163+
expect(events).toEqual(["messages", "messages", "history 1", "history 2"])
11641164
expect(updateTaskHistory.mock.calls.map(([item]) => item.usage)).toEqual([false, true])
11651165
})
11661166

1167+
it("does not make a message write wait for the metadata stage of the save before it", async () => {
1168+
const slowMetadata = createDeferred<ReturnType<typeof metadataFor>>()
1169+
mockSaveTaskMessages.mockResolvedValue([])
1170+
mockTaskMetadata.mockImplementationOnce(() => slowMetadata.promise)
1171+
const task = new Task({
1172+
provider: mockProvider,
1173+
apiConfiguration: mockApiConfig,
1174+
task: "test task",
1175+
startTask: false,
1176+
})
1177+
const save = () => getTaskPersistenceAccess(task).saveClineMessages()
1178+
1179+
const first = save()
1180+
await vi.waitFor(() => expect(mockTaskMetadata).toHaveBeenCalledTimes(1)) // the first is in its metadata stage
1181+
const second = save()
1182+
1183+
// The second message write reaches the disk while the first metadata is still pending.
1184+
await vi.waitFor(() => expect(mockSaveTaskMessages).toHaveBeenCalledTimes(2))
1185+
slowMetadata.resolve(metadataFor(false))
1186+
await Promise.all([first, second])
1187+
})
1188+
11671189
it("runs the next save even when the one before it failed", async () => {
11681190
mockSaveTaskMessages.mockRejectedValueOnce(new Error("write error")).mockResolvedValueOnce([])
11691191
const task = new Task({

‎src/services/language-models/__tests__/providerQuotas.spec.ts‎

Lines changed: 11 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -1338,17 +1338,20 @@ describe("loadProfiles", () => {
13381338
beforeEach(() => vi.useFakeTimers())
13391339
afterEach(() => vi.useRealTimers())
13401340

1341-
/** Counts the abort listeners that are added and not removed. */
1341+
/** The abort listeners that were added and not removed: a removal counts only for the function that was added. */
13421342
const trackListeners = (signal: AbortSignal) => {
1343-
let live = 0
1343+
const live = new Set<EventListenerOrEventListenerObject>()
13441344
const add = signal.addEventListener.bind(signal)
13451345
const remove = signal.removeEventListener.bind(signal)
1346-
signal.addEventListener = ((...args: Parameters<typeof add>) => (live++, add(...args))) as typeof add
1347-
signal.removeEventListener = ((...args: Parameters<typeof remove>) => (
1348-
live--,
1349-
remove(...args)
1350-
)) as typeof remove
1351-
return () => live
1346+
signal.addEventListener = ((...args: Parameters<typeof add>) => {
1347+
if (args[1]) live.add(args[1])
1348+
return add(...args)
1349+
}) as typeof add
1350+
signal.removeEventListener = ((...args: Parameters<typeof remove>) => {
1351+
if (args[1]) live.delete(args[1])
1352+
return remove(...args)
1353+
}) as typeof remove
1354+
return () => live.size
13521355
}
13531356

13541357
it("rejects at once while the list is pending, and leaves no timer or listener behind", async () => {

0 commit comments

Comments
 (0)