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

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

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

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

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

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

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

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

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