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);