From ca1b810c7bf53be88fdb99018624a310ccc7d298 Mon Sep 17 00:00:00 2001
From: Sean-McQ <37550899+Sean-McQ@users.noreply.github.com>
Date: Fri, 13 Jan 2023 16:03:00 -0500
Subject: [PATCH] Moves sockets into the advanced worker (#7760)

* Moves sockets into the advanced worker

* worker can die peacefully now, making switching between cluster work.

* Make waitFor generic, wire in to waitForTestFn

* General Changes
- Fixes for switching cluster
  - includes using common getPerformanceSetting
  - avoid new code to unsub before socket disconnect
- handle `watch` `stop` requests
- lots of TODO's (questions, work, checks, test, etc)
- use common

* Switch socket fixes
- isAdvancedWorker should only be true for cluster store
- advancedWorker to be wired in

* Fix socket id for cluster workers
- sockets use an incremented local var for id
- when we nuke the socket file within the worker this resets, so they all ahve id of 1
- work around this by applying the unix time

* Fix handling of new partical counts response
- seen in dex cluster explorer dashboard
- count cards would be removed when partial counts response received

* Make resourceWatcher the sole location for watch state
- getters canWatch, watchStarted now are worked around (they look at state in the UI thread)
  - we now don't call resource.stop or restart.start in subscription
- tidied up `forgetType`
- moved clearFromQueue from steve mutations into subscription mutations (better location)
- added and removed some TODOs
- fixed watch (stop handler should be higher up, include force watch handling)

* pushes the csrf value into worker and adds it to fetch request headers.

* refactors batchChanges to address ref concerns and be more performant

* Maintain schema reference whilst updating
- This change mutates input in a function, which is bad...
- but ensures the reference isn't broken, which is needed to maintain similar functionality as before

* Fix waitForTestFn
- Seen when creating or viewing clusters

* On unwatch ensure any pending watch requests are removed from the queue
- the probably would have been a problem if the worker wasn't nuked
- however as the codes there lets make it safe

Also added `trace` feature in advanced worker, will probably bring out to other places as well

* Fix navigation from cluster manager world to any cluster
- Ensure that we handle the case where the advanced worker was created but the resource watcher wasn't
- ... but fix case where this was happening (aka ensure that a blank cluster context is ignored)

* Tidy some TODOs

* Add perf settings page
- This will help test normal flow (when advanced worker is disabled)
- Note - setting is now in a bag. This may help us better support further settings (enable client side pagination, etc)
  ```
  advancedWorker: { enabled: false },
  ```

* FIX - Nav from cluster dashboard --> specific event --> cluster dashboard and events not re-subbed
- Ensure we block default handling of resource.start (keep state in resource watcher)

* Tidying up some TODOs

* Adds in a cache and uses it to validate SCHEMA messages before batching.

* Forgot to actually save CSRF to the resourceWatcher when instantiated.

* an empty resource in a batchChange to signal remove

* Move addSchemaIndexFields to and created removeSchemaIndexFields in new file
- this avoids bringing class files into the worker

* Fix disconnect/reconnect
- Remove `syncWatch` (do the watch/unwatch straight away)
- Test/Fix re-sub on reconnect
- Test/Fix growls on disconnect

* Tidying up some TODO's
- including clean of workerQueue on resource.stop (this is SUPER defensive)

* batchChanges will now handle aliases

* Fix pods list - WIP
- ensure podsByNamespace is updated on batchChange

TODO
- the final update to the pod is ignored
- removing a namespace cleans the cache correctly
- disabling advanced worker still works

* Fix pods list - fixes
- ensure podsByNamespace is updated on batchChange

Tested / Fixed
- the final update to the pod is ignored
- removing a namespace cleans the cache correctly
- disabling advanced worker still works

* Tidying TODOs

* Remove default same-origin header
- https://developer.mozilla.org/en-US/docs/Web/API/Request/credentials

* Fixed TODO description

* Refactor subscribe, make it clear which vuex feature relates to what

* Lots of Fixes
- batchChanges fixes
  - fix index is 0 issues (!/!!index)
  - only `set` if we have to
  - ensure we set the correct index after pushing to list
  - ensure map is updated after reducing list size with limit
- podsByNamespace fixes
  - ensure when ew replace... we don't use the same referenced object
- general service resource fixes
  - ensure service's pods list stays up to date with store

* Multiple improvements/fixes
- resourceCache - store the hash instead of the whole object. This means longer load time be reduces memory footprint
- resourceWatcher
  - don't re-sub on socket reconnect if watcher is in error
  - don't sub if watcher is in error
  - don't unwatch for 'failed to find schema' and 'too old' errors
    - this clears the error, we won't to keep it to ensure we don't watch
- Remove #5997 comments, follow on work #7917

* toggle debug, remap alias types, cleaned up comments and console

* Unit tests for batchChanges
Much more scope for some crazy content

* Logging tweaks
- disable logging by default
- initWorker comes in too late to affect initial trace, so just rely on the `debug` to toggle at runtime

Co-authored-by: Richard Cox <richard.cox@suse.com>
---
 shell/assets/translations/en-us.yaml          |   4 +
 shell/config/settings.ts                      |   3 +-
 shell/detail/service.vue                      |  11 +-
 shell/models/service.js                       |  27 +-
 .../pages/c/_cluster/settings/performance.vue |  16 +
 .../__tests__/mutations.spec.js               | 406 ++++++++++++++++++
 shell/plugins/dashboard-store/actions.js      |  34 +-
 shell/plugins/dashboard-store/mutations.js    | 136 +++++-
 .../plugins/dashboard-store/resource-class.js |  43 +-
 shell/plugins/steve/caches/resourceCache.js   |  60 +++
 shell/plugins/steve/getters.js                |  15 +-
 shell/plugins/steve/mutations.js              | 128 ++++--
 shell/plugins/steve/resourceWatcher.js        | 277 ++++++++++++
 shell/plugins/steve/schema.utils.js           |  25 ++
 shell/plugins/steve/subscribe.js              | 342 ++++++++++-----
 shell/plugins/steve/worker/index.js           |  17 +
 .../steve/worker/web-worker.advanced.js       | 302 +++++++++++++
 .../web-worker.basic.js}                      |  27 +-
 shell/utils/async.ts                          |  36 ++
 shell/utils/crypto/browserHashUtils.js        |  18 +
 shell/utils/socket.js                         |  34 +-
 21 files changed, 1685 insertions(+), 276 deletions(-)
 create mode 100644 shell/plugins/dashboard-store/__tests__/mutations.spec.js
 create mode 100644 shell/plugins/steve/caches/resourceCache.js
 create mode 100644 shell/plugins/steve/resourceWatcher.js
 create mode 100644 shell/plugins/steve/schema.utils.js
 create mode 100644 shell/plugins/steve/worker/index.js
 create mode 100644 shell/plugins/steve/worker/web-worker.advanced.js
 rename shell/plugins/steve/{web-worker.steve-sub-worker.js => worker/web-worker.basic.js} (85%)
 create mode 100644 shell/utils/async.ts

diff --git a/shell/assets/translations/en-us.yaml b/shell/assets/translations/en-us.yaml
index 3cae893b4eb..c8e7862dfeb 100644
--- a/shell/assets/translations/en-us.yaml
+++ b/shell/assets/translations/en-us.yaml
@@ -6813,6 +6813,10 @@ performance:
     count:
       inputLabel: Resource Threshold
       description: The threshold above which filtering by a namespace is required
+  advancedWorker:
+    label: Websocket Web Worker
+    description: Updates to resources pushed to the UI come via WebSocket and are handled in the UI thread. Enable this option to handle cluster WebSocket updates in a Web Worker in a separate thread. This should help the responsiveness of the UI in systems where resources change often.
+    checkboxLabel: Enable Advanced Websocket Web Worker
 
 banner:
   label: Fixed Banners
diff --git a/shell/config/settings.ts b/shell/config/settings.ts
index 73537e2603b..3e66b1d34e3 100644
--- a/shell/config/settings.ts
+++ b/shell/config/settings.ts
@@ -141,5 +141,6 @@ export const DEFAULT_PERF_SETTING = {
   forceNsFilter:                {
     enabled:   false,
     threshold: 1500,
-  }
+  },
+  advancedWorker: { enabled: false },
 };
diff --git a/shell/detail/service.vue b/shell/detail/service.vue
index df9207f08b2..93ea34876e0 100644
--- a/shell/detail/service.vue
+++ b/shell/detail/service.vue
@@ -11,7 +11,6 @@ import Tab from '@shell/components/Tabbed/Tab';
 import { CATTLE_PUBLIC_ENDPOINTS } from '@shell/config/labels-annotations';
 import { KEY, VALUE } from '@shell/config/table-headers';
 import { POD } from '@shell/config/types';
