[backend] Introduce doYield method (#12627) (#12628)

Author: xfournet

opencti-platform/opencti-graphql/src/database/engine.js

@@ -24,7 +24,6 @@ import {
   isEmptyField,
   isInferredIndex,
   isNotEmptyField,
-  MAX_EVENT_LOOP_PROCESSING_TIME,
   offsetToCursor,
   pascalize,
   READ_DATA_INDICES,
@@ -222,6 +221,7 @@ import { getPirWithAccessCheck } from '../modules/pir/pir-checkPirAccess';
 import { asyncFilter, asyncMap, uniqAsyncMap } from '../utils/data-processing';
 import { ENTITY_TYPE_PIR } from '../modules/pir/pir-types';
 import { isMetricsName } from '../modules/metrics/metrics-utils';
+import { doYield } from '../utils/eventloop-utils';
 
 const ELK_ENGINE = 'elk';
 const OPENSEARCH_ENGINE = 'opensearch';
@@ -1575,8 +1575,8 @@ const elDataConverter = (esHit) => {
 export const elConvertHitsToMap = async (elements, opts) => {
   const { mapWithAllIds = false } = opts;
   const convertedHitsMap = {};
-  let startProcessingTime = new Date().getTime();
   for (let n = 0; n < elements.length; n += 1) {
+    await doYield();
     const element = elements[n];
     convertedHitsMap[element.internal_id] = element;
     if (mapWithAllIds) {
@@ -1589,13 +1589,6 @@ export const elConvertHitsToMap = async (elements, opts) => {
         convertedHitsMap[id] = element;
       });
     }
-    // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-    if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-      startProcessingTime = new Date().getTime();
-      await new Promise((resolve) => {
-        setImmediate(resolve);
-      });
-    }
   }
   return convertedHitsMap;
 };
@@ -3841,19 +3834,12 @@ const buildRegardingOfFilter = async (context, user, elements, filters) => {
             sideIdManualInferred.set(sideId, toTypes);
           }
         };
-        let startProcessingTime = new Date().getTime();
         for (let relIndex = 0; relIndex < relationships.length; relIndex += 1) {
+          await doYield();
           const relation = relationships[relIndex];
           const relType = isInferredIndex(relation._index) ? 'inferred' : 'manual';
           addTypeSide(relation.fromId, relType);
           addTypeSide(relation.toId, relType);
-          // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-          if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-            startProcessingTime = new Date().getTime();
-            await new Promise((resolve) => {
-              setImmediate(resolve);
-            });
-          }
         }
       }
       return (element) => {
@@ -4182,24 +4168,17 @@ export const elRemoveRelationConnection = async (context, user, elementsImpact,
     const elIdsCache = {};
     const indexCache = {};
     const pirInformationCache = {};
-    let startProcessingTime = new Date().getTime();
     for (let idIndex = 0; idIndex < dataIds.length; idIndex += 1) {
+      await doYield();
       const element = dataIds[idIndex];
       elIdsCache[element.internal_id] = element._id;
       indexCache[element.internal_id] = element._index;
       pirInformationCache[element.internal_id] = element.pir_information;
-      // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-      if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-        startProcessingTime = new Date().getTime();
-        await new Promise((resolve) => {
-          setImmediate(resolve);
-        });
-      }
     }
     // Split by max operations, create the bulk
     const groupsOfImpacts = R.splitEvery(MAX_BULK_OPERATIONS, impacts);
-    startProcessingTime = new Date().getTime();
     for (let i = 0; i < groupsOfImpacts.length; i += 1) {
+      await doYield();
       const impactsBulk = groupsOfImpacts[i];
       const bodyUpdateRaw = impactsBulk.map(([impactId, elementMeta]) => {
         return Object.entries(elementMeta).map(([typeAndIndex, cleanupIds]) => {
@@ -4256,22 +4235,15 @@ export const elRemoveRelationConnection = async (context, user, elementsImpact,
       if (bodyUpdate.length > 0) {
         await elBulk({ refresh: forceRefresh, timeout: BULK_TIMEOUT, body: bodyUpdate });
       }
-      // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-      if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-        startProcessingTime = new Date().getTime();
-        await new Promise((resolve) => {
-          setImmediate(resolve);
-        });
-      }
     }
   }
 };
 
 export const computeDeleteElementsImpacts = async (cleanupRelations, toBeRemovedIds, relationsToRemoveMap) => {
   // Update all rel connections that will remain
   const elementsImpact = {};
-  let startProcessingTime = new Date().getTime();
   for (let i = 0; i < cleanupRelations.length; i += 1) {
+    await doYield();
     const relation = cleanupRelations[i];
     const fromWillNotBeRemoved = !relationsToRemoveMap.has(relation.fromId) && !toBeRemovedIds.includes(relation.fromId);
     const isFromCleanup = fromWillNotBeRemoved && isImpactedTypeAndSide(relation.entity_type, relation.fromType, relation.toType, ROLE_FROM);
@@ -4303,13 +4275,6 @@ export const computeDeleteElementsImpacts = async (cleanupRelations, toBeRemoved
         }
       }
     }
-    // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-    if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-      startProcessingTime = new Date().getTime();
-      await new Promise((resolve) => {
-        setImmediate(resolve);
-      });
-    }
   }
   return elementsImpact;
 };
