fix(cache): watch errors must call done handler

The type of `watchObject` was incorrect and has been updated to match
the actual request body.

Using this info it was clear that 'ERROR' events were not being handled
correctly. When the watch receives an error it is not always an http
status code, because the status code can only be sent when the stream is
starting. This means that `410` resourceVersion out of date errors could
only be handled if they were detected before the watch stream started
leaving watches running on channels that would never receive more events
and not notifying `ListWatch` consumers of the error.

Author: bbatha

src/cache.ts

@@ -140,7 +140,10 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
 
     private async doneHandler(err: any): Promise<void> {
         this._stop();
-        if (err && err.statusCode === 410) {
+        if (
+            err &&
+            ((err as { statusCode?: number }).statusCode === 410 || (err as { code?: number }).code === 410)
+        ) {
             this.resourceVersion = '';
         } else if (err) {
             this.callbackCache[ERROR].forEach((elt: ErrorCallback) => elt(err));
@@ -192,7 +195,11 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
         });
     }
 
-    private watchHandler(phase: string, obj: T, watchObj?: any): void {
+    private async watchHandler(
+        phase: string,
+        obj: T,
+        watchObj?: { type: string; object: KubernetesObject },
+    ): Promise<void> {
         switch (phase) {
             case 'ERROR':
                 if ((obj as { code?: number }).code === 410) {
@@ -214,10 +221,11 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
             case 'BOOKMARK':
                 // nothing to do, here for documentation, mostly.
                 break;
+            case 'ERROR':
+                await this.doneHandler(obj);
+                return;
         }
-        if (watchObj && watchObj.metadata) {
-            this.resourceVersion = watchObj.metadata.resourceVersion;
-        }
+        this.resourceVersion = obj.metadata!.resourceVersion || '';
     }
 }
 

src/cache_test.ts

@@ -1205,9 +1205,10 @@ describe('ListWatchCache', () => {
             {
                 metadata: {
                     name: 'name3',
+                    resourceVersion: '23456',
                 } as V1ObjectMeta,
             } as V1Namespace,
-            { metadata: { resourceVersion: '23456' } },
+            { type: 'ADDED', metadata: { resourceVersion: '23456' } },
         );
 
         await informer.stop();
@@ -1259,9 +1260,91 @@ describe('ListWatchCache', () => {
             {
                 metadata: {
                     name: 'name3',
+                    resourceVersion: '23456',
                 } as V1ObjectMeta,
             } as V1Namespace,
-            { metadata: { resourceVersion: '23456' } },
+            { type: 'ADDED', metadata: { resourceVersion: '23456' } },
+        );
+
+        await informer.stop();
+
+        let errorEmitted = false;
+        informer.on('error', () => (errorEmitted = true));
+
+        promise = new Promise((resolve) => {
+            mock.when(
+                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+            ).thenCall(() => {
+                resolve(new FakeRequest());
+            });
+        });
+
+        informer.start();
+        await promise;
+
+        const [, , , doneHandler] = mock.capture(fakeWatch.watch).last();
+
+        const object = {
+            kind: 'Status',
+            apiVersion: 'v1',
+            metadata: {},
+            status: 'Failure',
+            message: 'too old resource version: 12345 (1234)',
+            reason: 'Expired',
+            code: 410,
+        };
+        await watchHandler('ERROR', object, { type: 'ERROR', object });
+
+        mock.verify(
+            fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+        ).thrice();
+        expect(errorEmitted).to.equal(false);
+        expect(listCalls).to.be.equal(2);
+    });
+
+    it('should list if the watch errors from the last version', async () => {
+        const fakeWatch = mock.mock(Watch);
+        const list: V1Pod[] = [];
+        const listObj = {
+            metadata: {
+                resourceVersion: '12345',
+            } as V1ListMeta,
+            items: list,
+        } as V1NamespaceList;
+
+        let listCalls = 0;
+        const listFn: ListPromise<V1Namespace> = function(): Promise<{
+            response: http.IncomingMessage;
+            body: V1NamespaceList;
+        }> {
+            return new Promise<{ response: http.IncomingMessage; body: V1NamespaceList }>((resolve) => {
+                listCalls++;
+                resolve({ response: {} as http.IncomingMessage, body: listObj });
+            });
+        };
+        let promise = new Promise((resolve) => {
+            mock.when(
+                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+            ).thenCall(() => {
+                resolve(new FakeRequest());
+            });
+        });
+
+        const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn, false);
+
+        informer.start();
+        await promise;
+
+        const [, , watchHandler] = mock.capture(fakeWatch.watch).last();
+        watchHandler(
+            'ADDED',
+            {
+                metadata: {
+                    name: 'name3',
+                    resourceVersion: '23456',
+                } as V1ObjectMeta,
+            } as V1Namespace,
+            { type: 'ADDED', metadata: { resourceVersion: '23456' } },
         );
 
         await informer.stop();