From 6086604279685320c526afc90372b9f84272926d Mon Sep 17 00:00:00 2001 From: diouze Date: Mon, 24 Aug 2026 15:51:05 +0200 Subject: [PATCH] fix: use inactivity timeout for watch instead of whole-request timeout Watch.watch() passed AbortSignal.timeout(requestTimeoutMs) to fetch, which counts from request start and stays armed while the response body streams. Any watch, healthy or not, was therefore aborted with a TimeoutError after 30s. Replace it with a resettable timer on the AbortController: it is armed before the fetch (connect timeout), then re-armed on the 200 response and on every body chunk, so requestTimeoutMs now means "max time without data". The timer is unref'd and cleared when the watch is done. No public API change. Also reset ListWatch's reconnectDelayMs to 0 in the TimeoutError branch of doneHandler. A client-side timeout means we already waited the full request timeout, so reconnecting immediately cannot tight-loop, while backing off would leave quiet resources unwatched for up to MAX_RECONNECT_DELAY_MS. Exponential backoff still applies to real errors and server-side disconnects. --- src/cache.ts | 6 +++- src/cache_test.ts | 59 ++++++++++++++++++++++++++++++++- src/watch.ts | 30 ++++++++++++++--- src/watch_test.ts | 84 +++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 173 insertions(+), 6 deletions(-) diff --git a/src/cache.ts b/src/cache.ts index c0d2e653483..d7fcbc9dd6a 100644 --- a/src/cache.ts +++ b/src/cache.ts @@ -168,7 +168,11 @@ export class ListWatch implements ObjectCache, In ) { this.resourceVersion = ''; } else if (err && (err as { name?: string }).name === 'TimeoutError') { - // Watch client-side timeout — reconnect from last known resourceVersion + // Watch client-side timeout — reconnect from last known resourceVersion. + // The timeout itself already throttled us for the full request timeout, so + // reconnecting immediately cannot tight-loop and backing off would leave + // quiet resources unwatched for up to MAX_RECONNECT_DELAY_MS. + this.reconnectDelayMs = 0; } else if (err) { this.callbackCache[ERROR].forEach((elt: ErrorCallback) => elt(err)); return; diff --git a/src/cache_test.ts b/src/cache_test.ts index 59f5228ae35..20fb08f78cd 100644 --- a/src/cache_test.ts +++ b/src/cache_test.ts @@ -1639,7 +1639,7 @@ describe('ListWatchCache', () => { deepStrictEqual(delayValues, [1000]); }); - it('should reconnect with backoff on TimeoutError', async () => { + it('should reconnect on TimeoutError', async () => { const fakeWatch = mock.mock(Watch); const listObj = { metadata: { resourceVersion: '12345' } as V1ListMeta, @@ -1682,6 +1682,63 @@ describe('ListWatchCache', () => { strictEqual(watchCalls, 2); strictEqual(errorEmitted, false); }); + + it('should not back off between repeated TimeoutErrors', async () => { + const fakeWatch = mock.mock(Watch); + const listObj = { + metadata: { resourceVersion: '12345' } as V1ListMeta, + items: [] as V1Namespace[], + } as V1NamespaceList; + + const listFn: ListPromise = () => Promise.resolve(listObj); + + let watchCalls = 0; + const delayValues: number[] = []; + const promise = new Promise((resolve) => { + mock.when( + fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()), + ).thenCall(() => { + watchCalls++; + resolve(new AbortController()); + return Promise.resolve(new AbortController()); + }); + }); + + // ListWatch is constructed for its side effects (starts watching) + // eslint-disable-next-line @typescript-eslint/no-unused-vars + const cache = new ListWatch( + '/some/path', + mock.instance(fakeWatch), + listFn, + true, + undefined, + undefined, + { + delayFn: (ms: number) => { + delayValues.push(ms); + return Promise.resolve(); + }, + }, + ); + await promise; + strictEqual(watchCalls, 1); + + const [, , , doneHandler] = mock.capture(fakeWatch.watch).last(); + + const timeoutError = () => + new DOMException('The operation was aborted due to timeout', 'TimeoutError'); + + await doneHandler(timeoutError()); + await doneHandler(timeoutError()); + await doneHandler(timeoutError()); + + strictEqual(watchCalls, 4); + deepStrictEqual(delayValues, []); + + // Backoff is still applied to non-timeout reconnects. + await doneHandler(null); + deepStrictEqual(delayValues, [1000]); + }); }); describe('delete items', () => { diff --git a/src/watch.ts b/src/watch.ts index 85058117e2e..53a79b94315 100644 --- a/src/watch.ts +++ b/src/watch.ts @@ -40,8 +40,25 @@ export class Watch { } const controller = new AbortController(); - const timeoutSignal = AbortSignal.timeout(this.requestTimeoutMs); - const signal = AbortSignal.any([controller.signal, timeoutSignal]); + + let timedOut: boolean = false; + let timer: NodeJS.Timeout | undefined; + const clearTimer = () => { + if (timer !== undefined) { + clearTimeout(timer); + timer = undefined; + } + }; + const resetTimer = () => { + clearTimer(); + timer = setTimeout(() => { + timedOut = true; + controller.abort( + new DOMException('The operation was aborted due to timeout', 'TimeoutError'), + ); + }, this.requestTimeoutMs); + timer.unref(); + }; const ctx = new RequestContext(watchURL.toString(), HttpMethod.GET); await this.config.applySecurityAuthentication(ctx); @@ -50,8 +67,9 @@ export class Watch { const doneCallOnce = (err: any) => { if (!doneCalled) { doneCalled = true; + clearTimer(); controller.abort(); - if (err && timeoutSignal.aborted) { + if (err && timedOut) { done(new DOMException('The operation was aborted due to timeout', 'TimeoutError')); } else { done(err); @@ -60,14 +78,17 @@ export class Watch { }; try { + resetTimer(); const response = await fetch(watchURL, { method: 'GET', headers: ctx.getHeaders(), dispatcher: ctx.getDispatcher(), - signal, + signal: controller.signal, }); if (response.status === 200) { + // The connect timeout is over; from here on it becomes an inactivity timeout. + resetTimer(); const body = Readable.fromWeb(response.body! as any); body.on('error', doneCallOnce); @@ -86,6 +107,7 @@ export class Watch { // ignore parse errors } }); + body.on('data', resetTimer); } else { const statusText = response.statusText || STATUS_CODES[response.status] || 'Internal Server Error'; diff --git a/src/watch_test.ts b/src/watch_test.ts index 4aad61a05d9..0b953b1787b 100644 --- a/src/watch_test.ts +++ b/src/watch_test.ts @@ -386,6 +386,90 @@ describe('Watch', () => { strictEqual(doneErr.name, 'TimeoutError'); }); + it('should not timeout while the server keeps sending events', async (t) => { + const eventCount = 12; + const eventIntervalMs = 30; + const requestTimeoutMs = 150; + + const kc = await setupMockSystem(t, (_req, res) => { + let sent = 0; + const interval = setInterval(() => { + res.write(JSON.stringify({ type: 'ADDED', object: { name: `obj${sent}` } }) + '\n'); + sent += 1; + if (sent === eventCount) { + clearInterval(interval); + res.end(); + } + }, eventIntervalMs); + }); + const watch = new Watch(kc); + + // NOTE: Hack around the type system to make the timeout shorter + (watch as any).requestTimeoutMs = requestTimeoutMs; + + const receivedObjects: any[] = []; + let doneErr: any; + + let doneResolve: () => void; + const donePromise = new Promise((resolve) => { + doneResolve = resolve; + }); + + await watch.watch( + '/some/path/to/object', + {}, + (_phase: string, obj: any) => { + receivedObjects.push(obj); + }, + (err: any) => { + doneErr = err; + doneResolve(); + }, + ); + + await donePromise; + + // The stream lived well past requestTimeoutMs because every event reset the timeout. + strictEqual(receivedObjects.length, eventCount); + strictEqual(doneErr, null); + }); + + it('should timeout when the server goes silent after connecting', async (t) => { + const kc = await setupMockSystem(t, (_req, res) => { + res.write(JSON.stringify({ type: 'ADDED', object: { name: 'obj' } }) + '\n'); + // Then stay silent forever. + }); + const watch = new Watch(kc); + + // NOTE: Hack around the type system to make the timeout shorter + (watch as any).requestTimeoutMs = 100; + + const receivedObjects: any[] = []; + let doneErr: any; + + let doneResolve: () => void; + const donePromise = new Promise((resolve) => { + doneResolve = resolve; + }); + + await watch.watch( + '/some/path/to/object', + {}, + (_phase: string, obj: any) => { + receivedObjects.push(obj); + }, + (err: any) => { + doneErr = err; + doneResolve(); + }, + ); + + await donePromise; + + deepStrictEqual(receivedObjects, [{ name: 'obj' }]); + strictEqual(doneErr.name, 'TimeoutError'); + }); + it('should throw on empty config', async () => { const kc = new KubeConfig(); const watch = new Watch(kc);