@@ -4528,15 +4493,22 @@ const createDeleteOperationElement = async (context, user, mainElement, deletedE
 export const prepareElementForIndexing = async (element) => {
   const thing = {};
   const keyItems = Object.keys(element);
-  let startProcessingTime = new Date().getTime();
   for (let index = 0; index < keyItems.length; index += 1) {
+    await doYield();
     const key = keyItems[index];
     const value = element[key];
     if (Array.isArray(value)) { // Array of Date, objects, string or number
       const preparedArray = [];
-      let innerProcessingTime = new Date().getTime();
-      let extendLoopSplit = 0;
+      let yieldCount = 0;
       for (let valueIndex = 0; valueIndex < value.length; valueIndex += 1) {
+        if (await doYield()) {
+          // If we extend the preparation 5 times, log a warn
+          // It will help to understand what kind of key have so many elements
+          if (yieldCount === 5) {
+            logApp.warn('[ENGINE] Element preparation too many values', { id: element.id, key, size: value.length });
+          }
+          yieldCount += 1;
+        }
         const valueElement = value[valueIndex];
         if (valueElement) {
           if (isDateAttribute(key)) { // Date is an object but natively supported
@@ -4551,19 +4523,6 @@ export const prepareElementForIndexing = async (element) => {
             preparedArray.push(valueElement);
           }
         }
-        // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-        if (new Date().getTime() - innerProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-          // If we extends the preparation 5 times, log a warn
-          // It will help to understand what kind of key have so much elements
-          if (extendLoopSplit === 5) {
-            logApp.warn('[ENGINE] Element preparation too many values', { id: element.id, key, size: value.length });
-          }
-          extendLoopSplit += 1;
-          innerProcessingTime = new Date().getTime();
-          await new Promise((resolve) => {
-            setImmediate(resolve);
-          });
-        }
       }
       thing[key] = preparedArray;
     } else if (isDateAttribute(key)) { // Date is an object but natively supported
@@ -4579,13 +4538,6 @@ export const prepareElementForIndexing = async (element) => {
     } else { // For all other types (numeric, ...), no transform
       thing[key] = value;
     }
-    // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-    if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-      startProcessingTime = new Date().getTime();
-      await new Promise((resolve) => {
-        setImmediate(resolve);
-      });
-    }
   }
   return thing;
 };

opencti-platform/opencti-graphql/src/database/middleware.js

@@ -32,7 +32,6 @@ import {
   isInferredIndex,
   isNotEmptyField,
   isObjectPathTargetMultipleAttribute,
-  MAX_EVENT_LOOP_PROCESSING_TIME,
   READ_DATA_INDICES,
   READ_DATA_INDICES_INFERRED,
   READ_INDEX_HISTORY,
@@ -237,6 +236,7 @@ import { RELATION_ACCESSES_TO } from '../schema/internalRelationship';
 import { generateVulnerabilitiesUpdates } from '../utils/vulnerabilities';
 import { idLabel } from '../schema/schema-labels';
 import { pirExplanation } from '../modules/attributes/internalRelationship-registrationAttributes';
+import { doYield } from '../utils/eventloop-utils';
 
 // region global variables
 const MAX_BATCH_SIZE = nconf.get('elasticsearch:batch_loader_max_size') ?? 300;
@@ -369,22 +369,15 @@ const loadElementMetaDependencies = async (context, user, elements, args = {}) =
         const [key, values] = entries[index];
         const invalidRelations = [];
         const resolvedElementsWithRelation = [];
-        let startProcessingTime = new Date().getTime();
         for (let valueIndex = 0; valueIndex < values.length; valueIndex += 1) {
+          await doYield();
           const v = values[valueIndex];
           const resolvedElement = toResolvedElements[v.toId];
           if (resolvedElement) {
             resolvedElementsWithRelation.push({ ...resolvedElement, i_relation: v });
           } else {
             invalidRelations.push({ relation_id: v.id, target_id: v.toId });
           }
-          // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-          if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-            startProcessingTime = new Date().getTime();
-            await new Promise((resolve) => {
-              setImmediate(resolve);
-            });
-          }
         }
         if (invalidRelations.length > 0) {
           // Some targets can be unresolved in case of potential inconsistency between relation and target
@@ -440,8 +433,8 @@ export const loadElementsWithDependencies = async (context, user, elements, opts
   }
   const [fromAndToMap, depsElementsMap, fileMarkingsMap] = await Promise.all([fromAndToPromise, depsPromise, fileMarkingsPromise]);
   const loadedElements = [];
-  let startProcessingTime = new Date().getTime();
   for (let i = 0; i < elements.length; i += 1) {
+    await doYield();
     const element = elements[i];
     const files = [];
     if (isNotEmptyField(element.x_opencti_files) && isNotEmptyField(fileMarkingsMap)) {
@@ -482,13 +475,6 @@ export const loadElementsWithDependencies = async (context, user, elements, opts
     } else {
       loadedElements.push(R.mergeRight(element, { ...deps }));
     }
-    // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-    if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-      startProcessingTime = new Date().getTime();
-      await new Promise((resolve) => {
-        setImmediate(resolve);
-      });
-    }
   }
   return loadedElements;
 };
@@ -867,7 +853,6 @@ const inputResolveRefs = async (context, user, input, type, entitySetting) => {
     }
     const resolutionsMap = new Map();
     const resolvedIds = new Set();
-    let startProcessingTime = new Date().getTime();
     for (let i = 0; i < resolvedElements.length; i += 1) {
       const resolvedElement = resolvedElements[i];
       const instanceIds = getInstanceIds(resolvedElement);
@@ -879,17 +864,11 @@ const inputResolveRefs = async (context, user, input, type, entitySetting) => {
         }
       });
       for (let configIndex = 0; configIndex < matchingConfigs.length; configIndex += 1) {
+        await doYield();
         const c = matchingConfigs[configIndex];
         const data = { ...resolvedElement, i_group: c };
         const dataKey = `${resolvedElement.internal_id}|${c.destKey}`;
         resolutionsMap.set(dataKey, data);
-        // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-        if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-          startProcessingTime = new Date().getTime();
-          await new Promise((resolve) => {
-            setImmediate(resolve);
-          });
-        }
       }
     }
     const groupByTypeElements = R.groupBy((e) => e.i_group.destKey, resolutionsMap.values());