-import { allHash } from '@shell/utils/promise';
 import { findBy } from '@shell/utils/array';
 
 export default {
@@ -36,13 +35,7 @@ export default {
   },
 
   async fetch() {
-    const hash = { pods: this.value.pods() };
-
-    const res = await allHash(hash);
-
-    for (const k in res) {
-      this[k] = res[k];
-    }
+    await this.value.fetchPods();
   },
 
   data() {
@@ -159,7 +152,7 @@ export default {
       :weight="4"
     >
       <ResourceTable
-        :rows="pods"
+        :rows="value.pods"
         :headers="podTableHeaders"
         key-field="id"
         :table-actions="false"
diff --git a/shell/models/service.js b/shell/models/service.js
index 302bb3d3fd4..22a13c0e924 100644
--- a/shell/models/service.js
+++ b/shell/models/service.js
@@ -128,23 +128,24 @@ export default class extends SteveModel {
     return out;
   }
 
-  get pods() {
+  get podRelationship() {
     const { metadata:{ relationships = [] } } = this;
 
-    return async() => {
-      const podRelationship = (relationships || []).filter(relationship => relationship.toType === POD)[0];
-      let pods = [];
+    return (relationships || []).filter(relationship => relationship.toType === POD)[0];
+  }
 
-      if (podRelationship) {
-        pods = await this.$dispatch('cluster/findMatching', {
-          type:      POD,
-          selector:  podRelationship.selector,
-          namespace: this.namespace
-        }, { root: true });
-      }
+  async fetchPods() {
+    if (this.podRelationship) {
+      await this.$dispatch('cluster/findMatching', {
+        type:      POD,
+        selector:  this.podRelationship.selector,
+        namespace: this.namespace
+      }, { root: true });
+    }
+  }
 
-      return pods;
-    };
+  get pods() {
+    return this.podRelationship ? this.$getters.matching( POD, this.podRelationship.selector, this.namespace ) : [];
   }
 
   get serviceType() {
diff --git a/shell/pages/c/_cluster/settings/performance.vue b/shell/pages/c/_cluster/settings/performance.vue
index 12577c317fc..b7999136690 100644
--- a/shell/pages/c/_cluster/settings/performance.vue
+++ b/shell/pages/c/_cluster/settings/performance.vue
@@ -267,6 +267,22 @@ export default {
             />
           </div>
         </div>
+        <!-- Advanced Websocket Worker -->
+        <div class="mt-40">
+          <h2>{{ t('performance.advancedWorker.label') }}</h2>
+          <p>{{ t('performance.advancedWorker.description') }}</p>
+          <Banner
+            color="error"
+            label-key="performance.experimental"
+          />
+          <Checkbox
+            v-model="value.advancedWorker.enabled"
+            :mode="mode"
+            :label="t('performance.advancedWorker.checkboxLabel')"
+            class="mt-10 mb-20"
+            :primary="true"
+          />
+        </div>
       </div>
     </div>
     <template v-for="err in errors">
diff --git a/shell/plugins/dashboard-store/__tests__/mutations.spec.js b/shell/plugins/dashboard-store/__tests__/mutations.spec.js
new file mode 100644
index 00000000000..5688226e9f8
--- /dev/null
+++ b/shell/plugins/dashboard-store/__tests__/mutations.spec.js
@@ -0,0 +1,406 @@
+import { batchChanges } from '@shell/plugins/dashboard-store/mutations';
+import { POD, WORKLOAD_TYPES } from '@shell/config/types';
+import Resource from '@shell/plugins/dashboard-store/resource-class';
+
+describe('dashboard-store: mutations', () => {
+  const ctx = {
+    rootGetters: { 'type-map/optionsFor': type => ({}) },
+    getters:     {
+      classify:      resource => Resource,
+      cleanResource: (existing, resource) => resource
+    }
+  };
+
+  const create = type => ({
+    id: '1',
+    type
+  });
+
+  const createResource = (type, props = {}) => new Resource({
+    ...create(type),
+    ...props
+  });
+
+  const createPod = () => create(POD);
+
+  const createPodResource = (props = {}) => createResource(POD, props);
+
+  const createCache = props => ({
+    generation:    0,
+    haveAll:       false,
+    haveNamespace: undefined,
+    haveSelector:  {},
+    list:          [],
+    loadCounter:   0,
+    revision:      0,
+    map:           new Map(),
+    ...props
+  });
+
+  const createNoOp = () => {
+    const emptyState = { types: {} };
+
+    return {
+      params: [emptyState, {
+        ctx,
+        batch: { }
+      }],
+      expected: { state: emptyState }
+    };
+  };
+
+  const createRegister = () => {
+    return {
+      params: [{ types: {} }, {
+        ctx,
+        batch: { [POD]: {} }
+      }],
+      expected: { types: { [POD]: createCache({ generation: 1 }) } }
+    };
+  };
+
+  const createNewEntry = () => {
+    const pod = createPod();
+
+    return {
+      params: [
+        { types: {} },
+        {
+          ctx,
+          batch: { [POD]: { [pod.id]: pod } }
+        }
+      ],
+      expected: {
+        types: {
+          [POD]: createCache({
+            generation: 1,
+            list:       [new Resource(pod)],
+            map:        new Map([
+              [pod.id, new Resource(pod)]
+            ])
+          })
+        }
+      }
+    };
+  };
+
+  const createExisting = () => {
+    const existingPod = createPodResource();
+    const newPod = createPodResource({ change: true });
+
+    return {
+      params: [
+        {
+          types: {
+            [POD]: createCache({
+              list: [existingPod],
+              map:  new Map([
+                [existingPod.id, existingPod]
+              ])
+            })
+          }
+        },
+        {
+          ctx,
+          batch: { [POD]: { [newPod.id]: createPodResource({ change: true }) } }
+        }
+      ],
+      expected: {
+        types: {
+          [POD]: createCache({
+            generation: 1,
+            list:       [newPod],
+            map:        new Map([
+              [newPod.id, newPod]
+            ])
+          })
+
+        },
+      }
+    };
+  };
+
+  const createRemove = () => {
+    const existingPod = createPodResource();
+
+    return {
+      params: [
+        {
+          types: {
+            [POD]: createCache({
+              list: [existingPod],
+              map:  new Map([
+                [existingPod.id, existingPod]
+              ])
+            })
+          }
+        },
+        {
+          ctx,
+          batch: { [POD]: { [existingPod.id]: {} } }
+        }
+      ],
+      expected: {
+        types: {
+          [POD]: createCache({
+            generation: 1,
+            list:       [],
+            map:        new Map()
+          })
+        },
+      }
+    };
+  };
+
+  const createAddRemove = () => {
+    const existingPod1 = createPodResource({ id: '1' });
+    // const newPod1 = createPodResource({ id: '1', new: true });
+    const existingPod2 = createPodResource({ id: '2' });
+
+    return {
+      params: [
+        {
+          types: {
+            [POD]: createCache({
+              list: [existingPod2],
+              map:  new Map([
+                [existingPod2.id, existingPod2]
+              ])
+            })
+
+          }
+        },
+        {
+          ctx,
+          batch: {
+            [POD]: {
+              [existingPod1.id]: existingPod1,
+              [existingPod2.id]: {}
+            }
+          }
+        }
+      ],
+      expected: {
+        types: {
+          [POD]: createCache({
+            generation: 1,
+            list:       [existingPod1],
+            map:        new Map([
+              [existingPod1.id, existingPod1]
+            ])
+          })
+        },
+      }
+    };
+  };
+
+  const createRemoveAdd = () => {
+    const existingPod1 = createPodResource({ id: '1' });
+    // const newPod1 = createPodResource({ id: '1', new: true });
+    const existingPod2 = createPodResource({ id: '2' });
+
+    return {
+      params: [
+        {
+          types: {
+            [POD]: createCache({
+              list: [existingPod2],
+              map:  new Map([
+                [existingPod2.id, existingPod2]
+              ])
+            })
+          }
+        },
+        {
+          ctx,
+          batch: {
+            [POD]: {
+              [existingPod2.id]: {},
+              [existingPod1.id]: existingPod1,
+            }
+          }
+        }
+      ],
+      expected: {
+        types: {
+          [POD]: createCache({
+            generation: 1,
+            list:       [existingPod1],
+            map:        new Map([
+              [existingPod1.id, existingPod1]
+            ])
+          })
+        },
+      }
+    };
+  };
+
+  const createAddMultipleTypes = () => {
+    const pod = createPodResource({ id: '1' });
+    const deployment = createResource({ id: '2', type: WORKLOAD_TYPES.DEPLOYMENT });
+
+    return {
+      params: [
+        {
+          types: {
+            [POD]: createCache({
+              list: [],
+              map:  new Map()
+            }),
+            [WORKLOAD_TYPES.DEPLOYMENT]: createCache({
+              list: [],
+              map:  new Map()
+            })
+          }
+        },
+        {
+          ctx,
+          batch: {
+            [POD]:                        { [pod.id]: pod },
+            [ WORKLOAD_TYPES.DEPLOYMENT]: { [deployment.id]: deployment }
+          }
+        }
+      ],
+      expected: {
+        types: {
+          [POD]: createCache({
+            generation: 1,
+            list:       [pod],
+            map:        new Map([
+              [pod.id, pod]
+            ])
+          }),
+          [WORKLOAD_TYPES.DEPLOYMENT]: createCache({
+            generation: 1,
+            list:       [deployment],
+            map:        new Map([[deployment.id, deployment]])
+          })
+        },
+      }
+    };
+  };
+
+  const removeNonExisting = () => {
+    const pod1 = createPodResource({ id: '1' });
+    const pod2 = createPodResource({ id: '2' });
+
+    return {
+      params: [
+        {
+          types: {
+            [POD]: createCache({
+              list: [pod1],
+              map:  new Map([[pod1.id, pod1]])
+            }),
+          }
+        },
+        {
+          ctx,
+          batch: { [POD]: { [pod2.id]: {} } }
+        }
+      ],
+      expected: {
+        types: {
+          [POD]: createCache({
+            generation: 1,
+            list:       [pod1],
+            map:        new Map([[pod1.id, pod1]])
+          }),
+        },
+      }
+    };
+  };
+
+  const muchMuchChange = () => {
+    const pod1NoChange = createPodResource({ id: '1' });
+
+    const pod2Add = createPodResource({ id: '2' });
+    const pod3Remove = createPodResource({ id: '3' });
+    const pod4RemoveNoneExistant = createPodResource({ id: '4' });
+    const pod5Add = createPodResource({ id: '5' });
+    const pod6Original = createPodResource({ id: '6' });
+    const pod6WithChange = createPodResource({ id: '6', change: true }); // Cannot be same reference
+    const pod6WithChangeRef = createPodResource({ id: '6', change: true });
+
+    expect(pod6Original.id).toBe('6');
+    expect(pod6WithChange.id).toBe('6');
+    expect(pod6WithChangeRef.id).toBe('6');
+
+    return {
+      params: [
+        {
+          types: {
+            [POD]: createCache({
+              list: [pod1NoChange, pod3Remove, pod6Original],
+              map:  new Map([
+                [pod1NoChange.id, pod1NoChange],
+                [pod3Remove.id, pod3Remove],
+                [pod6Original.id, pod6Original],
+              ])
+            }),
+          }
+        },
+        {
+          ctx,
+          batch: {
+            [POD]: {
+              [pod2Add.id]:                pod2Add,
+              [pod3Remove.id]:             {},
+              [pod4RemoveNoneExistant.id]: {},
+              [pod5Add.id]:                pod5Add,
+              [pod6WithChangeRef.id]:      pod6WithChange // Cannot be same reference,
+            },
+          }
+        }
+      ],
+      expected: {
+        types: {
+          [POD]: createCache({
+            generation: 1,
+            list:       [pod1NoChange, pod6WithChangeRef, pod2Add, pod5Add], // Order important
+            map:        new Map([
+              [pod1NoChange.id, pod1NoChange],
+              [pod2Add.id, pod2Add],
+              [pod5Add.id, pod5Add],
+              [pod6WithChangeRef.id, pod6WithChangeRef],
+
+            ])
+          }),
+        },
+      }
+    };
+  };
+
+  describe('batchChanges', () => {
+    it.each([['No change', createNoOp()]])('%s', (_, run) => {
+      batchChanges(...run.params);
+
+      expect(run.params[0]).toStrictEqual(run.expected.state);
+    });
+
+    it.each([
+      ['Register the type', createRegister()],
+      ['Add a new pod', createNewEntry()],
+      ['Change a pod', createExisting()],
+      ['Remove a pod', createRemove()],
+      ['Create one pod, remove another', createAddRemove()],
+      ['Remove one pod, create another', createRemoveAdd()],
+      ['Add multiple types', createAddMultipleTypes()],
+      ['Remove a non-existing pod', removeNonExisting()],
+      ['Much much change', muchMuchChange()],
+    ])('%s', (_, run) => { // eslint-disable-line jest/no-identical-title
+      batchChanges(...run.params);
+      // Don't do straight up run.params[0] vs run.expected.... map doesn't get 'provided' so check fails
+      Object.entries(run.params[0].types).forEach(([type, cache]) => {
+        const { map: cacheMap, ...cacheState } = cache;
+        const { map: expectedMap, ...expected } = run.expected.types[type];
+
+        expect(cacheMap).toBeDefined();
+        expect(expectedMap).toBeDefined();
+
+        expect(cacheState).toStrictEqual(expected);
+        expect(cacheMap).toStrictEqual(expectedMap);
+      });
+    });
+  });
+});
diff --git a/shell/plugins/dashboard-store/actions.js b/shell/plugins/dashboard-store/actions.js
index 1b83dcfb9e7..78b25ef2f6b 100644
--- a/shell/plugins/dashboard-store/actions.js
+++ b/shell/plugins/dashboard-store/actions.js
@@ -6,6 +6,7 @@ import { createYaml } from '@shell/utils/create-yaml';
 import { classify } from '@shell/plugins/dashboard-store/classify';
 import { normalizeType } from './normalize';
 import garbageCollect from '@shell/utils/gc/gc';
+import { addSchemaIndexFields } from '@shell/plugins/steve/schema.utils';
 
 export const _ALL = 'all';
 export const _MERGE = 'merge';
@@ -48,10 +49,7 @@ export async function loadSchemas(ctx, watch = true) {
     res.data = res.concat(spoofedTypes);
   }
 
-  res.data.forEach((schema) => {
-    schema._id = normalizeType(schema.id);
-    schema._group = normalizeType(schema.attributes?.group);
-  });
+  res.data.forEach(addSchemaIndexFields);
 
   commit('loadAll', {
     ctx,
@@ -151,9 +149,7 @@ export default {
         namespace: opt.watchNamespace
       };
 
-      // if we are coming from a resource that wasn't watched
-      // but for which we have results already, just return the results but start watching it
-      if (opt.watch !== false && !getters.watchStarted(args)) {
+      if (opt.watch !== false ) {
         dispatch('watch', args);
       }
 
@@ -488,6 +484,15 @@ export default {
     });
   },
 
+  batchChanges(ctx, batch) {
+    const { commit } = ctx;
+
+    commit('batchChanges', {
+      ctx,
+      batch
+    });
+  },
+
   loadAll(ctx, { type, data }) {
     const { commit } = ctx;
 
@@ -521,20 +526,7 @@ export default {
   // Forget a type in the store
   // Remove all entries for that type and stop watching it
   forgetType({ commit, getters, dispatch }, type) {
-    const obj = {
-      type,
-      stop: true, // Stops the watch on a type
-    };
-
-    if (getters['schemaFor'](type) && getters['watchStarted'](obj)) {
-      // Set that we don't want to watch this type
-      // Otherwise, the dispatch to unwatch below will just cause a re-watch when we
-      // detect the stop message from the backend over the web socket
-      commit('setWatchStopped', obj);
-      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);
-    }
+    dispatch('unwatch', type);
 
     commit('forgetType', type);
   },
diff --git a/shell/plugins/dashboard-store/mutations.js b/shell/plugins/dashboard-store/mutations.js
index cd5d44ed059..55ab133c9bf 100644
--- a/shell/plugins/dashboard-store/mutations.js
+++ b/shell/plugins/dashboard-store/mutations.js
@@ -1,7 +1,8 @@
 import Vue from 'vue';
 import { addObject, addObjects, clear, removeObject } from '@shell/utils/array';
-import { SCHEMA } from '@shell/config/types';
-import { normalizeType } from '@shell/plugins/dashboard-store/normalize';
+import { SCHEMA, COUNT } from '@shell/config/types';
+import { normalizeType, keyFieldFor } from '@shell/plugins/dashboard-store/normalize';
+import { addSchemaIndexFields } from '@shell/plugins/steve/schema.utils';
 import { classify } from '@shell/plugins/dashboard-store/classify';
 import garbageCollect from '@shell/utils/gc/gc';
 
@@ -32,6 +33,24 @@ function registerType(state, type) {
   return cache;
 }
 
+export function replace(existing, data) {
+  for ( const k of Object.keys(existing) ) {
+    delete existing[k];
+  }
+
+  for ( const k of Object.keys(data) ) {
+    Vue.set(existing, k, data[k]);
+  }
+
+  return existing;
+}
+
+function replaceResource(existing, data, getters) {
+  data = getters.cleanResource(existing, data);
+
+  return replace(existing, data);
+}
+
 export function load(state, { data, ctx, existing }) {
   const { getters } = ctx;
   let type = normalizeType(data.type);
@@ -41,8 +60,7 @@ export function load(state, { data, ctx, existing }) {
 
   // Inject special fields for indexing schemas
   if ( type === SCHEMA ) {
-    data._id = normalizeType(data.id);
-    data._group = normalizeType(data.attributes?.group);
+    addSchemaIndexFields(data);
   }
 
   const id = data[keyField];
@@ -53,24 +71,10 @@ export function load(state, { data, ctx, existing }) {
 
   let entry;
 
-  function replace(existing, data) {
-    data = getters.cleanResource(existing, data);
-
-    for ( const k of Object.keys(existing) ) {
-      delete existing[k];
-    }
-
-    for ( const k of Object.keys(data) ) {
-      Vue.set(existing, k, data[k]);
-    }
-
-    return existing;
-  }
-
   if ( existing && !existing.id ) {
     // A specific proxy instance to used was passed in (for create -> save),
     // use it instead of making a new proxy
-    entry = replace(existing, data);
+    entry = replaceResource(existing, data, getters);
     addObject(cache.list, entry);
     cache.map.set(id, entry);
     // console.log('### Mutation added from existing proxy', type, id);
@@ -79,7 +83,7 @@ export function load(state, { data, ctx, existing }) {
 
     if ( entry ) {
       // There's already an entry in the store, update it
-      replace(entry, data);
+      replaceResource(entry, data, getters);
       // console.log('### Mutation Updated', type, id);
     } else {
       // There's no entry, make a new proxy
@@ -165,6 +169,93 @@ export function remove(state, obj, getters) {
   }
 }
 
+export function batchChanges(state, { ctx, batch }) {
+  const batchTypes = Object.keys(batch);
+  const combinedBatch = {};
+
+  batchTypes.forEach((batchType) => {
+    combinedBatch[batchType] = batch[batchType];
+    const typeOption = ctx.rootGetters['type-map/optionsFor'](batchType);
+
+    if (typeOption?.alias?.length > 0) {
+      const alias = typeOption?.alias || [];
+
+      alias.forEach((aliasType) => {
+        combinedBatch[aliasType] = {};
+        for (const [key, value] of Object.entries(batch[batchType])) {
+          combinedBatch[aliasType][key] = {
+            ...value,
+            type: aliasType
+          };
+        }
+      });
+    }
+  });
+
+  const combinedBatchTypes = Object.keys(combinedBatch);
+
+  combinedBatchTypes.forEach((type) => {
+    const normalizedType = normalizeType(type === 'counts' ? COUNT : type);
+    const keyField = keyFieldFor(normalizedType);
+    const typeCache = registerType(state, normalizedType);
+
+    // making a map for every resource's location in the list is gonna ensure we only have to loop through the big list once.
+    const typeCacheIndexMap = {};
+
+    typeCache.list.forEach((resource, index) => {
+      typeCacheIndexMap[resource[keyField]] = index;
+    });
+
+    const removeAtIndexes = [];
+
+    // looping through the batch, executing changes, deferring creates and removes since they change the array length
+    Object.keys(batch[normalizedType]).forEach((id) => {
+      const index = typeCacheIndexMap[id];
+      const resource = batch[normalizedType][id];
+
+      // an empty resource passed into batch changes is how we'll signal which ones to delete
+      if (Object.keys(resource).length === 0 && index !== undefined) {
+        typeCache.map.delete(id);
+        removeAtIndexes.push(index);
+      } else if (Object.keys(resource).length === 0) {
+        // No op. We're removing it... but we don't have it in the cache
+      } else {
+        if (normalizedType === SCHEMA) {
+          addSchemaIndexFields(resource);
+        }
+        const classyResource = classify(ctx, resource);
+
+        if (index === undefined) {
+          typeCache.list.push(classyResource);
+          typeCache.map.set(id, classyResource);
+
+          typeCacheIndexMap[classyResource[keyField]] = typeCache.list.length - 1;
+        } else {
+          replaceResource(typeCache.list[index], resource, ctx.getters);
+        }
+      }
+    });
+
+    // looping through the removeAtIndexes, making sure to offset by iteration so the array changing doesn't mess us up
+    removeAtIndexes.sort().forEach((cacheIndex, loopIndex) => {
+      typeCache.list.splice(cacheIndex - loopIndex, 1);
+    });
+
+    const opts = ctx.rootGetters[`type-map/optionsFor`](type);
+    const limit = opts?.limit;
+
+    // If there is a limit to the number of resources we can store for this type then
+    // remove the first one to keep the list size to that limit
+    if (limit && typeCache.list.length > limit) {
+      const rm = typeCache.list.shift();
+
+      typeCache.map.delete(rm.id);
+    }
+
+    typeCache.generation++;
+  });
+}
+
 export function loadAll(state, {
   type,
   data,
@@ -242,6 +333,11 @@ export default {
 
   loadAll,
 
+  /**
+   * Handles changes (add, update, remove) to multiple resources for multiple types
+   */
+  batchChanges,
+
   loadMerge(state, { type, data: allLatest, ctx }) {
     const { commit, getters } = ctx;
     // const allLatest = await dispatch('findAll', { type, opt: { force: true, load, _NONE } });
diff --git a/shell/plugins/dashboard-store/resource-class.js b/shell/plugins/dashboard-store/resource-class.js
index 5452df1e8c1..22468a0c3c0 100644
--- a/shell/plugins/dashboard-store/resource-class.js
+++ b/shell/plugins/dashboard-store/resource-class.js
@@ -25,6 +25,7 @@ import {
   validateLength,
 } from '@shell/utils/validators';
 import formRulesGenerator from '@shell/utils/validators/formRules/index';
+import { waitFor } from '@shell/utils/async';
 import jsyaml from 'js-yaml';
 import compact from 'lodash/compact';
 import forIn from 'lodash/forIn';
@@ -752,41 +753,7 @@ export default class Resource {
   // ------------------------------------------------------------------
 
   waitForTestFn(fn, msg, timeoutMs, intervalMs) {
-    console.log('Starting wait for', msg); // eslint-disable-line no-console
-
-    if ( !timeoutMs ) {
-      timeoutMs = DEFAULT_WAIT_TMIMEOUT;
-    }
-
-    if ( !intervalMs ) {
-      intervalMs = DEFAULT_WAIT_INTERVAL;
-    }
-
-    return new Promise((resolve, reject) => {
-      // Do a first check immediately
-      if ( fn.apply(this) ) {
-        console.log('Wait for', msg, 'done immediately'); // eslint-disable-line no-console
-        resolve(this);
-      }
-
-      const timeout = setTimeout(() => {
-        console.log('Wait for', msg, 'timed out'); // eslint-disable-line no-console
-        clearInterval(interval);
-        clearTimeout(timeout);
-        reject(new Error(`Failed waiting for: ${ msg }`));
-      }, timeoutMs);
-
-      const interval = setInterval(() => {
-        if ( fn.apply(this) ) {
-          console.log('Wait for', msg, 'done'); // eslint-disable-line no-console
-          clearInterval(interval);
-          clearTimeout(timeout);
-          resolve(this);
-        } else {
-          console.log('Wait for', msg, 'not done yet'); // eslint-disable-line no-console
-        }
-      }, intervalMs);
-    });
+    return waitFor(() => fn.apply(this), msg, timeoutMs || DEFAULT_WAIT_TMIMEOUT, intervalMs || DEFAULT_WAIT_INTERVAL, true);
   }
 
   waitForState(state, timeout, interval) {
@@ -798,19 +765,19 @@ export default class Resource {
   waitForTransition() {
     return this.waitForTestFn(() => {
       return !this.transitioning;
-    }, 'transition completion');
+    }, 'transition completion', undefined, undefined);
   }
 
   waitForAction(name) {
     return this.waitForTestFn(() => {
       return this.hasAction(name);
-    }, `action=${ name }`);
+    }, `action=${ name }`, undefined, undefined);
   }
 
   waitForLink(name) {
     return this.waitForTestFn(() => {
       return this.hasLink(name);
-    }, `link=${ name }`);
+    }, `link=${ name }`, undefined, undefined);
   }
 
   hasCondition(condition) {
diff --git a/shell/plugins/steve/caches/resourceCache.js b/shell/plugins/steve/caches/resourceCache.js
new file mode 100644
index 00000000000..3f2cd8d1145
--- /dev/null
+++ b/shell/plugins/steve/caches/resourceCache.js
@@ -0,0 +1,60 @@
+import { SCHEMA, COUNT } from '@shell/config/types';
+import { keyFieldFor, normalizeType } from '@shell/plugins/dashboard-store/normalize';
+import { addSchemaIndexFields } from '@shell/plugins/steve/schema.utils';
+import { hashObj } from '@shell/utils/crypto/browserHashUtils';
+
+export default class ResourceCache {
+  resources = {};
+  type;
+  keyField;
+  constructor(type) {
+    this.type = normalizeType(type === 'counts' ? COUNT : type);
+    this.keyField = keyFieldFor(this.type);
+  }
+
+  load(collection = []) {
+    // console.time('startSchemaLoads!!!');
+    for (let i = 0; i < collection.length; i++) {
+      const resource = collection[i];
+
+      if ( this.type === SCHEMA ) {
+        addSchemaIndexFields(resource);
+      }
+      const id = resource[this.keyField];
+
+      // Store the hash instead of the whole object. This means longer load time be reduces memory footprint
+      // Perf Note: 3.328125 ms to load ~2500 schemas as objects into cache
+      // Perf Note: 67.450927734375 ms to load ~2500 schemas as hashes into cache
+      this.resources[id] = hashObj(resource);
+    }
+    // console.timeEnd('startSchemaLoads!!!');
+
+    return this;
+  }
+
+  change(resource, callback) {
+    if ( this.type === SCHEMA ) {
+      addSchemaIndexFields(resource);
+    }
+
+    const existingResourceHash = this.resources[resource[this.keyField]] || {};
+    const newResourceHash = hashObj(resource);
+
+    if (existingResourceHash !== newResourceHash) {
+      this.resources[resource[this.keyField]] = resource;
+      callback();
+    }
+  }
+
+  create(resource, callback) {
+    // ToDo: the logic for create is identical to change in these caches but the worker doesn't know that
+    this.change(resource, callback);
+  }
+
+  remove(id, callback) {
+    if (this.resources[id]) {
+      delete this.resources[id];
+      callback();
+    }
+  }
+}
diff --git a/shell/plugins/steve/getters.js b/shell/plugins/steve/getters.js
index 4c5ae83c20c..2c4908fad54 100644
--- a/shell/plugins/steve/getters.js
+++ b/shell/plugins/steve/getters.js
@@ -7,6 +7,7 @@ import SteveModel from './steve-class';
 import HybridModel, { cleanHybridResources } from './hybrid-class';
 import NormanModel from './norman-class';
 import { urlFor } from '@shell/plugins/dashboard-store/getters';
+import { normalizeType } from '@shell/plugins/dashboard-store/normalize';
 
 export const STEVE_MODEL_TYPES = {
   NORMAN:  'norman',
@@ -150,6 +151,18 @@ export default {
 
   gcIgnoreTypes: () => {
     return GC_IGNORE_TYPES;
-  }
+  },
+
+  currentGeneration: state => (type) => {
+    type = normalizeType(type);
+
+    const cache = state.types[type];
+
+    if ( !cache ) {
+      return null;
+    }
+
+    return cache.generation;
+  },
 
 };
diff --git a/shell/plugins/steve/mutations.js b/shell/plugins/steve/mutations.js
index ca8ae9c6ec2..fc5a429cbf0 100644
--- a/shell/plugins/steve/mutations.js
+++ b/shell/plugins/steve/mutations.js
@@ -1,15 +1,18 @@
-import { addObject, removeObject } from '@shell/utils/array';
+import { addObject } from '@shell/utils/array';
 import { NAMESPACE, POD, SCHEMA } from '@shell/config/types';
 import {
   forgetType,
   resetStore,
   loadAll,
   load,
-  remove
+  remove,
+  batchChanges,
+  replace
 } from '@shell/plugins/dashboard-store/mutations';
-import { keyForSubscribe } from '@shell/plugins/steve/subscribe';
+import { keyForSubscribe } from '@shell/plugins/steve/resourceWatcher';
 import { perfLoadAll } from '@shell/plugins/steve/performanceTesting';
 import Vue from 'vue';
+import { classify } from '@shell/plugins/dashboard-store/classify';
 
 function registerNamespace(state, namespace) {
   let cache = state.podsByNamespace[namespace];
@@ -26,7 +29,93 @@ function registerNamespace(state, namespace) {
   return cache;
 }
 
+/**
+ * update the podsByNamespace cache with new or changed pods
+ */
+function updatePodsByNamespaceCache(state, ctx, pods, loadAll) {
+  if (loadAll) {
+    // Clear the entire cache - this is a fresh load
+    Object.keys(state.podsByNamespace).forEach((ns) => {
+      delete state.podsByNamespace[ns];
+    });
+  }
+
+  // Go through all of the pods and populate cache by namespace
+  pods.forEach((entry) => {
+    const classyResource = state.types[POD].map.get(entry.id) || classify(ctx, entry);
+
+    const cache = registerNamespace(state, classyResource.namespace); // Raw entry.namespace doesn't exist, so use classy
+    const existing = cache.map.get(entry.id);
+
+    if (existing) {
+      // CANNOT BE THE SAME REFERENCE
+      replace(existing, entry);
+    } else {
+      addObject(cache.list, classyResource);
+      cache.map.set(entry.id, classyResource);
+    }
+  });
+}
+
+/**
+ * clean the podsByNamespace cache of removed pods
+ */
+function cleanPodsByNamespaceCache(state, resource) {
+  if (resource && resource.type === POD) {
+    const cache = state.podsByNamespace[resource.namespace];
+
+    // Extra defensive check that the cache exists for the namespace being removed
+    if (cache) {
+      const inList = cache.list.findIndex(p => p.id === resource.id);
+
+      if ( inList >= 0 ) {
+        cache.list.splice(inList, 1);
+      }
+      cache.map.delete(resource.id);
+    }
+  } else if (resource && resource.type === NAMESPACE) {
+    // Namespace deleted
+    delete state.podsByNamespace[resource.id];
+  }
+}
+
 export default {
+  batchChanges(state, { ctx, batch }) {
+    batchChanges(state, { ctx, batch });
+
+    if (batch[POD]) {
+      const newAndChangedPods = Object.entries(batch[POD]).reduce((pods, [id, pod]) => {
+        if (pod.id) {
+          // resource.create and resource.change
+          pods.push(pod);// must NOT be same reference from store
+        } else {
+          // resource.remove (note - we've already lost the resource in the store, so pass through mocked one)
+          cleanPodsByNamespaceCache(state, {
+            id,
+            type:      POD,
+            namespace: id.substring(0, id.indexOf('/'))
+          });
+        }
+
+        return pods;
+      }, []);
+
+      updatePodsByNamespaceCache(state, ctx, newAndChangedPods, false);
+    }
+
+    if (batch[NAMESPACE]) {
+      Object.entries(batch[NAMESPACE]).forEach(([id, namespace]) => {
+        if (!namespace.id) {
+          // resource.remove (note - we've already lost the resource in the store, so pass through mocked one)
+          cleanPodsByNamespaceCache(state, {
+            id,
+            type: NAMESPACE,
+          });
+        }
+      });
+    }
+  },
+
   loadAll(state, {
     type,
     data,
@@ -45,18 +134,7 @@ export default {
 
     // If we loaded a set of pods, then update the podsByNamespace cache
     if (type === POD) {
-      // Clear the entire cache - this is a fresh load
-      Object.keys(state.podsByNamespace).forEach((ns) => {
-        delete state.podsByNamespace[ns];
-      });
-
-      // Go through all of the pods and populate cache by namespace
-      proxies.forEach((entry) => {
-        const cache = registerNamespace(state, entry.namespace);
-
-        addObject(cache.list, entry);
-        cache.map.set(entry.id, entry);
-      });
+      updatePodsByNamespaceCache(state, ctx, proxies, true);
     }
 
     // Notify the web worker of the initial load of schemas
@@ -84,13 +162,6 @@ export default {
     state.podsByNamespace = {};
   },
 
-  clearFromQueue(state, type) {
-    // Remove anything in the queue that is a resource update for the given type
-    state.queue = state.queue.filter((item) => {
-      return item.body?.type !== type;
-    });
-  },
-
   loadMulti(state, { data, ctx }) {
     for (const entry of data) {
       const resource = load(state, { data: entry, ctx });
@@ -107,17 +178,6 @@ export default {
   remove(state, obj) {
     remove(state, obj, this.getters);
 
-    if (obj && obj.type === POD) {
-      const cache = state.podsByNamespace[obj.namespace];
-
-      // Extra defensive check that the cache exists for the namespace being removed
-      if (cache) {
-        removeObject(cache.list, obj);
-        cache.map.delete(obj.id);
-      }
-    } else if (obj && obj.type === NAMESPACE) {
-      // Namespace deleted
-      delete state.podsByNamespace[obj.namespace];
-    }
+    cleanPodsByNamespaceCache(state, obj);
   }
 };
diff --git a/shell/plugins/steve/resourceWatcher.js b/shell/plugins/steve/resourceWatcher.js
new file mode 100644
index 00000000000..62b739771f1
--- /dev/null
+++ b/shell/plugins/steve/resourceWatcher.js
@@ -0,0 +1,277 @@
+/**
+ * Imports in a worker cannot include exports from the file invoking the worker or from files importing the invoking file.
+ */
+
+import Socket, {
+  NO_WATCH,
+  NO_SCHEMA,
+  EVENT_MESSAGE,
+  EVENT_CONNECTED,
+} from '@shell/utils/socket';
+import { addParam } from '@shell/utils/url';
+
+export const WATCH_STATUSES = {
+  /**
+   * watch has been asked for this resource but not request has not successfully been sent
+   */
+  WATCH_PENDING:    'pending',
+  /**
+   * requested but not confirmed by the socket yet
+   */
+  WATCH_REQUESTED:  'requested',
+  /**
+   * confirmed as active by the socket
+   */
+  WATCHING:         'watching',
+  /**
+   * temporarily stopped via message from the socket, a watch should immediately be triggered but the maintenance cycle will pick it up if that doesn't happen.
+   */
+  STOPPED:          'stopped',
+  /**
+   * stop has been asked for this resource, but request has not successfully been sent
+   */
+  REMOVE_PENDING:   'removed_pending',
+  /**
+   * stop request has been sent to the socket or it's been stopped by the socket itself and is now awaiting a resource.stop message
+   */
+  REMOVE_REQUESTED: 'removed_requested'
+};
+
+export const keyForSubscribe = ({
+  resourceType, type, namespace, id, selector
+} = {}) => {
+  return [(resourceType || type), namespace, id, selector] // each watch param in an array
+    .filter(param => !!param) // filter out all the empty ones // the filter makes these keys neater
+    .join('/'); // join into a string so we can use it as an object key
+};
+
+export const watchKeyFromMessage = (msg) => {
+  const {
+    resourceType,
+    namespace,
+    id,
+    selector
+  } = msg;
+
+  const watchObject = {
+    resourceType,
+    id,
+    namespace,
+    selector
+  };
+
+  return keyForSubscribe(watchObject);
+};
+
+const {
+  WATCH_PENDING, WATCH_REQUESTED, WATCHING, STOPPED, REMOVE_PENDING, REQUESTED_REMOVE
+} = WATCH_STATUSES;
+
+export default class ResourceWatcher extends Socket {
+  watches = {};
+  status = '';
+  debugWatcher = false;
+  csrf;
+
+  constructor(url, autoReconnect = true, frameTimeout = null, protocol = null, maxTries = null, csrf) {
+    super(url, autoReconnect, frameTimeout, protocol, maxTries, true);
+    this.baseUrl = self.location.origin + url.replace('subscribe', '');
+    this.csrf = csrf;
+
+    this.addEventListener(EVENT_CONNECTED, (e) => {
+      this.trace(EVENT_CONNECTED, ': processing previously requested or watched resources');
+
+      Object.values(this.watches).forEach((watch) => {
+        const { status, error } = watch;
+        const watchKey = keyForSubscribe(watch);
+
+        if ([WATCH_PENDING, WATCH_REQUESTED, WATCHING].includes(status) && !error) {
+          this.trace(EVENT_CONNECTED, ': re-watching previously required resource', watchKey, status);
+          this.watches[watchKey].status = WATCH_PENDING;
+          this.watch(watchKey);
+        } else if ([REMOVE_PENDING].includes(status)) {
+          this.trace(EVENT_CONNECTED, ': un-watching previously watched resource', watchKey, status);
+          this.watches[watchKey].status = REMOVE_PENDING;
+          this.unwatch(watchKey);
+        }
+      });
+    });
+  }
+
+  trace(...args) {
+    this.debugWatcher && console.info('Resource Watcher:', ...args); // eslint-disable-line no-console
+  }
+
+  setDebug(on) {
+    this.debugWatcher = !!on;
+  }
+
+  watchExists(watchKey) {
+    return !!this.watches?.[watchKey];
+  }
+
+  async watch(watchKey, providedResourceVersion, providedResourceVersionTime, providedKeyParts = {}, providedSkipResourceVersion) {
+    const {
+      resourceType: providedResourceType,
+      id: providedId,
+      namespace: providedNamespace,
+      selector: providedSelector
+    } = providedKeyParts;
+
+    this.trace('watch:', 'requested', watchKey);
+
+    if ([WATCH_REQUESTED, WATCHING].includes(this.watches?.[watchKey]?.status)) {
+      this.trace('watch:', 'already requested or watching, aborting', watchKey);
+
+      return;
+    }
+
+    if (this.watches?.[watchKey]?.error) {
+      this.trace('watch:', 'in error, aborting', watchKey);
+
+      return;
+    }
+
+    const resourceType = providedResourceType || this.watches?.[watchKey]?.resourceType;
+    const id = providedId || this.watches?.[watchKey]?.id;
+    const namespace = providedNamespace || this.watches?.[watchKey]?.namespace;
+    const selector = providedSelector || this.watches?.[watchKey]?.selector;
+    let skipResourceVersion = this.watches?.[watchKey]?.skipResourceVersion || providedSkipResourceVersion;
+
+    const watchObject = {
+      resourceType,
+      id,
+      namespace,
+      selector
+    };
+
+    let resourceVersionTime = providedResourceVersionTime || this.watches?.[watchKey]?.resourceVersionTime;
+    let resourceVersion = providedResourceVersion || this.watches?.[watchKey]?.resourceVersion;
+
+    if (!skipResourceVersion && (!resourceVersion || Date.now() - resourceVersionTime > 300000)) { // 300000ms is 5minutes
+      this.trace('watch:', 'revision update required', watchKey);
+
+      const resourceUrl = this.baseUrl + resourceType;
+      const limitedResourceUrl = addParam(resourceUrl, 'limit', 1);
+      const opt = {
+        method:  'get',
+        headers: { accept: 'application/json' },
+      };
+
+      if (this.csrf) {
+        opt.headers['x-api-csrf'] = this.csrf;
+      }
+
+      await fetch(limitedResourceUrl, opt)
+        .then((res) => {
+          this.watches[watchKey] = { ...watchObject };
+          if (!res.ok) {
+            this.watches[watchKey].error = res.json();
+            console.warn(`Resource error retrieving resourceVersion`, resourceType, ':', res.json()); // eslint-disable-line no-console
+          } else {
+            this.watches[watchKey].error = undefined;
+          }
+
+          return res.json();
+        })
+        .then((res) => {
+          if (res.revision) {
+            resourceVersionTime = Date.now();
+            resourceVersion = res.revision;
+          } else if (!this.watches[watchKey].error) {
+            // if there wasn't a revision in the response and there wasn't an error we wrote to the watch then the resource doesn't get a revision and we can skip it on subsequent rewatches
+            skipResourceVersion = true;
+          }
+        });
+      // When this fails we should re-fetch all resources (aka same as resyncWatch, or we actually call it). #7917
+      // This would match the old approach
+    }
+
+    const success = this.send(JSON.stringify({
+      ...watchObject,
+      resourceVersion: !skipResourceVersion ? resourceVersion : undefined
+    }));
+
+    this.watches[watchKey] = {
+      ...watchObject,
+      status: success ? WATCH_STATUSES.WATCH_REQUESTED : WATCH_STATUSES.WATCH_PENDING,
+      resourceVersion,
+      resourceVersionTime,
+      skipResourceVersion
+    };
+  }
+
+  unwatch(watchKey) {
+    const watch = this.watches?.[watchKey] || {};
+    const {
+      resourceType, id, namespace, selector
+    } = watch;
+    const watchObject = {
+      resourceType,
+      id,
+      namespace,
+      selector
+    };
+
+    if (resourceType && this.watches[watchKey].status !== REQUESTED_REMOVE) {
+      const success = this.send(JSON.stringify({
+        ...watchObject,
+        stop: true
+      }));
+
+      this.watches[watchKey].status = success ? REQUESTED_REMOVE : REMOVE_PENDING;
+    }
+  }
+
+  /**
+   * Handles message from Backend to UI
+   */
+  _onmessage(event) {
+    const {
+      name: eventName, resourceType, data: { type }, id, namespace, selector, data
+    } = JSON.parse(event.data);
+    const watchKey = keyForSubscribe({
+      resourceType,
+      type,
+      id,
+      namespace,
+      selector
+    });
+
+    if (eventName === 'resource.start' && this.watches?.[watchKey]?.status === WATCH_REQUESTED) {
+      this.watches[watchKey].status = WATCHING;
+    } else if (eventName === 'resource.stop' && this.watches?.[watchKey]) {
+      if (this.watches?.[watchKey]?.status === REQUESTED_REMOVE) {
+        delete this.watches[watchKey];
+      } else {
+        this.watches[watchKey].status = STOPPED;
+        delete this.watches[watchKey].resourceVersion;
+        delete this.watches[watchKey].resourceVersionTime;
+        this.watch(watchKey);
+        this.dispatchEvent(new CustomEvent(EVENT_MESSAGE, { detail: event }));
+      }
+    } else if (eventName === 'resource.error') {
+      const err = data?.error?.toLowerCase();
+
+      if ( this.watches[watchKey] && err.includes('watch not allowed') ) {
+        this.watches[watchKey].error = { type: resourceType, reason: NO_WATCH };
+      } else if ( this.watches[watchKey] && err.includes('failed to find schema') ) {
+        // This can happen when the cattle-cluster-agent goes down (redeploy deployment, kill pod, etc)
+        // The previous method was just to track the error and block any further attempts to watch (canWatch)
+        // This method means we can retry on the next findX (should be safe, unless there are other use cases...)
+
+        this.watches[watchKey].error = { type: resourceType, reason: NO_SCHEMA };
+      } else if ( err.includes('too old') ) {
+        // We don't actually know the gap between the requested revision and the oldest available revision.
+        // For this case we should re-fetch all resources (aka same as resyncWatch, or we actually call it). #7917
+        // This would match the old approach
+        delete this.watches[watchKey].resourceVersion;
+        delete this.watches[watchKey].resourceVersionTime;
+        delete this.watches[watchKey].skipResourceVersion;
+        this.watch(watchKey);
+      }
+    }
+
+    super._onmessage(event);
+  }
+}
diff --git a/shell/plugins/steve/schema.utils.js b/shell/plugins/steve/schema.utils.js
new file mode 100644
index 00000000000..595491955ca
--- /dev/null
+++ b/shell/plugins/steve/schema.utils.js
@@ -0,0 +1,25 @@
+import { normalizeType } from '@shell/plugins/dashboard-store/normalize';
+
+/**
+ * Inject special fields for indexing schemas
+ *
+ * Note
+ * This mutates input in a function, which is bad...
+ * but ensures the reference isn't broken, which is needed to maintain similar functionality as before
+ */
+export function addSchemaIndexFields(schema) {
+  schema._id = normalizeType(schema.id);
+  schema._group = normalizeType(schema.attributes?.group);
+}
+
+/**
+ * Remove special fields for indexing schemas
+ *
+ * Note
+ * This mutates input in a function, which is bad...
+ * but ensures the reference isn't broken, which is needed to maintain similar functionality as before
+ */
+export function removeSchemaIndexFields(schema) {
+  delete schema._id;
+  delete schema._group;
+}
diff --git a/shell/plugins/steve/subscribe.js b/shell/plugins/steve/subscribe.js
index 09da3d05f0e..b77f5d0315a 100644
--- a/shell/plugins/steve/subscribe.js
+++ b/shell/plugins/steve/subscribe.js
@@ -1,6 +1,16 @@
+/**
+ * Handles subscriptions to websockets which receive updates to resources
+ *
+ * Covers three use cases
+ * 1) Handles subscription within this file
+ * 2) Handles `cluster` subscriptions for some basic types in a web worker (SETTING.UI_PERFORMANCE advancedWorker = false)
+ * 2) Handles `cluster` subscriptions and optimisations in an advanced worker (SETTING.UI_PERFORMANCE advancedWorker = true)
+ */
+
 import { addObject, clear, removeObject } from '@shell/utils/array';
 import { get } from '@shell/utils/object';
 import { SCHEMA } from '@shell/config/types';
+import { CSRF } from '@shell/config/cookies';
 import { getPerformanceSetting } from '@shell/utils/settings';
 import Socket, {
   EVENT_CONNECTED,
@@ -8,25 +18,47 @@ import Socket, {
   EVENT_MESSAGE,
   //  EVENT_FRAME_TIMEOUT,
   EVENT_CONNECT_ERROR,
-  EVENT_DISCONNECT_ERROR
+  EVENT_DISCONNECT_ERROR,
+  NO_WATCH,
+  NO_SCHEMA,
 } from '@shell/utils/socket';
 import { normalizeType } from '@shell/plugins/dashboard-store/normalize';
 import day from 'dayjs';
 import { DATE_FORMAT, TIME_FORMAT } from '@shell/store/prefs';
 import { escapeHtml } from '@shell/utils/string';
+import { keyForSubscribe } from '@shell/plugins/steve/resourceWatcher';
+import { waitFor } from '@shell/utils/async';
 
 // eslint-disable-next-line
-import webworker from './web-worker.steve-sub-worker.js';
-
-export const NO_WATCH = 'NO_WATCH';
-export const NO_SCHEMA = 'NO_SCHEMA';
+import storeWorker from './worker/index.js';
+import { BLANK_CLUSTER } from '@shell/store/index.js';
 
 // minimum length of time a disconnect notification is shown
 const MINIMUM_TIME_NOTIFIED = 3000;
 
+const waitForManagement = (store) => {
+  const managementReady = () => store.state?.managementReady;
+
+  return waitFor(managementReady, 'Management');
+};
+
+const isAdvancedWorker = (ctx) => {
+  const { rootGetters, getters } = ctx;
+  const storeName = getters.storeName;
+  const clusterId = rootGetters.clusterId;
+
+  if (storeName !== 'cluster' || clusterId === BLANK_CLUSTER) {
+    return false;
+  }
+
+  const perfSetting = getPerformanceSetting(rootGetters);
+
+  return perfSetting?.advancedWorker.enabled;
+};
+
 // We only create a worker for the cluster store
-export function createWorker(store, ctx) {
-  const { getters } = ctx;
+export async function createWorker(store, ctx) {
+  const { getters, dispatch } = ctx;
   const storeName = getters.storeName;
 
   store.$workers = store.$workers || {};
@@ -35,24 +67,45 @@ export function createWorker(store, ctx) {
     return;
   }
 
+  await waitForManagement(store);
+  // getting perf setting in a separate constant here because it'll provide other values we'll want later.
+  const advancedWorker = isAdvancedWorker(ctx);
+
   const workerActions = {
     load: (resource) => {
       queueChange(ctx, resource, true, 'Change');
     },
     destroyWorker: () => {
       if (store.$workers) {
+        store.$workers[storeName].terminate();
         delete store.$workers[storeName];
       }
+    },
+    batchChanges: (batch) => {
+      dispatch('batchChanges', batch);
+    },
+    dispatch: (msg) => {
+      dispatch(`ws.${ msg.name }`, msg);
+    },
+    [EVENT_CONNECT_ERROR]: (e) => {
+      dispatch('error', e );
+    },
+    [EVENT_DISCONNECT_ERROR]: (e) => {
+      dispatch('error', e );
     }
   };
 
   if (!store.$workers[storeName]) {
-    const worker = new webworker();
+    const workerMode = advancedWorker ? 'advanced' : 'basic';
+    const worker = storeWorker(workerMode);
 
     store.$workers[storeName] = worker;
 
     worker.postMessage({ initWorker: { storeName } });
 
+    /**
+     * Covers message from Worker to UI thread
+     */
     store.$workers[storeName].onmessage = (e) => {
       /* on the off chance there's more than key in the message, we handle them in the order that they "keys" method provides which is
       // good enough for now considering that we never send more than one message action at a time right now */
@@ -65,14 +118,11 @@ export function createWorker(store, ctx) {
   }
 }
 
-export function keyForSubscribe({
-  resourceType, type, namespace, id, selector
-} = {}) {
-  return `${ resourceType || type || '' }/${ namespace || '' }/${ id || '' }/${ selector || '' }`;
-}
-
 export function equivalentWatch(a, b) {
-  if ( a.type !== b.type ) {
+  const aresourceType = a.resourceType || a.type;
+  const bresourceType = b.resourceType || b.type;
+
+  if ( aresourceType !== bresourceType ) {
     return false;
   }
 
@@ -136,12 +186,17 @@ function growlsDisabled(rootGetters) {
   return getPerformanceSetting(rootGetters)?.disableWebsocketNotification;
 }
 
-export const actions = {
-  subscribe(ctx, opt) {
+/**
+ * Actions that cover all cases (see file description)
+ */
+const sharedActions = {
+  async subscribe(ctx, opt) {
     const {
       state, commit, dispatch, getters, rootGetters
     } = ctx;
 
+    // ToDo: need to keep the worker up to date on CSRF cookie
+
     if (rootGetters['isSingleProduct']?.disableSteveSockets) {
       return;
     }
@@ -157,13 +212,28 @@ export const actions = {
     state.debugSocket && console.info(`Subscribe [${ getters.storeName }]`); // eslint-disable-line no-console
 
     const url = `${ state.config.baseUrl }/subscribe`;
+    const maxTries = growlsDisabled(rootGetters) ? null : 3;
+    const connectionMetadata = get(opt, 'metadata');
 
-    if ( socket ) {
+    if (isAdvancedWorker(ctx)) {
+      if (!this.$workers[getters.storeName]) {
+        await createWorker(this, ctx);
+      }
+
+      // if the worker is in advanced mode then it'll contain it's own socket which it calls a 'watcher'
+      this.$workers[getters.storeName].postMessage({
+        createWatcher: {
+          connectionMetadata,
+          url:  `${ state.config.baseUrl }/subscribe`,
+          csrf: this.$cookies.get(CSRF, { parseJSON: false }),
+          maxTries
+        }
+      });
+    } else if ( socket ) {
       socket.setAutoReconnect(true);
       socket.setUrl(url);
+      socket.connect(connectionMetadata);
     } else {
-      const maxTries = growlsDisabled(rootGetters) ? null : 3;
-
       socket = new Socket(`${ state.config.baseUrl }/subscribe`, true, null, null, maxTries);
 
       commit('setSocket', socket);
@@ -194,75 +264,28 @@ export const actions = {
           }
         }
       });
+      socket.connect(connectionMetadata);
     }
-
-    socket.connect(get(opt, 'metadata') );
   },
 
   unsubscribe({ commit, getters, state }) {
     const socket = state.socket;
-    const worker = (this.$workers || {})[getters.storeName];
 
     commit('setWantSocket', false);
+    const cleanupTasks = [];
+
+    const worker = (this.$workers || {})[getters.storeName];
 
     if (worker) {
       worker.postMessage({ destroyWorker: true }); // we're only passing the boolean here because the key needs to be something truthy to ensure it's passed on the object.
+      cleanupTasks.push(waitFor(() => !this.$workers[getters.storeName], 'Worker is destroyed'));
     }
 
     if ( socket ) {
-      return socket.disconnect();
-    }
-  },
-
-  async flush({
-    state, commit, dispatch, getters
-  }) {
-    const queue = state.queue;
-    let toLoad = [];
-
-    if ( !queue.length ) {
-      return;
+      cleanupTasks.push(socket.disconnect());
     }
 
-    const started = new Date().getTime();
-
-    state.queue = [];
-
-    state.debugSocket && console.debug(`Subscribe Flush [${ getters.storeName }]`, queue.length, 'items'); // eslint-disable-line no-console
-
-    for ( const { action, event, body } of queue ) {
-      if ( action === 'dispatch' && event === 'load' ) {
-        // Group loads into one loadMulti when possible
-        toLoad.push(body);
-      } else {
-        // When we hit a different kind of event, process all the previous loads, then the other event.
-        if ( toLoad.length ) {
-          await dispatch('loadMulti', toLoad);
-          toLoad = [];
-        }
-
-        if ( action === 'dispatch' ) {
-          await dispatch(event, body);
-        } else if ( action === 'commit' ) {
-          commit(event, body);
-        } else {
-          throw new Error('Invalid queued action');
-        }
-      }
-    }
-
-    // Process any remaining loads
-    if ( toLoad.length ) {
-      await dispatch('loadMulti', toLoad);
-    }
-
-    state.debugSocket && console.debug(`Subscribe Flush [${ getters.storeName }] finished`, (new Date().getTime()) - started, 'ms'); // eslint-disable-line no-console
-  },
-
-  rehydrateSubscribe({ state, dispatch }) {
-    if ( process.client && state.wantSocket && !state.socket ) {
-      dispatch('subscribe');
-    }
+    return Promise.all(cleanupTasks);
   },
 
   watch({
@@ -283,6 +306,7 @@ export const actions = {
       return;
     }
 
+    // If socket is in error don't try to watch.... unless we `force` it
     if ( !stop && !force && !getters.canWatch(params) ) {
       console.error(`Cannot Watch [${ getters.storeName }]`, JSON.stringify(params)); // eslint-disable-line no-console
 
@@ -323,9 +347,122 @@ export const actions = {
       msg.selector = selector;
     }
 
+    const worker = this.$workers[getters.storeName] || {};
+
+    if (worker.mode === 'advanced') {
+      if ( force ) {
+        msg.force = true;
+      }
+
+      worker.postMessage({ watch: msg });
+
+      return;
+    }
+
     return dispatch('send', msg);
   },
 
+  unwatch(ctx, type) {
+    const { commit, getters, dispatch } = ctx;
+
+    if (getters['schemaFor'](type)) {
+      const obj = {
+        type,
+        stop: true, // Stops the watch on a type
+      };
+
+      if (isAdvancedWorker(ctx)) {
+        dispatch('watch', obj); // Ask the backend to stop watching the type
+      } else if (getters['watchStarted'](obj)) {
+        // Set that we don't want to watch this type
+        // Otherwise, the dispatch to unwatch below will just cause a re-watch when we
+        // detect the stop message from the backend over the web socket
+        commit('setWatchStopped', obj);
+        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);
+      }
+    }
+  },
+
+  'ws.ping'({ getters, dispatch }, msg) {
+    if ( getters.storeName === 'management' ) {
+      const version = msg?.data?.version || null;
+
+      dispatch('updateServerVersion', version, { root: true });
+      console.info(`Ping [${ getters.storeName }] from ${ version || 'unknown version' }`); // eslint-disable-line no-console
+    }
+  },
+};
+
+/**
+ * Mutations that cover all cases (both subscriptions here and in advanced worker)
+ */
+const sharedMutations = {
+  debug(state, on, store) {
+    state.debugSocket = on !== false;
+    if (store && this.$workers[store]) {
+      this.$workers[store].postMessage({ toggleDebug: on !== false });
+    }
+  },
+};
+
+/**
+ * Actions that cover cases 1 & 2 (see file description)
+ */
+const defaultActions = {
+
+  async flush({
+    state, commit, dispatch, getters
+  }) {
+    const queue = state.queue;
+    let toLoad = [];
+
+    if ( !queue.length ) {
+      return;
+    }
+
+    const started = new Date().getTime();
+
+    state.queue = [];
+
+    state.debugSocket && console.debug(`Subscribe Flush [${ getters.storeName }]`, queue.length, 'items'); // eslint-disable-line no-console
+
+    for ( const { action, event, body } of queue ) {
+      if ( action === 'dispatch' && event === 'load' ) {
+        // Group loads into one loadMulti when possible
+        toLoad.push(body);
+      } else {
+        // When we hit a different kind of event, process all the previous loads, then the other event.
+        if ( toLoad.length ) {
+          await dispatch('loadMulti', toLoad);
+          toLoad = [];
+        }
+
+        if ( action === 'dispatch' ) {
+          await dispatch(event, body);
+        } else if ( action === 'commit' ) {
+          commit(event, body);
+        } else {
+          throw new Error('Invalid queued action');
+        }
+      }
+    }
+
+    // Process any remaining loads
+    if ( toLoad.length ) {
+      await dispatch('loadMulti', toLoad);
+    }
+
+    state.debugSocket && console.debug(`Subscribe Flush [${ getters.storeName }] finished`, (new Date().getTime()) - started, 'ms'); // eslint-disable-line no-console
+  },
+
+  rehydrateSubscribe({ state, dispatch }) {
+    if ( process.client && state.wantSocket && !state.socket ) {
+      dispatch('subscribe');
+    }
+  },
+
   reconnectWatches({
     state, getters, commit, dispatch
   }) {
@@ -535,15 +672,6 @@ export const actions = {
     }
   },
 
-  'ws.ping'({ getters, dispatch }, msg) {
-    if ( getters.storeName === 'management' ) {
-      const version = msg?.data?.version || null;
-
-      dispatch('updateServerVersion', version, { root: true });
-      console.info(`Ping [${ getters.storeName }] from ${ version || 'unknown version' }`); // eslint-disable-line no-console
-    }
-  },
-
   /**
    * Steve only event
    */
@@ -577,6 +705,7 @@ export const actions = {
    * Steve only seems to send out `resource.stop` messages for two reasons
    * - We have requested that the resource watch should be stopped and we receive this event as confirmation
    * - Steve tells us that the resource is no longer watched
+   *
    */
   'ws.resource.stop'({ getters, commit, dispatch }, msg) {
     const type = msg.resourceType;
@@ -591,8 +720,6 @@ export const actions = {
 
     // If we're trying to watch this event, attempt to re-watch
     if ( getters['schemaFor'](type) && getters['watchStarted'](obj) ) {
-      // Try reconnecting once
-
       commit('setWatchStopped', obj);
 
       // In summary, we need to re-watch but with a reliable `revision` (to avoid `too old` message kicking off a full re-fetch of all
@@ -712,7 +839,10 @@ export const actions = {
   },
 };
 
-export const mutations = {
+/**
+ * Mutations that cover cases 1 & 2 (see file description)
+ */
+const defaultMutations = {
   setSocket(state, socket) {
     state.socket = socket;
   },
@@ -761,21 +891,29 @@ export const mutations = {
     delete state.inError[key];
   },
 
-  debug(state, on) {
-    state.debugSocket = on !== false;
-  },
-
   resetSubscriptions(state) {
+    // Clear out socket state. This is only ever called from reset... which is always called after we `disconnect` above.
+    // This could probably be folded in to there
     clear(state.started);
     clear(state.pendingFrames);
     clear(state.queue);
     clearTimeout(state.queueTimer);
     state.deferredRequests = {};
     state.queueTimer = null;
-  }
+  },
+
+  clearFromQueue(state, type) {
+    // Remove anything in the queue that is a resource update for the given type
+    state.queue = state.queue.filter((item) => {
+      return item.body?.type !== type;
+    });
+  },
 };
 
-export const getters = {
+/**
+ * Getters that cover cases 1 & 2 (see file description)
+ */
+const defaultGetters = {
   canWatch: state => (obj) => {
     return !state.inError[keyForSubscribe(obj)];
   },
@@ -818,16 +956,16 @@ export const getters = {
 
     return null;
   },
+};
 
-  currentGeneration: state => (type) => {
-    type = normalizeType(type);
-
-    const cache = state.types[type];
-
-    if ( !cache ) {
-      return null;
-    }
+export const actions = {
+  ...sharedActions,
+  ...defaultActions,
+};
 
-    return cache.generation;
-  },
+export const mutations = {
+  ...sharedMutations,
+  ...defaultMutations,
 };
+
+export const getters = { ...defaultGetters };
diff --git a/shell/plugins/steve/worker/index.js b/shell/plugins/steve/worker/index.js
new file mode 100644
index 00000000000..25ca6723276
--- /dev/null
+++ b/shell/plugins/steve/worker/index.js
@@ -0,0 +1,17 @@
+// eslint-disable-next-line no-unused-vars
+import basicWorkerConstructor from '@shell/plugins/steve/worker/web-worker.basic.js';
+// eslint-disable-next-line no-unused-vars
+import advancedWorkerConstructor from '@shell/plugins/steve/worker/web-worker.advanced.js';
+
+export default function storeWorker(mode, options = {}, closures = {}) {
+  let worker;
+
+  if (mode === 'advanced') {
+    worker = new advancedWorkerConstructor();
+  } else {
+    worker = new basicWorkerConstructor();
+  }
+  worker.mode = mode;
+
+  return worker;
+}
diff --git a/shell/plugins/steve/worker/web-worker.advanced.js b/shell/plugins/steve/worker/web-worker.advanced.js
new file mode 100644
index 00000000000..1345a2e9fd8
--- /dev/null
+++ b/shell/plugins/steve/worker/web-worker.advanced.js
@@ -0,0 +1,302 @@
+/**
+ * Advanced Worker is enabled by performance setting
+ * relocates cluster resource sockets off the UI thread and into a webworker
+ */
+
+import { SCHEMA, COUNT } from '@shell/config/types';
+import ResourceWatcher, { watchKeyFromMessage } from '@shell/plugins/steve/resourceWatcher';
+import ResourceCache from '@shell/plugins/steve/caches/resourceCache';
+import { EVENT_MESSAGE, EVENT_CONNECT_ERROR, EVENT_DISCONNECT_ERROR } from '@shell/utils/socket';
+import { normalizeType, keyFieldFor } from '@shell/plugins/dashboard-store/normalize';
+import { addSchemaIndexFields } from '@shell/plugins/steve/schema.utils';
+
+const caches = {};
+
+const state = {
+  watcher:      undefined,
+  store:        '', // Store name
+  /**
+   * Store `watch`/`unwatch` events to process when the socket is created
+   */
+  workerQueue:  [],
+  batchChanges: {},
+  debugWorker:  false
+};
+
+const trace = (...args) => {
+  state.debugWorker && console.info('Advanced Worker:', ...args); // eslint-disable-line no-console
+};
+
+trace('created');
+
+const maintenanceInterval = setInterval(() => {
+  if (Object.keys(state.batchChanges).length) {
+    self.postMessage({ batchChanges: state.batchChanges });
+    state.batchChanges = {};
+  }
+}, 5000); // 5 seconds
+
+const makeResourceProps = (msg) => {
+  const { resourceType, data: { type }, data } = msg;
+  const rawType = resourceType || type;
+  const normalizedType = normalizeType(rawType === 'counts' ? COUNT : rawType);
+  const keyField = keyFieldFor(normalizedType);
+
+  if ( normalizedType === SCHEMA ) {
+    addSchemaIndexFields(data);
+  }
+
+  return {
+    type: normalizedType,
+    id:   data[keyField],
+    data
+  };
+};
+
+/**
+ * Pass the EVENT_CONNECT_ERROR / EVENT_DISCONNECT_ERROR back to the UI thread
+ */
+const handleConnectionError = (eventType, event, watcher) => {
+  trace('createWatcher', eventType, event);
+  self.postMessage({
+    [eventType]: {
+      type:       event.type,
+      detail:     event.detail,
+      srcElement: {
+        disconnectedAt: watcher.disconnectedAt,
+        url:            watcher.url,
+      }
+    }
+  });
+};
+
+/**
+ * Remove any pending messages related to this resource from the queue
+ */
+const removeFromWorkerQueue = (watchKey) => {
+  state.workerQueue = state.workerQueue.filter((workerMessage) => {
+    const [, msg] = Object.entries(workerMessage)[0];
+    const workerMessageWatchKey = watchKeyFromMessage(msg);
+
+    return watchKey !== workerMessageWatchKey;
+  });
+};
+
+/**
+ * These are things that we do when we get a message from the UI thread
+ */
+const workerActions = {
+  // ToDo: SM we'll make a generic loader for all resource types when we need it but it'll be pretty similar to this
+  loadSchemas: (collection) => {
+    if (!caches[SCHEMA]) {
+      caches[SCHEMA] = new ResourceCache(SCHEMA);
+    }
+    caches[SCHEMA].load(collection);
+  },
+  createWatcher: (metadata) => {
+    trace('createWatcher', metadata);
+
+    const {
+      connectionMetadata, maxTries, url, csrf
+    } = metadata;
+
+    if (!state.watcher) {
+      state.watcher = new ResourceWatcher(url, true, null, null, maxTries, csrf);
+
+      state.watcher.addEventListener(EVENT_MESSAGE, (e) => {
+        const event = e.detail;
+
+        if (event.data) {
+          const msg = JSON.parse(event.data);
+
+          if (msg.name) {
+            if (resourceWatcherActions[msg.name]) {
+              resourceWatcherActions[msg.name](msg);
+            } else {
+              resourceWatcherActions.dispatch(msg);
+            }
+          }
+        }
+      });
+
+      state.watcher.addEventListener(EVENT_CONNECT_ERROR, (e) => {
+        handleConnectionError(EVENT_CONNECT_ERROR, e, state.watcher);
+      });
+
+      state.watcher.addEventListener(EVENT_DISCONNECT_ERROR, (e) => {
+        handleConnectionError(EVENT_DISCONNECT_ERROR, e, state.watcher);
+      });
+
+      state.watcher.setDebug(state.debugWorker);
+
+      state.watcher.connect(connectionMetadata);
+
+      // Flush the workerQueue
+      while (state.workerQueue.length > 0) {
+        trace('createWatcher', 'flushing workerQueue', state.workerQueue);
+
+        const workerMessage = state.workerQueue.shift();
+        const [action, msg] = Object.entries(workerMessage)[0];
+
+        if (workerActions[action]) {
+          workerActions[action](msg);
+        } else {
+          console.warn('no associated action for:', action); // eslint-disable-line no-console
+        }
+      }
+    }
+  },
+  watch: (msg) => {
+    trace('watch', msg);
+
+    const watchKey = watchKeyFromMessage(msg);
+
+    if (msg.stop) {
+      workerActions.unwatch(watchKey);
+
+      return;
+    }
+
+    // If socket is in error don't try to watch.... unless we `force` it
+    if (!msg.force && !!state.watcher?.watches[watchKey]?.error) {
+      return;
+    }
+
+    if (!state.watcher) {
+      state.workerQueue.push({ watch: msg });
+
+      return;
+    }
+
+    const {
+      resourceType,
+      namespace,
+      id,
+      selector,
+      resourceVersion
+    } = msg;
+
+    const resourceVersionTime = resourceVersion ? Date.now() : undefined;
+    const skipResourceVersion = [SCHEMA, COUNT].includes(resourceType);
+
+    const watchObject = {
+      resourceType,
+      id,
+      namespace,
+      selector
+    };
+
+    state.watcher.watch(watchKey, resourceVersion, resourceVersionTime, watchObject, skipResourceVersion);
+  },
+  unwatch: (watchKey) => {
+    trace('unwatch', watchKey);
+
+    removeFromWorkerQueue(watchKey);
+
+    if (!state.watcher) {
+      return;
+    }
+
+    state.watcher.unwatch(watchKey);
+  },
+  initWorker: ({ storeName }) => {
+    trace('initWorker', storeName);
+
+    state.store = storeName;
+  },
+  destroyWorker: () => {
+    trace('destroyWorker');
+
+    clearInterval(maintenanceInterval);
+
+    function destroyWorkerComplete() {
+      delete self.onmessage;
+      self.postMessage({ destroyWorker: true }); // we're only passing the boolean here because the key needs to be something truthy to ensure it's passed on the object.
+    }
+
+    // disconnect takes a callback which we'll use to close the webworker
+    if (state.watcher) {
+      state.watcher?.disconnect().then(destroyWorkerComplete);
+    } else {
+      destroyWorkerComplete();
+    }
+  },
+
+  toggleDebug: ({ on }) => {
+    state.debugWorker = !!on;
+    state.watcher.setDebug(!!on);
+  },
+  updateBatch(type, id, change) {
+    if (!state.batchChanges[type]) {
+      state.batchChanges[type] = {};
+    }
+    state.batchChanges[type][id] = change;
+  }
+};
+
+/**
+ * These are things that we do when we get a message from the resourceWatcher
+ */
+const resourceWatcherActions = {
+  'resource.change': (msg) => {
+    const { type, id, data } = makeResourceProps(msg);
+
+    if (caches[type]) {
+      caches[type].change(data, () => workerActions.updateBatch(type, id, data));
+    } else {
+      workerActions.updateBatch(type, id, data);
+    }
+  },
+  // ToDo: SM create is functionally identical to change in the cache but the worker isn't supposed to know that hence the near-duplicate code
+  'resource.create': (msg) => {
+    const { type, id, data } = makeResourceProps(msg);
+
+    if (caches[type]) {
+      caches[type].create(data, () => workerActions.updateBatch(type, id, data));
+    } else {
+      workerActions.updateBatch(type, id, data);
+    }
+  },
+  'resource.start': (msg) => {
+    // State is handled in the resourceWatcher, no need to bubble out to UI thread
+  },
+  'resource.remove': (msg) => {
+    const { type, id } = makeResourceProps(msg);
+
+    if (caches[type]) {
+      caches[type].remove(id, () => workerActions.updateBatch(type, id, {}));
+    } else {
+      workerActions.updateBatch(type, id, {});
+    }
+  },
+  'resource.stop': (msg) => {
+    // State is handled in the resourceWatcher, no need to bubble out to UI thread
+    const watchKey = watchKeyFromMessage(msg);
+
+    removeFromWorkerQueue(watchKey);
+  },
+  'resource.error': (msg) => {
+    // State is handled in the resourceWatcher, no need to bubble out to UI thread
+    console.warn(`Resource error [${ state.store }]`, msg.resourceType, ':', msg.data.error); // eslint-disable-line no-console
+  },
+  dispatch: (msg) => {
+    self.postMessage({ dispatch: msg });
+  }
+};
+
+/**
+ * Covers message from UI Thread to Worker
+ */
+onmessage = (e) => {
+  /* on the off chance there's more than key in the message, we handle them in the order that they "keys" method provides which is
+  // good enough for now considering that we never send more than one message action at a time right now */
+  const messageActions = Object.keys(e?.data);
+
+  messageActions.forEach((action) => {
+    if (workerActions[action]) {
+      workerActions[action](e?.data[action]);
+    } else {
+      console.warn('no associated action for:', action); // eslint-disable-line no-console
+    }
+  });
+}; // bind everything to the worker's onmessage handler via the workerActions
diff --git a/shell/plugins/steve/web-worker.steve-sub-worker.js b/shell/plugins/steve/worker/web-worker.basic.js
similarity index 85%
rename from shell/plugins/steve/web-worker.steve-sub-worker.js
rename to shell/plugins/steve/worker/web-worker.basic.js
index 42f59b78035..13ca0d8937b 100644
--- a/shell/plugins/steve/web-worker.steve-sub-worker.js
+++ b/shell/plugins/steve/worker/web-worker.basic.js
@@ -1,4 +1,6 @@
 import { SCHEMA } from '@shell/config/types';
+import { hashObj } from '@shell/utils/crypto/browserHashUtils';
+import { removeSchemaIndexFields } from '@shell/plugins/steve/schema.utils';
 
 const SCHEMA_FLUSH_TIMEOUT = 2500;
 
@@ -9,25 +11,6 @@ const state = {
   schemas:    {} // Map of schema id to hash to track when a schema actually changes
 };
 
-// Quick, simple hash function
-function hash(str) {
-  let hash = 0;
-
-  for (let i = 0; i < str.length; i++) {
-    const char = str.charCodeAt(i);
-
-    hash = (hash << 5) - hash + char;
-    hash &= hash;
-  }
-
-  return new Uint32Array([hash])[0].toString(36);
-}
-
-// Quick, simple hash function to generate hash for an object
-function hashObj(obj) {
-  return hash(JSON.stringify(obj, null, 2));
-}
-
 function flush() {
   state.queue.forEach((schema) => {
     const hash = hashObj(schema);
@@ -81,9 +64,6 @@ const workerActions = {
     clearTimeout(state.flushTimer);
 
     self.postMessage({ destroyWorker: true }); // we're only passing the boolean here because the key needs to be something truthy to ensure it's passed on the object.
-
-    // Web worker global function to terminate the web worker
-    close();
   },
 
   // Called to load schema
@@ -91,8 +71,7 @@ const workerActions = {
     schemas.forEach((schema) => {
       // These properties are added to the object, but aren't on the raw object, so remove them
       // otherwise our comparison will show changes when there aren't any
-      delete schema._id;
-      delete schema._group;
+      removeSchemaIndexFields(schema);
 
       state.schemas[schema.id] = hashObj(schema);
     });
diff --git a/shell/utils/async.ts b/shell/utils/async.ts
new file mode 100644
index 00000000000..e83db4c0ecb
--- /dev/null
+++ b/shell/utils/async.ts
@@ -0,0 +1,36 @@
+export const waitFor = (testFn: Function, msg = '', timeoutMs = 3000000, intervalMs = 500, log = false) => {
+  gatedLog('Starting wait for', msg);
+
+  return new Promise((resolve, reject) => {
+    if (testFn()) {
+      gatedLog('Wait for', msg || 'unknown', 'done immediately');
+      resolve(this);
+    }
+    const timeout = setTimeout(() => {
+      gatedLog('Wait for', msg, 'timed out');
+      clearInterval(interval);
+      clearTimeout(timeout);
+      if (msg) {
+        reject(new Error(`Failed waiting for: ${ msg }`));
+      } else {
+        throw new Error(`waitFor timed out after ${ timeoutMs / 1000 } seconds`);
+      }
+    }, timeoutMs);
+    const interval = setInterval(() => {
+      if ( testFn() ) {
+        gatedLog('Wait for', msg, 'done');
+        clearInterval(interval);
+        clearTimeout(timeout);
+        resolve(this);
+      } else if (msg) {
+        gatedLog('Wait for', msg, 'not done yet');
+      }
+    }, intervalMs);
+  });
+
+  function gatedLog(...args: any[]) {
+    if (log) {
+      console.log(...args); // eslint-disable-line no-console
+    }
+  }
+};
diff --git a/shell/utils/crypto/browserHashUtils.js b/shell/utils/crypto/browserHashUtils.js
index 0c61faec442..a593b11352e 100644
--- a/shell/utils/crypto/browserHashUtils.js
+++ b/shell/utils/crypto/browserHashUtils.js
@@ -1,5 +1,23 @@
 import { Buffer } from 'buffer';
 
+function hashString(str) {
+  let hash = 0;
+
+  for (let i = 0; i < str.length; i++) {
+    const char = str.charCodeAt(i);
+
+    hash = (hash << 5) - hash + char;
+    hash &= hash;
+  }
+
+  return new Uint32Array([hash])[0].toString(36);
+}
+
+// Quick, simple hash function to generate hash for an object
+export function hashObj(obj) {
+  return hashString(JSON.stringify(obj, null, 2));
+}
+
 /**
  * @api private
  */
diff --git a/shell/utils/socket.js b/shell/utils/socket.js
index 7376042b8c4..05a502580db 100644
--- a/shell/utils/socket.js
+++ b/shell/utils/socket.js
@@ -10,8 +10,9 @@ const INSECURE = 'ws://';
 const SECURE = 'wss://';
 
 const STATE_DISCONNECTED = 'disconnected';
-const STATE_CONNECTING = 'connecting';
-const STATE_CONNECTED = 'connected';
+
+export const STATE_CONNECTING = 'connecting';
+export const STATE_CONNECTED = 'connected';
 const STATE_CLOSING = 'closing';
 const STATE_RECONNECTING = 'reconnecting';
 
@@ -23,6 +24,9 @@ export const EVENT_FRAME_TIMEOUT = 'frame_timeout';
 export const EVENT_CONNECT_ERROR = 'connect_error';
 export const EVENT_DISCONNECT_ERROR = 'disconnect_error';
 
+export const NO_WATCH = 'NO_WATCH';
+export const NO_SCHEMA = 'NO_SCHEMA';
+
 export default class Socket extends EventTarget {
   url;
   autoReconnect = true;
@@ -33,6 +37,7 @@ export default class Socket extends EventTarget {
   protocol = null;
   maxTries = null;
   tries = 0;
+  idAsTimestamp = false;
 
   // "Private"
   socket = null;
@@ -40,11 +45,11 @@ export default class Socket extends EventTarget {
   framesReceived = 0;
   frameTimer;
   reconnectTimer;
-  disconnectCbs = [];
+  disconnectCallBacks = [];
   disconnectedAt = 0;
   closingId = 0;
 
-  constructor(url, autoReconnect = true, frameTimeout = null, protocol = null, maxTries = null) {
+  constructor(url, autoReconnect = true, frameTimeout = null, protocol = null, maxTries = null, idAsTimestamp = false) {
     super();
 
     this.setUrl(url);
@@ -53,6 +58,7 @@ export default class Socket extends EventTarget {
     // maxTries = null === never stop trying to reconnect
     // allow maxTries to be defined on individual sockets bc not all will clearly warn the user that we've stopped trying
     this.maxTries = maxTries;
+    this.idAsTimestamp = idAsTimestamp;
 
     if ( frameTimeout !== null ) {
       this.frameTimeout = frameTimeout;
@@ -84,7 +90,7 @@ export default class Socket extends EventTarget {
 
     Object.assign(this.metadata, metadata);
 
-    const id = sockId++;
+    const id = this.idAsTimestamp ? new Date().getTime() : sockId++;
     const url = addParam(this.url, 'sockId', id);
 
     this._baseLog('connecting', { id, url: url.replace(/\?.*/, '') });
@@ -122,9 +128,9 @@ export default class Socket extends EventTarget {
     return false;
   }
 
-  disconnect(cb) {
-    if ( cb ) {
-      this.disconnectCbs.push(cb);
+  disconnect(callBack) {
+    if ( callBack ) {
+      this.disconnectCallBacks.push(callBack);
     }
 
     const self = this;
@@ -140,7 +146,7 @@ export default class Socket extends EventTarget {
 
       this.addEventListener(EVENT_CONNECT_ERROR, onError);
 
-      this.disconnectCbs.push(() => {
+      this.disconnectCallBacks.push(() => {
         this.removeEventListener(EVENT_CONNECT_ERROR, onError);
         resolve();
       });
@@ -280,10 +286,10 @@ export default class Socket extends EventTarget {
     clearTimeout(this.reconnectTimer);
     clearTimeout(this.frameTimer);
 
-    const cbs = this.disconnectCbs;
+    const callBacks = this.disconnectCallBacks;
 
-    while ( cbs.length ) {
-      const fn = cbs.pop();
+    while ( callBacks.length ) {
+      const fn = callBacks.pop();
 
       if ( fn ) {
         fn.apply(this);
@@ -310,17 +316,19 @@ export default class Socket extends EventTarget {
       this.state = STATE_RECONNECTING;
 
       if (this.maxTries && this.tries > 1 && this.tries <= this.maxTries) {
-        // dispatch an event which will trigger a growl from steve-plugin sockets warning users that we've lost connection and are attemping to reconnect
+        // dispatch an event which will trigger a growl from steve-plugin sockets warning users that we've lost connection and are attempting to reconnect
         const e = new CustomEvent(EVENT_CONNECT_ERROR);
 
         this.dispatchEvent(e);
       }
 
       if (this.maxTries && this.tries > this.maxTries) {
+        this._log('closed. Will not reconnect (hit max attempts)');
         this.state = STATE_DISCONNECTED;
         // dispatch an event which will trigger a growl from steve-plugin sockets warning users that we've given up trying to reconnect
         this.dispatchEvent(new CustomEvent(EVENT_DISCONNECT_ERROR));
       } else {
+        this._log('closed. Attempting to reconnect');
         const delay = Math.max(1000, Math.min(1000 * this.tries, 30000));
 
         this.reconnectTimer = setTimeout(() => {