fix(cache): watch errors must call done handler

src/cache.ts

@@ -120,7 +120,10 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
 
     private async doneHandler(err: any): Promise<any> {
         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));
@@ -187,7 +190,11 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
         addOrUpdateObject(namespaceList, obj);
     }
 
-    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 'ADDED':
             case 'MODIFIED':
@@ -213,10 +220,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

@@ -1096,9 +1096,10 @@ describe('ListWatchCache', () => {
             {
                 metadata: {
                     name: 'name3',
+                    resourceVersion: '23456',
                 } as V1ObjectMeta,
             } as V1Namespace,
-            { metadata: { resourceVersion: '23456' } },
+            { type: 'ADDED', metadata: { resourceVersion: '23456' } },
         );
 
         await informer.stop();
@@ -1153,9 +1154,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();