chore: web worker shreds cache

Author: asuzuki-jumptrading

src/api/useSetAtomWsData.ts

@@ -7,9 +7,10 @@ import type {
   FromWorkerMessage,
   HistoryArrayKey,
   KeyedValuesWithHistory,
+  LiveShredsItem,
   WsEntity,
 } from "./worker/types";
-import { isEmaObjectKey } from "./worker/types";
+import { isEmaObjectKey, liveShredsKey } from "./worker/types";
 import { DateTime } from "luxon";
 import { useInterval } from "react-use";
 import { useThrottledCallback, useDebouncedCallback } from "use-debounce";
@@ -36,7 +37,7 @@ import {
   deleteSupermajorityDeltaEntriesAtom,
   resetSupermajorityAtom,
 } from "../atoms";
-import { shredsAtoms } from "../features/Overview/ShredsProgression/atoms";
+import { liveShredsDataAtom } from "../features/Overview/ShredsProgression/atoms";
 import { rateLiveWaterfallAtom } from "../features/Overview/SlotPerformance/atoms";
 import {
   addTurbineSlotsAtom,
@@ -118,12 +119,11 @@ import type {
   PeerRemove,
 } from "./types";
 import { SocketState } from "./ws/types";
-import { xRangeMs } from "../features/Overview/ShredsProgression/const";
+import { useServerMessages } from "./ws/utils";
 import {
-  bootProgressPhaseAtom,
   showStartupProgressAtom,
+  bootProgressPhaseAtom,
 } from "../features/StartupProgress/atoms";
-import { useServerMessages } from "./ws/utils";
 
 export function useSetAtomWsData() {
   const setSocketState = useSetAtom(socketStateAtom);
@@ -169,6 +169,18 @@ export function useSetAtomWsData() {
     [setTileTimerHistory],
   );
 
+  const setLiveShreds = useSetAtom(liveShredsDataAtom);
+  const updateLiveShredsObject = useCallback(
+    ({ key, data }: LiveShredsItem) => {
+      switch (key) {
+        case liveShredsKey:
+          setLiveShreds(data);
+          break;
+      }
+    },
+    [setLiveShreds],
+  );
+
   const onMessage = useCallback(
     (msg: FromWorkerMessage) => {
       switch (msg.type) {
@@ -207,6 +219,11 @@ export function useSetAtomWsData() {
             updateEmaHistoryObject(item);
           }
           break;
+        case "liveShredsObject":
+          for (const item of msg.items) {
+            updateLiveShredsObject(item);
+          }
+          break;
       }
     },
     [
@@ -215,6 +232,7 @@ export function useSetAtomWsData() {
       updateEmaHistoryArray,
       updateHistoryArray,
       updateEmaHistoryObject,
+      updateLiveShredsObject,
     ],
   );
 
@@ -430,8 +448,6 @@ function useUpdateAtoms() {
   );
   const setSlotCaughtUp = useSetAtom(slotCaughtUpAtom);
 
-  const addLiveShreds = useSetAtom(shredsAtoms.addShredEvents);
-
   const setLiveProgramCache = useSetAtom(liveProgramCacheAtom);
 
   const peersBuffer = useRef(new Map<string, Peer>());
@@ -717,7 +733,7 @@ function useUpdateAtoms() {
               break;
             }
             case "live_shreds": {
-              addLiveShreds(value);
+              // processed by web worker
               break;
             }
             case "late_votes_history": {
@@ -755,7 +771,6 @@ function useUpdateAtoms() {
       }
     },
     [
-      addLiveShreds,
       addRepairSlot,
       addRepairSlots,
       addSkippedClusterSlots,
@@ -847,14 +862,6 @@ function useUpdateAtoms() {
   const isSocketDisconnected =
     useAtomValue(socketStateAtom) === SocketState.Disconnected;
 
-  const deleteLiveShreds = useSetAtom(shredsAtoms.deleteSlots);
-
-  useEffect(() => {
-    if (isSocketDisconnected) {
-      deleteLiveShreds(isSocketDisconnected, isStartup);
-    }
-  }, [deleteLiveShreds, isSocketDisconnected, isStartup]);
-
   const resetTurbineSlots = useSetAtom(resetTurbineSlotsAtom);
   const resetRepairSlots = useSetAtom(resetRepairSlotsAtom);
   useEffect(() => {
@@ -864,13 +871,6 @@ function useUpdateAtoms() {
     }
   }, [isStartup, resetRepairSlots, resetTurbineSlots]);
 
-  useInterval(
-    () => {
-      deleteLiveShreds(isSocketDisconnected, isStartup);
-    },
-    isStartup ? 1_000 : xRangeMs / 4,
-  );
-
   const deleteSupermajorityDeltaEntries = useSetAtom(
     deleteSupermajorityDeltaEntriesAtom,
   );

src/api/worker/cache/batchPublisher.ts

@@ -22,6 +22,7 @@ interface BatchPublisherConfig<
   /** Sends the collected batch of messages to the main thread. */
   post: (items: TMessage[]) => void;
   onReset?: (entry: TEntry) => void;
+  onStop?: (entry: TEntry) => void;
 }
 
 export function createBatchPublisher<
@@ -45,6 +46,24 @@ export function createBatchPublisher<
     return e.lastPublishMs + e.publishIntervalMs;
   }
 
+  function collectAndPost(entries: Iterable<TEntry>, isForced: boolean) {
+    const nowMs = performance.now();
+    const batch: TMessage[] = [];
+    for (const e of entries) {
+      if (!isForced && (!e.subscribed || nowMs < nextDueAt(e))) {
+        continue;
+      }
+      const item = config.collect(e, nowMs);
+      if (item !== undefined) {
+        batch.push(item);
+      }
+      e.lastPublishMs = nowMs;
+    }
+    if (batch.length) {
+      config.post(batch);
+    }
+  }
+
   function schedule() {
     if (timer) return;
 
@@ -61,23 +80,7 @@ export function createBatchPublisher<
     timer = setTimeout(
       () => {
         timer = undefined;
-        const nowMs = performance.now();
-        const batch: TMessage[] = [];
-
-        for (const e of entries.values()) {
-          if (!e.subscribed) continue;
-          if (nowMs < nextDueAt(e)) continue;
-
-          const item = config.collect(e, nowMs);
-          if (item !== undefined) {
-            batch.push(item);
-          }
-          e.lastPublishMs = nowMs;
-        }
-
-        if (batch.length) {
-          config.post(batch);
-        }
+        collectAndPost(entries.values(), false);
         schedule();
       },
       Math.max(0, nextDue),
@@ -114,6 +117,12 @@ export function createBatchPublisher<
       return entries.get(key);
     },
 
+    publishNow: (key: TEntry["key"]) => {
+      const e = entries.get(key);
+      if (!e) return;
+      collectAndPost([e], true);
+    },
+
     reset: (key?: TEntry["key"]) => {
       const toReset = key
         ? [entries.get(key)].filter(isDefined)
@@ -133,6 +142,7 @@ export function createBatchPublisher<
 
       for (const e of entries.values()) {
         e.subscribed = false;
+        config.onStop?.(e);
       }
     },
   };

src/api/worker/cache/consts.ts

@@ -5,3 +5,5 @@ export const gossipHealthHistoryBufferMs = 5_000;
 export const overviewPublishIntervalMs = 500;
 export const overviewRenderWindowMs = 60_000;
 export const overviewHistoryBufferMs = 5_000;
+
+export const shredsPublishIntervalMs = 50;