WIP Undici migration p2

examples/raw-example.js

@@ -1,5 +1,5 @@
 import * as k8s from '@kubernetes/client-node';
-import fetch from 'node-fetch';
+import { fetch } from 'undici';
 import https from 'node:https';
 
 const kc = new k8s.KubeConfig();

package.json

@@ -58,14 +58,12 @@
     "dependencies": {
         "@types/js-yaml": "^4.0.1",
         "@types/node": "^25.0.0",
-        "@types/node-fetch": "^2.6.13",
         "@types/stream-buffers": "^3.0.3",
         "form-data": "^4.0.0",
         "hpagent": "^1.2.0",
         "isomorphic-ws": "^5.0.0",
         "js-yaml": "^4.1.0",
         "jsonpath-plus": "^10.3.0",
-        "node-fetch": "^2.7.0",
         "openid-client": "^6.1.3",
         "rfc4648": "^1.3.0",
         "socks": "^2.8.4",
@@ -83,7 +81,6 @@
         "eslint-plugin-erasable-syntax-only": "^0.4.0",
         "husky": "^9.0.6",
         "mock-fs": "^5.2.0",
-        "nock": "^14.0.5",
         "prettier": "^3.0.0",
         "pretty-quick": "^4.0.0",
         "ts-mockito": "^2.3.1",

src/cache.ts

@@ -24,12 +24,19 @@ export interface ObjectCache<T> {
 export type CacheMap<T extends KubernetesObject> = Map<string, Map<string, T>>;
 
 export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, Informer<T> {
+    private static readonly BASE_RECONNECT_DELAY_MS = 1000;
+    private static readonly MAX_RECONNECT_DELAY_MS = 30000;
+
     private objects: CacheMap<T> = new Map();
     private resourceVersion: string;
     private readonly indexCache: { [key: string]: T[] } = {};
     private readonly callbackCache: { [key: string]: (ObjectCallback<T> | ErrorCallback)[] } = {};
     private request: AbortController | undefined;
     private stopped: boolean = false;
+    private reconnectDelayMs: number = 0;
+    private hasConnected: boolean = false;
+    private delayFn: (ms: number) => Promise<void> = (ms) =>
+        new Promise((resolve) => setTimeout(resolve, ms));
     private readonly path: string;
     private readonly watch: Watch;
     private readonly listFn: ListPromise<T>;
@@ -63,6 +70,8 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
 
     public async start(): Promise<void> {
         this.stopped = false;
+        this.reconnectDelayMs = 0;
+        this.hasConnected = false;
         await this.doneHandler(null);
     }
 
@@ -151,6 +160,8 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
             ((err as { statusCode?: number }).statusCode === 410 || (err as { code?: number }).code === 410)
         ) {
             this.resourceVersion = '';
+        } else if (err && (err as { name?: string }).name === 'TimeoutError') {
+            // Watch client-side timeout — reconnect from last known resourceVersion
         } else if (err) {
             this.callbackCache[ERROR].forEach((elt: ErrorCallback) => elt(err));
             return;
@@ -186,6 +197,16 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
         if (this.fieldSelector !== undefined) {
             queryParams.fieldSelector = ObjectSerializer.serialize(this.fieldSelector, 'string');
         }
+        if (this.reconnectDelayMs > 0 && this.hasConnected) {
+            await this.delayFn(this.reconnectDelayMs);
+        }
+        if (this.hasConnected) {
+            this.reconnectDelayMs = Math.min(
+                this.reconnectDelayMs > 0 ? this.reconnectDelayMs * 2 : ListWatch.BASE_RECONNECT_DELAY_MS,
+                ListWatch.MAX_RECONNECT_DELAY_MS,
+            );
+        }
+        this.hasConnected = true;
         this.request = await this.watch.watch(
             this.path,
             queryParams,
@@ -236,6 +257,7 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
                 // nothing to do, here for documentation, mostly.
                 break;
         }
+        this.reconnectDelayMs = 0;
         this.resourceVersion = obj.metadata ? obj.metadata!.resourceVersion || '' : '';
     }
 }