opencti-platform/opencti-graphql/src/database/utils.js

@@ -15,13 +15,12 @@ import { isStixRefRelationship, RELATION_OBJECT_MARKING } from '../schema/stixRe
 import { schemaAttributesDefinition } from '../schema/schema-attributes';
 import { getDraftContext } from '../utils/draftContext';
 import { INPUT_OBJECTS } from '../schema/general';
+import { doYield } from '../utils/eventloop-utils';
 
 export const ES_INDEX_PREFIX = conf.get('elasticsearch:index_prefix') || 'opencti';
 const rabbitmqPrefix = conf.get('rabbitmq:queue_prefix');
 export const RABBIT_QUEUE_PREFIX = rabbitmqPrefix ? `${rabbitmqPrefix}_` : '';
 
-export const MAX_EVENT_LOOP_PROCESSING_TIME = 50;
-
 export const REDACTED_INFORMATION = '*** Redacted ***';
 export const RESTRICTED_INFORMATION = 'Restricted';
 
@@ -390,18 +389,11 @@ export const isObjectPathTargetMultipleAttribute = (instance, object_path) => {
 
 export const asyncListTransformation = async (elements, preparatoryFunction, opts = {}) => {
   const preparedElements = [];
-  let startProcessingTime = new Date().getTime();
   for (let n = 0; n < elements.length; n += 1) {
+    await doYield();
     const element = elements[n];
     const preparedElement = await preparatoryFunction(element, opts);
     preparedElements.push(preparedElement);
-    // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-    if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-      startProcessingTime = new Date().getTime();
-      await new Promise((resolve) => {
-        setImmediate(resolve);
-      });
-    }
   }
   return preparedElements;
 };

