Merge pull request #3022 from duizabojul/fix/watch-inactivity-timeout

fix: use inactivity timeout for watch instead of whole-request timeout

Author: kubernetes-prow[bot]

src/cache.ts

@@ -168,7 +168,11 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, 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;

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<V1Namespace> = () => 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', () => {

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

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