fix(anthropic-recovery): fix retry timer memory leak in context window recovery
🤖 Generated with assistance of OhMyOpenCode
This commit is contained in:
@@ -90,6 +90,7 @@ describe("executeCompact lock management", () => {
|
|||||||
pendingCompact: new Set<string>(),
|
pendingCompact: new Set<string>(),
|
||||||
errorDataBySession: new Map(),
|
errorDataBySession: new Map(),
|
||||||
retryStateBySession: new Map(),
|
retryStateBySession: new Map(),
|
||||||
|
retryTimerBySession: new Map(),
|
||||||
truncateStateBySession: new Map(),
|
truncateStateBySession: new Map(),
|
||||||
emptyContentAttemptBySession: new Map(),
|
emptyContentAttemptBySession: new Map(),
|
||||||
compactionInProgress: new Set<string>(),
|
compactionInProgress: new Set<string>(),
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import type { ExperimentalConfig, OhMyOpenCodeConfig } from "../../config"
|
|||||||
import { parseAnthropicTokenLimitError } from "./parser"
|
import { parseAnthropicTokenLimitError } from "./parser"
|
||||||
import { executeCompact, getLastAssistant } from "./executor"
|
import { executeCompact, getLastAssistant } from "./executor"
|
||||||
import { attemptDeduplicationRecovery } from "./deduplication-recovery"
|
import { attemptDeduplicationRecovery } from "./deduplication-recovery"
|
||||||
|
import { clearSessionState } from "./state"
|
||||||
import { log } from "../../shared/logger"
|
import { log } from "../../shared/logger"
|
||||||
|
|
||||||
export interface AnthropicContextWindowLimitRecoveryOptions {
|
export interface AnthropicContextWindowLimitRecoveryOptions {
|
||||||
@@ -17,6 +18,7 @@ function createRecoveryState(): AutoCompactState {
|
|||||||
pendingCompact: new Set<string>(),
|
pendingCompact: new Set<string>(),
|
||||||
errorDataBySession: new Map<string, ParsedTokenLimitError>(),
|
errorDataBySession: new Map<string, ParsedTokenLimitError>(),
|
||||||
retryStateBySession: new Map(),
|
retryStateBySession: new Map(),
|
||||||
|
retryTimerBySession: new Map(),
|
||||||
truncateStateBySession: new Map(),
|
truncateStateBySession: new Map(),
|
||||||
emptyContentAttemptBySession: new Map(),
|
emptyContentAttemptBySession: new Map(),
|
||||||
compactionInProgress: new Set<string>(),
|
compactionInProgress: new Set<string>(),
|
||||||
@@ -30,7 +32,7 @@ export function createAnthropicContextWindowLimitRecoveryHook(
|
|||||||
) {
|
) {
|
||||||
const autoCompactState = createRecoveryState()
|
const autoCompactState = createRecoveryState()
|
||||||
const experimental = options?.experimental
|
const experimental = options?.experimental
|
||||||
const pluginConfig = options?.pluginConfig!
|
const pluginConfig = options?.pluginConfig ?? {} as OhMyOpenCodeConfig
|
||||||
const pendingCompactionTimeoutBySession = new Map<string, ReturnType<typeof setTimeout>>()
|
const pendingCompactionTimeoutBySession = new Map<string, ReturnType<typeof setTimeout>>()
|
||||||
|
|
||||||
const eventHandler = async ({ event }: { event: { type: string; properties?: unknown } }) => {
|
const eventHandler = async ({ event }: { event: { type: string; properties?: unknown } }) => {
|
||||||
@@ -45,12 +47,7 @@ export function createAnthropicContextWindowLimitRecoveryHook(
|
|||||||
pendingCompactionTimeoutBySession.delete(sessionInfo.id)
|
pendingCompactionTimeoutBySession.delete(sessionInfo.id)
|
||||||
}
|
}
|
||||||
|
|
||||||
autoCompactState.pendingCompact.delete(sessionInfo.id)
|
clearSessionState(autoCompactState, sessionInfo.id)
|
||||||
autoCompactState.errorDataBySession.delete(sessionInfo.id)
|
|
||||||
autoCompactState.retryStateBySession.delete(sessionInfo.id)
|
|
||||||
autoCompactState.truncateStateBySession.delete(sessionInfo.id)
|
|
||||||
autoCompactState.emptyContentAttemptBySession.delete(sessionInfo.id)
|
|
||||||
autoCompactState.compactionInProgress.delete(sessionInfo.id)
|
|
||||||
}
|
}
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -28,6 +28,11 @@ export function clearSessionState(
|
|||||||
autoCompactState: AutoCompactState,
|
autoCompactState: AutoCompactState,
|
||||||
sessionID: string,
|
sessionID: string,
|
||||||
): void {
|
): void {
|
||||||
|
const retryTimer = autoCompactState.retryTimerBySession.get(sessionID)
|
||||||
|
if (retryTimer !== undefined) {
|
||||||
|
clearTimeout(retryTimer)
|
||||||
|
autoCompactState.retryTimerBySession.delete(sessionID)
|
||||||
|
}
|
||||||
autoCompactState.pendingCompact.delete(sessionID)
|
autoCompactState.pendingCompact.delete(sessionID)
|
||||||
autoCompactState.errorDataBySession.delete(sessionID)
|
autoCompactState.errorDataBySession.delete(sessionID)
|
||||||
autoCompactState.retryStateBySession.delete(sessionID)
|
autoCompactState.retryStateBySession.delete(sessionID)
|
||||||
@@ -36,6 +41,26 @@ export function clearSessionState(
|
|||||||
autoCompactState.compactionInProgress.delete(sessionID)
|
autoCompactState.compactionInProgress.delete(sessionID)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
export function setRetryTimer(
|
||||||
|
autoCompactState: AutoCompactState,
|
||||||
|
sessionID: string,
|
||||||
|
timeout: ReturnType<typeof setTimeout>,
|
||||||
|
): void {
|
||||||
|
const existingTimer = autoCompactState.retryTimerBySession.get(sessionID)
|
||||||
|
if (existingTimer !== undefined) {
|
||||||
|
clearTimeout(existingTimer)
|
||||||
|
}
|
||||||
|
autoCompactState.retryTimerBySession.set(sessionID, timeout)
|
||||||
|
}
|
||||||
|
|
||||||
|
export function clearRetryTimer(autoCompactState: AutoCompactState, sessionID: string): void {
|
||||||
|
const retryTimer = autoCompactState.retryTimerBySession.get(sessionID)
|
||||||
|
if (retryTimer !== undefined) {
|
||||||
|
clearTimeout(retryTimer)
|
||||||
|
autoCompactState.retryTimerBySession.delete(sessionID)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
export function getEmptyContentAttempt(
|
export function getEmptyContentAttempt(
|
||||||
autoCompactState: AutoCompactState,
|
autoCompactState: AutoCompactState,
|
||||||
sessionID: string,
|
sessionID: string,
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ function createAutoCompactState(): AutoCompactState {
|
|||||||
pendingCompact: new Set<string>(),
|
pendingCompact: new Set<string>(),
|
||||||
errorDataBySession: new Map<string, ParsedTokenLimitError>(),
|
errorDataBySession: new Map<string, ParsedTokenLimitError>(),
|
||||||
retryStateBySession: new Map<string, RetryState>(),
|
retryStateBySession: new Map<string, RetryState>(),
|
||||||
|
retryTimerBySession: new Map(),
|
||||||
truncateStateBySession: new Map(),
|
truncateStateBySession: new Map(),
|
||||||
emptyContentAttemptBySession: new Map(),
|
emptyContentAttemptBySession: new Map(),
|
||||||
compactionInProgress: new Set<string>(),
|
compactionInProgress: new Set<string>(),
|
||||||
@@ -97,10 +98,11 @@ describe("runSummarizeRetryStrategy", () => {
|
|||||||
return 1 as unknown as ReturnType<typeof setTimeout>
|
return 1 as unknown as ReturnType<typeof setTimeout>
|
||||||
}) as typeof setTimeout
|
}) as typeof setTimeout
|
||||||
|
|
||||||
|
autoCompactState.pendingCompact.add(sessionID)
|
||||||
autoCompactState.retryStateBySession.set(sessionID, {
|
autoCompactState.retryStateBySession.set(sessionID, {
|
||||||
attempt: 0,
|
attempt: 0,
|
||||||
lastAttemptTime: Date.now(),
|
lastAttemptTime: Date.now(),
|
||||||
firstAttemptTime: Date.now() - 119900,
|
firstAttemptTime: Date.now() - 100000,
|
||||||
})
|
})
|
||||||
summarizeMock.mockRejectedValueOnce(new Error("rate limited"))
|
summarizeMock.mockRejectedValueOnce(new Error("rate limited"))
|
||||||
|
|
||||||
@@ -117,6 +119,36 @@ describe("runSummarizeRetryStrategy", () => {
|
|||||||
//#then
|
//#then
|
||||||
expect(timeoutCalls.length).toBe(1)
|
expect(timeoutCalls.length).toBe(1)
|
||||||
expect(timeoutCalls[0]!.delay).toBeGreaterThan(0)
|
expect(timeoutCalls[0]!.delay).toBeGreaterThan(0)
|
||||||
expect(timeoutCalls[0]!.delay).toBeLessThanOrEqual(300)
|
expect(timeoutCalls[0]!.delay).toBeLessThanOrEqual(2000)
|
||||||
|
})
|
||||||
|
|
||||||
|
test("#given pending retry timer after session cleanup #when scheduled callback fires #then it does not recreate retry state", async () => {
|
||||||
|
//#given
|
||||||
|
let scheduledCallback: (() => void) | undefined
|
||||||
|
globalThis.setTimeout = ((callback: (...args: unknown[]) => void, _delay?: number) => {
|
||||||
|
scheduledCallback = () => callback()
|
||||||
|
return 1 as unknown as ReturnType<typeof setTimeout>
|
||||||
|
}) as typeof setTimeout
|
||||||
|
|
||||||
|
autoCompactState.pendingCompact.add(sessionID)
|
||||||
|
summarizeMock.mockRejectedValueOnce(new Error("rate limited"))
|
||||||
|
|
||||||
|
await runSummarizeRetryStrategy({
|
||||||
|
sessionID,
|
||||||
|
msg: { providerID: "anthropic", modelID: "claude-sonnet-4-6" },
|
||||||
|
autoCompactState,
|
||||||
|
client: client as never,
|
||||||
|
directory,
|
||||||
|
pluginConfig: {} as OhMyOpenCodeConfig,
|
||||||
|
})
|
||||||
|
|
||||||
|
autoCompactState.pendingCompact.delete(sessionID)
|
||||||
|
autoCompactState.retryStateBySession.delete(sessionID)
|
||||||
|
|
||||||
|
//#when
|
||||||
|
scheduledCallback?.()
|
||||||
|
|
||||||
|
//#then
|
||||||
|
expect(autoCompactState.retryStateBySession.has(sessionID)).toBe(false)
|
||||||
})
|
})
|
||||||
})
|
})
|
||||||
|
|||||||
@@ -2,7 +2,13 @@ import type { AutoCompactState } from "./types"
|
|||||||
import type { OhMyOpenCodeConfig } from "../../config"
|
import type { OhMyOpenCodeConfig } from "../../config"
|
||||||
import { RETRY_CONFIG } from "./types"
|
import { RETRY_CONFIG } from "./types"
|
||||||
import type { Client } from "./client"
|
import type { Client } from "./client"
|
||||||
import { clearSessionState, getEmptyContentAttempt, getOrCreateRetryState } from "./state"
|
import {
|
||||||
|
clearRetryTimer,
|
||||||
|
clearSessionState,
|
||||||
|
getEmptyContentAttempt,
|
||||||
|
getOrCreateRetryState,
|
||||||
|
setRetryTimer,
|
||||||
|
} from "./state"
|
||||||
import { sanitizeEmptyMessagesBeforeSummarize } from "./message-builder"
|
import { sanitizeEmptyMessagesBeforeSummarize } from "./message-builder"
|
||||||
import { fixEmptyMessages } from "./empty-content-recovery"
|
import { fixEmptyMessages } from "./empty-content-recovery"
|
||||||
|
|
||||||
@@ -19,6 +25,11 @@ export async function runSummarizeRetryStrategy(params: {
|
|||||||
errorType?: string
|
errorType?: string
|
||||||
messageIndex?: number
|
messageIndex?: number
|
||||||
}): Promise<void> {
|
}): Promise<void> {
|
||||||
|
if (!params.autoCompactState.pendingCompact.has(params.sessionID)) {
|
||||||
|
clearRetryTimer(params.autoCompactState, params.sessionID)
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
const retryState = getOrCreateRetryState(params.autoCompactState, params.sessionID)
|
const retryState = getOrCreateRetryState(params.autoCompactState, params.sessionID)
|
||||||
const now = Date.now()
|
const now = Date.now()
|
||||||
|
|
||||||
@@ -42,6 +53,8 @@ export async function runSummarizeRetryStrategy(params: {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
clearRetryTimer(params.autoCompactState, params.sessionID)
|
||||||
|
|
||||||
if (params.errorType?.includes("non-empty content")) {
|
if (params.errorType?.includes("non-empty content")) {
|
||||||
const attempt = getEmptyContentAttempt(params.autoCompactState, params.sessionID)
|
const attempt = getEmptyContentAttempt(params.autoCompactState, params.sessionID)
|
||||||
if (attempt < 3) {
|
if (attempt < 3) {
|
||||||
@@ -52,9 +65,11 @@ export async function runSummarizeRetryStrategy(params: {
|
|||||||
messageIndex: params.messageIndex,
|
messageIndex: params.messageIndex,
|
||||||
})
|
})
|
||||||
if (fixed) {
|
if (fixed) {
|
||||||
setTimeout(() => {
|
const timeout = setTimeout(() => {
|
||||||
|
params.autoCompactState.retryTimerBySession.delete(params.sessionID)
|
||||||
void runSummarizeRetryStrategy(params)
|
void runSummarizeRetryStrategy(params)
|
||||||
}, 500)
|
}, 500)
|
||||||
|
setRetryTimer(params.autoCompactState, params.sessionID, timeout)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
@@ -138,9 +153,11 @@ export async function runSummarizeRetryStrategy(params: {
|
|||||||
Math.pow(RETRY_CONFIG.backoffFactor, retryState.attempt - 1)
|
Math.pow(RETRY_CONFIG.backoffFactor, retryState.attempt - 1)
|
||||||
const cappedDelay = Math.min(delay, RETRY_CONFIG.maxDelayMs, remainingTimeMs)
|
const cappedDelay = Math.min(delay, RETRY_CONFIG.maxDelayMs, remainingTimeMs)
|
||||||
|
|
||||||
setTimeout(() => {
|
const timeout = setTimeout(() => {
|
||||||
|
params.autoCompactState.retryTimerBySession.delete(params.sessionID)
|
||||||
void runSummarizeRetryStrategy(params)
|
void runSummarizeRetryStrategy(params)
|
||||||
}, cappedDelay)
|
}, cappedDelay)
|
||||||
|
setRetryTimer(params.autoCompactState, params.sessionID, timeout)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
|
|||||||
@@ -23,6 +23,7 @@ export interface AutoCompactState {
|
|||||||
pendingCompact: Set<string>
|
pendingCompact: Set<string>
|
||||||
errorDataBySession: Map<string, ParsedTokenLimitError>
|
errorDataBySession: Map<string, ParsedTokenLimitError>
|
||||||
retryStateBySession: Map<string, RetryState>
|
retryStateBySession: Map<string, RetryState>
|
||||||
|
retryTimerBySession: Map<string, ReturnType<typeof setTimeout>>
|
||||||
truncateStateBySession: Map<string, TruncateState>
|
truncateStateBySession: Map<string, TruncateState>
|
||||||
emptyContentAttemptBySession: Map<string, number>
|
emptyContentAttemptBySession: Map<string, number>
|
||||||
compactionInProgress: Set<string>
|
compactionInProgress: Set<string>
|
||||||
|
|||||||
Reference in New Issue
Block a user