fix(sessions): abort in-flight catalog refresh on dispose

`dispose()` only cleared timers, so a refresh already in flight kept its
provider fetch alive for the rest of the timeout budget and could delay
daemon shutdown, which is exactly when sessiond disposes the refresher.
A refresher-lifetime AbortController is now combined with the per-run
timeout via `AbortSignal.any`, and `dispose()` aborts it. A
dispose-triggered abort logs as expected shutdown info rather than a
timeout warning or an error, whether the runtime resolves as aborted or
rejects.

`start()` is now idempotent: a second call previously overwrote both
timer handles and leaked the first pair, which kept firing.

Also replaces the `then().catch()` bookkeeping chain in `queueRefresh()`
with an awaited private `runCycle()`, keeping the coalescing, retry, and
dispose semantics unchanged.
This commit is contained in:
Federico Jaramillo Martinez
2026-07-25 12:56:31 +02:00
parent 3d3538c76b
commit c5af390ab9
2 changed files with 107 additions and 22 deletions
@@ -173,6 +173,67 @@ describe("ModelCatalogRefresher", () => {
expect(runtime.refresh).not.toHaveBeenCalled(); expect(runtime.refresh).not.toHaveBeenCalled();
}); });
it("aborts the in-flight refresh when disposed", async () => {
const { logger, info, warn, error } = createLogger();
let observed: AbortSignal | undefined;
// Stand in for a provider fetch that only settles when its signal aborts,
// which is what makes an unaborted run delay daemon shutdown.
const refresh = vi.fn((options?: RefreshCall) => {
observed = options?.signal;
return new Promise<RefreshResult>((resolve) => {
options?.signal?.addEventListener("abort", () => { resolve(abortedResult()); });
});
});
const refresher = new ModelCatalogRefresher({ runtime: { refresh }, logger });
refresher.requestRefresh();
await flushMicrotasks();
expect(observed?.aborted).toBe(false);
refresher.dispose();
await flushMicrotasks();
expect(observed?.aborted).toBe(true);
// A deliberate shutdown abort is expected, not a timeout or a fault.
expect(info).toHaveBeenCalledWith({}, "model catalog refresh aborted by dispose; keeping cached catalogs");
expect(warn).not.toHaveBeenCalled();
expect(error).not.toHaveBeenCalled();
});
it("does not report a refresh that rejects because dispose aborted it as a failure", async () => {
const { logger, info, warn, error } = createLogger();
const refresh = vi.fn((options?: RefreshCall) => new Promise<RefreshResult>((_resolve, reject) => {
options?.signal?.addEventListener("abort", () => { reject(new Error("This operation was aborted")); });
}));
const refresher = new ModelCatalogRefresher({ runtime: { refresh }, logger });
refresher.requestRefresh();
await flushMicrotasks();
refresher.dispose();
await flushMicrotasks();
expect(info).toHaveBeenCalledWith({}, "model catalog refresh aborted by dispose; keeping cached catalogs");
expect(warn).not.toHaveBeenCalled();
expect(error).not.toHaveBeenCalled();
});
it("keeps the timers of the first start when start is called again", async () => {
const runtime = createRuntime();
const refresher = new ModelCatalogRefresher({ runtime, initialDelayMs: 1_000, intervalMs: 60_000 });
refresher.start();
refresher.start();
await vi.advanceTimersByTimeAsync(61_000);
// A leaked second timer pair would double every scheduled refresh.
expect(runtime.refresh).toHaveBeenCalledTimes(2);
refresher.dispose();
await vi.advanceTimersByTimeAsync(120_000);
expect(runtime.refresh).toHaveBeenCalledTimes(2);
});
it("does not run a queued follow-up after dispose", async () => { it("does not run a queued follow-up after dispose", async () => {
const gate = deferred<RefreshResult>(); const gate = deferred<RefreshResult>();
const refresh = vi.fn().mockImplementation(() => gate.promise); const refresh = vi.fn().mockImplementation(() => gate.promise);
+38 -14
View File
@@ -72,8 +72,8 @@ const noopLogger: ModelCatalogRefresherLogger = {
* The shared ModelRuntime is constructed offline (see authService.ts), so its * The shared ModelRuntime is constructed offline (see authService.ts), so its
* own refreshes never touch the network and stay fast on request paths. This * own refreshes never touch the network and stay fast on request paths. This
* refresher is the single place that deliberately performs network refreshes — * refresher is the single place that deliberately performs network refreshes —
* bounded by an abort timeout, serialized through one in-flight run, and off * bounded by an abort timeout, serialized through one in-flight run, stopped by
* any request path. `requestRefresh()` additionally asks for a prompt forced * `dispose()` even mid-flight, and off any request path. `requestRefresh()` additionally asks for a prompt forced
* refresh after events that change what should be listed, such as provider * refresh after events that change what should be listed, such as provider
* logins, where the cached catalog is known to be wrong. * logins, where the cached catalog is known to be wrong.
* *
@@ -92,9 +92,16 @@ export class ModelCatalogRefresher {
private initialTimer?: NodeJS.Timeout; private initialTimer?: NodeJS.Timeout;
private intervalTimer?: NodeJS.Timeout; private intervalTimer?: NodeJS.Timeout;
private retryTimer: NodeJS.Timeout | undefined; private retryTimer: NodeJS.Timeout | undefined;
private inflight: Promise<RunOutcome> | undefined; private inflight: Promise<void> | undefined;
private queuedMode: RefreshMode | undefined; private queuedMode: RefreshMode | undefined;
private started = false;
private disposed = false; private disposed = false;
/**
* Aborted by `dispose()` so a refresh already in flight stops with the
* refresher instead of holding its fetch open for the rest of the timeout
* budget, which would delay daemon shutdown.
*/
private readonly lifetime = new AbortController();
constructor(options: ModelCatalogRefresherOptions) { constructor(options: ModelCatalogRefresherOptions) {
this.runtime = options.runtime; this.runtime = options.runtime;
@@ -106,8 +113,10 @@ export class ModelCatalogRefresher {
this.retryDelayMs = options.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS; this.retryDelayMs = options.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS;
} }
/** Idempotent: a second call keeps the timers the first call installed. */
start(): void { start(): void {
if (this.disposed) return; if (this.disposed || this.started) return;
this.started = true;
if (this.offline) { if (this.offline) {
this.logger.info({}, "offline mode is enabled; skipping background model catalog refreshes"); this.logger.info({}, "offline mode is enabled; skipping background model catalog refreshes");
return; return;
@@ -128,11 +137,13 @@ export class ModelCatalogRefresher {
this.queueRefresh("forced"); this.queueRefresh("forced");
} }
/** Terminal: stops the schedule, drops any queued follow-up, and aborts an in-flight run. */
dispose(): void { dispose(): void {
this.disposed = true; this.disposed = true;
if (this.initialTimer !== undefined) clearTimeout(this.initialTimer); if (this.initialTimer !== undefined) clearTimeout(this.initialTimer);
if (this.intervalTimer !== undefined) clearInterval(this.intervalTimer); if (this.intervalTimer !== undefined) clearInterval(this.intervalTimer);
this.clearRetryTimer(); this.clearRetryTimer();
this.lifetime.abort();
} }
/** /**
@@ -149,22 +160,25 @@ export class ModelCatalogRefresher {
this.queuedMode = strongestMode(this.queuedMode, mode); this.queuedMode = strongestMode(this.queuedMode, mode);
return; return;
} }
const run = this.run(mode); // runCycle() reports every failure through the logger and never rejects.
this.inflight = run; this.inflight = this.runCycle(mode, isRetry);
run.then((outcome) => { }
/** One run plus the follow-up it leads to: a queued request, a single retry, or nothing. */
private async runCycle(mode: RefreshMode, isRetry: boolean): Promise<void> {
const outcome = await this.run(mode);
this.inflight = undefined; this.inflight = undefined;
const queued = this.queuedMode; const queued = this.queuedMode;
this.queuedMode = undefined; this.queuedMode = undefined;
if (queued !== undefined) { if (queued !== undefined) {
// A request that arrived during the run replaces any retry this run // A request that arrived during the run replaces any retry this run would
// would have earned; it runs now instead. // have earned; it runs now instead.
this.queueRefresh(queued); this.queueRefresh(queued);
return; return;
} }
// Only a first attempt earns a retry, so a failing provider can never // Only a first attempt earns a retry, so a failing provider can never turn
// turn into a retry loop. // into a retry loop.
if (outcome === "incomplete" && !isRetry) this.scheduleRetry(mode); if (outcome === "incomplete" && !isRetry) this.scheduleRetry(mode);
}).catch(() => { /* run() reports failures through its logger and never rejects */ });
} }
private scheduleRetry(mode: RefreshMode): void { private scheduleRetry(mode: RefreshMode): void {
@@ -188,8 +202,12 @@ export class ModelCatalogRefresher {
const result = await this.runtime.refresh({ const result = await this.runtime.refresh({
allowNetwork: true, allowNetwork: true,
force: mode === "forced", force: mode === "forced",
signal: AbortSignal.timeout(this.timeoutMs), signal: AbortSignal.any([this.lifetime.signal, AbortSignal.timeout(this.timeoutMs)]),
}); });
if (result.aborted && this.lifetime.signal.aborted) {
this.logStoppedByDispose();
return "incomplete";
}
if (result.aborted) { if (result.aborted) {
this.logger.warn({ timeoutMs: this.timeoutMs }, "model catalog refresh timed out; keeping cached catalogs"); this.logger.warn({ timeoutMs: this.timeoutMs }, "model catalog refresh timed out; keeping cached catalogs");
} }
@@ -199,8 +217,14 @@ export class ModelCatalogRefresher {
} }
return result.aborted || result.errors.size > 0 ? "incomplete" : "complete"; return result.aborted || result.errors.size > 0 ? "incomplete" : "complete";
} catch (error: unknown) { } catch (error: unknown) {
this.logger.error({ err: error }, "model catalog refresh failed; keeping cached catalogs"); if (this.lifetime.signal.aborted) this.logStoppedByDispose();
else this.logger.error({ err: error }, "model catalog refresh failed; keeping cached catalogs");
return "incomplete"; return "incomplete";
} }
} }
/** A shutdown abort is expected, so it must not be reported as a timeout or a fault. */
private logStoppedByDispose(): void {
this.logger.info({}, "model catalog refresh aborted by dispose; keeping cached catalogs");
}
} }