opencti-platform/opencti-graphql/src/domain/user.js

@@ -32,15 +32,7 @@ import {
 } from '../database/middleware-loader';
 import { delEditContext, notify, setEditContext } from '../database/redis';
 import { killUserSessions } from '../database/session';
-import {
-  buildPagination,
-  isEmptyField,
-  isNotEmptyField,
-  MAX_EVENT_LOOP_PROCESSING_TIME,
-  READ_INDEX_INTERNAL_OBJECTS,
-  READ_INDEX_STIX_DOMAIN_OBJECTS,
-  READ_RELATIONSHIPS_INDICES
-} from '../database/utils';
+import { buildPagination, isEmptyField, isNotEmptyField, READ_INDEX_INTERNAL_OBJECTS, READ_INDEX_STIX_DOMAIN_OBJECTS, READ_RELATIONSHIPS_INDICES } from '../database/utils';
 import { extractEntityRepresentativeName } from '../database/entity-representative';
 import { publishUserAction } from '../listener/UserActionListener';
 import { authorizedMembers } from '../schema/attribute-definition';
@@ -90,6 +82,7 @@ import { addServiceAccountIntoUserCount, addUserEmailSendCount, addUserIntoServi
 import { sendMail } from '../database/smtp';
 import { checkEnterpriseEdition } from '../enterprise-edition/ee';
 import { ENTITY_TYPE_EMAIL_TEMPLATE } from '../modules/emailTemplate/emailTemplate-types';
+import { doYield } from '../utils/eventloop-utils';
 
 const BEARER = 'Bearer ';
 const BASIC = 'Basic ';
@@ -1473,8 +1466,8 @@ export const buildCompleteUsers = async (context, clients) => {
   const groupsRoles = new Map();
   const groupsMarkings = new Map();
   const rolesCapabilities = new Map();
-  let startProcessingTime = new Date().getTime();
   for (let index = 0; index < relations.length; index += 1) {
+    await doYield();
     const { fromId, entity_type, toId } = relations[index];
     // group <- RELATION_ACCESSES_TO -> marking
     if (entity_type === RELATION_ACCESSES_TO) {
@@ -1537,13 +1530,6 @@ export const buildCompleteUsers = async (context, clients) => {
         groupsRoles.set(fromId, [toId]);
       }
     }
-    // Prevent event loop locking more than MAX_EVENT_LOOP_PROCESSING_TIME
-    if (new Date().getTime() - startProcessingTime > MAX_EVENT_LOOP_PROCESSING_TIME) {
-      startProcessingTime = new Date().getTime();
-      await new Promise((resolve) => {
-        setImmediate(resolve);
-      });
-    }
   }
   const ids = [...Array.from(groupIds), ...Array.from(roleIds), ...Array.from(organizationIds), ...Array.from(capabilityIds)];
   const resolvedObject = await internalFindByIds(context, SYSTEM_USER, ids, { toMap: true });