[lib] Refactor threads queue in DMOpsQueueHandler

Summary:
https://linear.app/comm/issue/ENG-9808/consider-modifying-the-queues-logic-to-be-more-resilient

Depends on D13860

Test Plan: Add entities to the threads queue by calling processDMOps from useProcessDMOperation and verify entities are processed.

Reviewers: tomek, kamil

Reviewed By: tomek

Subscribers: ashoat

Differential Revision: https://phab.comm.dev/D13861

lib/shared/dm-ops/dm-ops-queue-handler.react.js

@@ -63,7 +63,6 @@ function DMOpsQueueHandler(): React.Node {
   }, [prune]);
 
   const threadInfos = useSelector(threadInfoSelector);
-  const prevThreadInfosRef = React.useRef({});
 
   const queuedThreadOperations = useSelector(
     state => state.queuedDMOperations.threadQueue,
@@ -95,14 +94,20 @@ function DMOpsQueueHandler(): React.Node {
 
   const { enqueue } = useActionsQueue(processItem);
 
-  React.useEffect(() => {
-    const prevThreadInfos = prevThreadInfosRef.current;
-    prevThreadInfosRef.current = threadInfos;
+  const runningThreadOperations = React.useRef<Set<string>>(new Set());
 
+  React.useEffect(() => {
     for (const threadID in queuedThreadOperations) {
-      if (!threadInfos[threadID] || prevThreadInfos[threadID]) {
+      if (!threadInfos[threadID]) {
+        continue;
+      }
+
+      if (runningThreadOperations.current.has(threadID)) {
         continue;
       }
+
+      runningThreadOperations.current.add(threadID);
+
       enqueue([
         ...queuedThreadOperations[threadID].map(item => ({
           type: 'operation',
@@ -117,6 +122,10 @@ function DMOpsQueueHandler(): React.Node {
             },
           },
         },
+        {
+          type: 'function',
+          itemFunction: () => runningThreadOperations.current.delete(threadID),
+        },
       ]);
     }
   }, [dispatch, enqueue, queuedThreadOperations, threadInfos]);