Repository: kubernetes-client/javascript
Pull request #2389 (2025-04-24 09:17:38)

watch: prevent done callback from being called twice on connection loss
Avoids multiple invocations of the done callback when a watch connection is closed unexpectedly (e.g., a premature socket close). The fix sets `doneCalled = true` before calling `controller.abort()`, to ensure that any `AbortError` triggered by aborting does not result in another call to the done callback.

A regression test is included to simulate an abrupt connection termination. It uses a real HTTP server that flushes headers and then destroys the connection immediately. This reliably triggers the combination of `AbortError` and "Premature close" needed to reproduce the issue.

Refs: https://github.com/kubernetes-client/javascript/issues/2387


Patch:
diff --git a/examples/patch-example.js b/examples/patch-example.js
index d77df111..d3358c48 100644
--- a/examples/patch-example.js
+++ b/examples/patch-example.js
@@ -20,7 +20,7 @@ try {
 
     await k8sApi.patchNamespacedPod(
         { name: res?.items?.[0]?.metadata?.name ?? '', namespace, body: patch },
-        k8s.setHeaderOptions('Content-Type', k8s.PatchStrategy.JsonPatch),
+        k8s.setHeaderOptions('Content-Type', k8s.JsonPatch),
     );
 
     console.log('Patched.');
diff --git a/src/watch.ts b/src/watch.ts
index 702f0f3d..98db927d 100644
--- a/src/watch.ts
+++ b/src/watch.ts
@@ -46,8 +46,8 @@ export class Watch {
         let doneCalled: boolean = false;
         const doneCallOnce = (err: any) => {
             if (!doneCalled) {
-                controller.abort();
                 doneCalled = true;
+                controller.abort();
                 done(err);
             }
         };


Test patch:
diff --git a/src/watch_test.ts b/src/watch_test.ts
index fd2fe60d..58252b12 100644
--- a/src/watch_test.ts
+++ b/src/watch_test.ts
@@ -5,7 +5,8 @@ import { PassThrough } from 'node:stream';
 import { KubeConfig } from './config.js';
 import { Cluster, Context, User } from './config_types.js';
 import { Watch } from './watch.js';
-import { IncomingMessage } from 'node:http';
+import { IncomingMessage, createServer } from 'node:http';
+import { AddressInfo } from 'node:net';
 
 const server = 'https://foo.company.com';
 
@@ -167,6 +168,69 @@ describe('Watch', () => {
         stream.destroy();
     });
 
+    it('should not call the done callback more than once on unexpected connection loss', async () => {
+        // Create a server that accepts the connection and flushes headers, then
+        // immediately destroys the connection (causing a "Premature close"
+        // error).
+        //
+        // This reproduces a bug where AbortController.abort() inside
+        // doneCallOnce could cause done() to be invoked twice.
+
+        const mockServer = createServer((req, res) => {
+            res.writeHead(200, {
+                'Content-Type': 'application/json',
+                'Transfer-Encoding': 'chunked',
+            });
+
+            res.flushHeaders();
+            res.destroy(); // Prematurely close the connection
+        });
+
+        const mockServerPort = await new Promise<number>((resolve) => {
+            mockServer.listen(0, () => {
+                resolve((mockServer.address() as AddressInfo).port);
+            });
+        });
+
+        const kc = new KubeConfig();
+
+        kc.loadFromClusterAndUser(
+            {
+                name: 'cluster',
+                server: `http://localhost:${mockServerPort}`,
+                skipTLSVerify: true,
+            },
+            {
+                name: 'user',
+            },
+        );
+
+        const watch = new Watch(kc);
+
+        let doneCalled = 0;
+        let doneResolve: () => void;
+
+        const donePromise = new Promise<void>((resolve) => {
+            doneResolve = resolve;
+        });
+
+        await watch.watch(
+            '/some/path/to/object',
+            {},
+            () => {},
+            () => {
+                doneCalled += 1;
+                doneResolve();
+            },
+        );
+
+        await donePromise;
+
+        mockServer.close();
+
+        strictEqual(doneCalled, 1);
+    });
+
     it('should call setKeepAlive on the socket to extend the default of 5 mins', async () => {
         const kc = new KubeConfig();