Pagination watch fixes

shell/components/nav/TopLevelMenu.helper.ts

@@ -1,6 +1,7 @@
 import { CAPI, MANAGEMENT } from '@shell/config/types';
 import { STORE } from '@shell/store/store-types';
 import { PaginationParam, PaginationParamFilter, PaginationSort } from '@shell/types/store/pagination.types';
+import { STEVE_WATCH_EVENT_LISTENER_ARGS } from '@shell/types/store/subscribe.types';
 import { VuexStore } from '@shell/types/store/vuex';
 import { filterHiddenLocalCluster, filterOnlyKubernetesClusters, paginationFilterClusters } from '@shell/utils/cluster';
 import PaginationWrapper from '@shell/utils/pagination-wrapper';
@@ -202,9 +203,9 @@
     this.clustersOthersWrapper = new PaginationWrapper({
       $store,
       id:       'tlm-unpinned-clusters',
-      onChange: () => {
+      onChange: async() => {
         if (this.args) {
-          this.update(this.args);
+          await this.update(this.args);
         }
       },
       enabledFor: {
@@ -220,9 +221,9 @@
     this.provClusterWrapper = new PaginationWrapper({
       $store,
       id:       'tlm-prov-clusters',
-      onChange: () => {
+      onChange: async() => {
         if (this.args) {
-          this.update(this.args);
+          await this.update(this.args);
         }
       },
       enabledFor: {

shell/detail/provisioning.cattle.io.cluster.vue

@@ -89,6 +89,7 @@ export default {
   },
 
   async fetch() {
+    await this.$store.dispatch(`management/find`, { type: MANAGEMENT.CLUSTER, id: this.value.mgmtClusterId });
     await this.value.waitForProvisioner();
 
     // Support for the 'provisioner' extension

shell/plugins/dashboard-store/actions.js

@@ -497,7 +497,8 @@ export default {
       result: {
         count:     out.count,
         pages:     out.pages || Math.ceil(out.count / (opt.pagination.pageSize || Number.MAX_SAFE_INTEGER)),
-        timestamp: new Date().getTime()
+        timestamp: new Date().getTime(),
+        revision:  out.revision,
       }
     } : undefined;
 

shell/plugins/steve/index.js

@@ -17,16 +17,17 @@ export function SteveFactory(namespace, baseUrl) {
     state() {
       return {
         ...coreStoreState(namespace, baseUrl),
-        socket:           null,
-        queue:            [], // For change event coalescing
-        wantSocket:       false,
-        debugSocket:      false,
-        allowStreaming:   true,
-        pendingFrames:    [],
-        deferredRequests: {},
-        started:          [],
-        inError:          {},
-        podsByNamespace:  {}, // Cache of pods by namespace
+        socket:            null,
+        queue:             [], // For change event coalescing
+        wantSocket:        false,
+        debugSocket:       false,
+        allowStreaming:    true,
+        pendingFrames:     [],
+        deferredRequests:  {},
+        started:           [],
+        inError:           {},
+        isSocketTransient: {},
+        podsByNamespace:   {}, // Cache of pods by namespace
       };
     },
 

shell/plugins/steve/resourceWatcher.js

@@ -40,10 +40,10 @@ export const WATCH_STATUSES = {
  * Create a unique key for a specific resource watch's params
  */
 export const keyForSubscribe = ({
-  resourceType, type, namespace, id, selector
+  resourceType, type, namespace, id, selector, mode
 } = {}) => {
   const keyMap = {
-    type: resourceType || type, namespace, id, selector
+    type: resourceType || type, namespace, id, selector, mode // TODO: RC merge conflict, does this break new stuff?
   };
 
   return Object.entries(keyMap)

shell/plugins/steve/steve-pagination-utils.ts

@@ -180,7 +180,7 @@ class StevePaginationUtils extends NamespaceProjectFilters {
       { field: 'spec.internal' },
       { field: 'spec.displayName' },
       { field: `status.provider` },
-      { field: `metadata.labels["${ CAPI_LAB_AND_ANO.PROVIDER }]` },
+      { field: `metadata.labels[${ CAPI_LAB_AND_ANO.PROVIDER }]` },
       { field: `status.connected` },
     ],
     [CONFIG_MAP]: [

shell/plugins/steve/subscribe.js

@@ -510,12 +510,12 @@ const sharedActions = {
    * @param {STEVE_WATCH_PARAMS} params
    */
   watch({
-    state, dispatch, getters, rootGetters
+    state, dispatch, getters, rootGetters, commit
   }, params) {
     state.debugSocket && console.info(`Watch Request [${ getters.storeName }]`, JSON.stringify(params)); // eslint-disable-line no-console
     let {
       // eslint-disable-next-line prefer-const
-      type, selector, id, revision, namespace, stop, force, mode
+      type, selector, id, revision, namespace, stop, force, mode, transient
     } = params;
 
     namespace = acceptOrRejectSocketMessage.subscribeNamespace(namespace);
@@ -560,14 +560,18 @@ const sharedActions = {
       return;
     }
 
+    if (transient) {
+      commit('setTransient', { msg: params, transient });
+    }
+
     if (!stop) {
       dispatch('unwatchIncompatible', messageMeta);
     }
 
     // Watch errors mean we make a http request to get latest revision (which is still missing) and try to re-watch with it...
     // etc
     if (typeof revision === 'undefined') {
-      revision = getters.nextResourceVersion(type, id);
+      revision = getters.nextResourceVersion(type, id, params);
     }
 
     const msg = { resourceType: type };
@@ -648,6 +652,7 @@ const sharedActions = {
           dispatch('watch', obj); // Ask the backend to stop watching the type
           // Make sure anything in the pending queue for the type is removed, since we've now removed the type
           commit('clearFromQueue', type);
+          commit('setTransient', { obj, transient: false });
         }
         // Ensure anything pinging in the background is stopped
         backOff.resetPrefix(getters.backOffId(obj));
@@ -667,10 +672,20 @@ const sharedActions = {
 
   /**
    * Unwatch watches that are incompatible with the new type
+   *
+   * This is mainly to prevent the cache being polluted with resources that aren't compatible with it's aim
+   *
+   * For instance if the store contains a page then we don't want to receive updates for watches on specific other resources
+   *
    */
   unwatchIncompatible({
     state, dispatch, getters, commit
   }, messageMeta) {
+    // If the watch is transient (aka changes won't ever hit cache) we don't need to continue
+    if (getters.isTransient(messageMeta)) {
+      return;
+    }
+
     // Step 1 - Clear incompatible watches that have STARTED
     const watchesOfType = getters.watchesOfType(messageMeta.type);
 
@@ -902,7 +917,7 @@ const defaultActions = {
         const listener = listeners[STEVE_WATCH_MODE.RESOURCE_CHANGES].find((sl) => equivalentWatch(sl.params, params));
 
         if (listener) {
-          Object.values(listener.callbacks).forEach((cb) => cb());
+          Object.values(listener.callbacks).forEach((listenerCb) => listenerCb({ forceWatch: opt.forceWatch }));
         }
       } else {
         have = getters['all'](resourceType).slice();
@@ -1087,7 +1102,8 @@ const defaultActions = {
     state.started.filter((entry) => {
       if (
         entry.type === newWatch.type &&
-        entry.namespace !== newWatch.namespace
+        entry.namespace !== newWatch.namespace &&
+        entry.mode !== newWatch.mode
       ) {
         return true;
       }
@@ -1309,6 +1325,12 @@ const defaultMutations = {
     }
   },
 
+  setTransient(state, { msg, transient }) {
+    const key = keyForSubscribe(msg);
+
+    state.isSocketTransient[key] = transient;
+  },
+
   setInError(state, { msg, reason }) {
     const key = keyForSubscribe(msg);
 
@@ -1379,6 +1401,10 @@ const defaultGetters = {
     return state.inError[keyForSubscribe(obj)]?.reason;
   },
 
+  isTransient: (state) => (obj) => {
+    return state.isSocketTransient[keyForSubscribe(obj)];
+  },
+
   watchesOfType: (state) => (type) => {
     return state.started.filter((entry) => type === (entry.resourceType || entry.type));
   },
@@ -1402,7 +1428,13 @@ const defaultGetters = {
    *
    * Returns string, non-zero number or null
    */
-  nextResourceVersion: (state, getters) => (type, id) => {
+  nextResourceVersion: (state, getters) => (type, id, obj) => {
+    if (getters.isTransient(obj)) {
+      // We haven't stored the results for this type in the cache, don't look for revision there
+      // Instead do best effort - a low / non-existent reference which will force a re-fetch
+      return '0';
+    }
+
     type = normalizeType(type);
     let revision = 0;
 

shell/types/store/pagination.types.ts

@@ -451,6 +451,7 @@ export interface StorePaginationResult {
    * The last time the resource was updated. Used to assist list watching for changes
    */
   timestamp: number,
+  revision: string,
 }
 
 export interface StorePaginationRequest {

shell/types/store/subscribe.types.ts

@@ -25,10 +25,12 @@ export interface STEVE_WATCH_PARAMS {
   namespace?: string,
   stop?: boolean,
   force?: boolean,
-  mode?: STEVE_WATCH_MODE
+  mode?: STEVE_WATCH_MODE,
+  transient?: boolean,
 }
 
-export type STEVE_WATCH_EVENT_LISTENER_CALLBACK = () => void
+export type STEVE_WATCH_EVENT_LISTENER_ARGS = { forceWatch: boolean }
+export type STEVE_WATCH_EVENT_LISTENER_CALLBACK = (args: STEVE_WATCH_EVENT_LISTENER_ARGS) => void
 export interface STEVE_WATCH_EVENT_LISTENER {
   params: STEVE_WATCH_PARAMS,
   callbacks: { [id: string]: STEVE_WATCH_EVENT_LISTENER_CALLBACK},

shell/utils/pagination-wrapper.ts

@@ -3,7 +3,8 @@ import { PaginationArgs, PaginationResourceContext } from '@shell/types/store/pa
 import { VuexStore } from '@shell/types/store/vuex';
 import { ActionFindPageArgs, ActionFindPageTransientResult } from '@shell/types/store/dashboard-store.types';
 import {
-  STEVE_WATCH_EVENT_LISTENER_CALLBACK, STEVE_UNWATCH_EVENT_PARAMS, STEVE_WATCH_EVENT, STEVE_WATCH_EVENT_PARAMS, STEVE_WATCH_EVENT_PARAMS_COMMON, STEVE_WATCH_MODE
+  STEVE_WATCH_EVENT_LISTENER_CALLBACK, STEVE_UNWATCH_EVENT_PARAMS, STEVE_WATCH_EVENT, STEVE_WATCH_EVENT_PARAMS, STEVE_WATCH_EVENT_PARAMS_COMMON, STEVE_WATCH_MODE,
+  STEVE_WATCH_EVENT_LISTENER_ARGS
 } from '@shell/types/store/subscribe.types';
 import { Reactive, reactive } from 'vue';
 
@@ -20,7 +21,7 @@ interface Args {
   /**
    * Callback called when the resource is changed (notified by socket)
    */
-  onChange?: () => void,
+  onChange?: () => Promise<any>,
 
   formatResponse?: {
     /**
@@ -66,7 +67,16 @@ class PaginationWrapper<T extends object> {
     this.$store = $store;
     this.id = id;
     this.enabledFor = enabledFor;
-    this.onChange = onChange;
+    this.onChange = onChange ? async(args: STEVE_WATCH_EVENT_LISTENER_ARGS) => {
+      onChange().then(() => {
+        if (args.forceWatch) {
+          this.$store.dispatch(`${ this.enabledFor.store }/watch`, {
+            ...this.steveWatchParams?.params,
+            force: true,
+          });
+        }
+      });
+    } : undefined;
     this.classify = formatResponse?.classify || false;
     this.reactive = formatResponse?.reactive || false;
 
@@ -90,17 +100,23 @@ class PaginationWrapper<T extends object> {
     const out: ActionFindPageTransientResult<T> = await this.$store.dispatch(`${ this.enabledFor.store }/findPage`, { opt, type: this.enabledFor.resource?.id });
 
     // Watch
-    if (this.onChange && !this.steveWatchParams) {
-      this.steveWatchParams = {
-        event:  STEVE_WATCH_EVENT.CHANGES,
-        id:     this.id,
-        params: {
-          type: this.enabledFor.resource?.id as string,
-          mode: STEVE_WATCH_MODE.RESOURCE_CHANGES,
-        }
-      };
-
-      this.watch();
+    if (this.onChange) {
+      if (this.steveWatchParams) {
+        this.steveWatchParams.params.revision = out.pagination.result.revision;
+      } else {
+        this.steveWatchParams = {
+          event:  STEVE_WATCH_EVENT.CHANGES,
+          id:     this.id,
+          params: {
+            type:      this.enabledFor.resource?.id as string,
+            mode:      STEVE_WATCH_MODE.RESOURCE_CHANGES,
+            revision:  out.pagination.result.revision,
+            transient: true,
+          }
+        };
+
+        this.watch();
+      }
     }
 
     // Convert Response
@@ -126,7 +142,7 @@ class PaginationWrapper<T extends object> {
     }
     const watchParams: STEVE_WATCH_EVENT_PARAMS = {
       ...this.steveWatchParams,
-      callback: this.onChange as STEVE_WATCH_EVENT_LISTENER_CALLBACK, // we must have it by now
+      callback: this.onChange as STEVE_WATCH_EVENT_LISTENER_CALLBACK, // we must have onChange by now
     };
 
     await this.$store.dispatch(`${ this.enabledFor.store }/watchEvent`, watchParams);