Bugfix/reduce number of docs synced bulk (#2396)

CrowdDotDev · May 10, 2024 · f92176e · f92176e
1 parent 562a8be
commit f92176e
Show file tree

Hide file tree

Showing 2 changed files with 16 additions and 28 deletions.
diff --git a/services/libs/opensearch/src/service/member.sync.service.ts b/services/libs/opensearch/src/service/member.sync.service.ts
@@ -286,7 +286,7 @@ export class MemberSyncService {
 
   public async syncMembers(memberIds: string[], segmentIds?: string[]): Promise<IMemberSyncResult> {
     const CONCURRENT_DATABASE_QUERIES = 10
-    const BULK_INDEX_DOCUMENT_BATCH_SIZE = 2500
+    const BULK_INDEX_DOCUMENT_BATCH_SIZE = 100
 
     // get all memberId-segmentId couples
     const memberSegmentCouples: IMemberSegmentMatrix =
@@ -399,7 +399,9 @@ export class MemberSyncService {
         })
 
         // databaseStreams will create syncStreams items in processSegmentsStream, which'll later be used to sync to opensearch in bulk
-        if (databaseStream.length >= CONCURRENT_DATABASE_QUERIES) {
+        const isLastSegment = i === totalMemberIds - 1 && j === totalSegments - 1
+
+        if (isLastSegment || databaseStream.length >= CONCURRENT_DATABASE_QUERIES) {
           await processSegmentsStream(databaseStream)
           databaseStream = []
         }
@@ -413,18 +415,10 @@ export class MemberSyncService {
           syncStream = syncStream.slice(BULK_INDEX_DOCUMENT_BATCH_SIZE)
         }
 
-        // if we're processing the last segment
-        if (i === totalMemberIds - 1 && j === totalSegments - 1) {
-          // check if there are remaining databaseStreams to process
-          if (databaseStream.length > 0) {
-            await processSegmentsStream(databaseStream)
-          }
-
-          // check if there are remaining syncStreams to process
-          if (syncStream.length > 0) {
-            await this.openSearchService.bulkIndex(OpenSearchIndex.MEMBERS, syncStream)
-            documentsIndexed += syncStream.length
-          }
+        // check if there are remaining syncStreams to process
+        if (isLastSegment && syncStream.length > 0) {
+          await this.openSearchService.bulkIndex(OpenSearchIndex.MEMBERS, syncStream)
+          documentsIndexed += syncStream.length
         }
         successfullySyncedMembers.push(memberId)
       }

diff --git a/services/libs/opensearch/src/service/organization.sync.service.ts b/services/libs/opensearch/src/service/organization.sync.service.ts
@@ -260,7 +260,7 @@ export class OrganizationSyncService {
     segmentIds?: string[],
   ): Promise<IOrganizationSyncResult> {
     const CONCURRENT_DATABASE_QUERIES = 5
-    const BULK_INDEX_DOCUMENT_BATCH_SIZE = 2500
+    const BULK_INDEX_DOCUMENT_BATCH_SIZE = 100
 
     // get all orgId-segmentId couples
     const orgSegmentCouples: IOrganizationSegmentMatrix =
@@ -360,7 +360,9 @@ export class OrganizationSyncService {
         })
 
         // databaseStreams will create syncStreams items in processSegmentsStream, which'll later be used to sync to opensearch in bulk
-        if (databaseStream.length >= CONCURRENT_DATABASE_QUERIES) {
+        const isLastSegment = i === totalOrgIds - 1 && j === totalSegments - 1
+
+        if (isLastSegment || databaseStream.length >= CONCURRENT_DATABASE_QUERIES) {
           await processSegmentsStream(databaseStream)
           databaseStream = []
         }
@@ -374,18 +376,10 @@ export class OrganizationSyncService {
           syncStream = syncStream.slice(BULK_INDEX_DOCUMENT_BATCH_SIZE)
         }
 
-        // if we're processing the last segment
-        if (i === totalOrgIds - 1 && j === totalSegments - 1) {
-          // check if there are remaining databaseStreams to process
-          if (databaseStream.length > 0) {
-            await processSegmentsStream(databaseStream)
-          }
-
-          // check if there are remaining syncStreams to process
-          if (syncStream.length > 0) {
-            await this.openSearchService.bulkIndex(OpenSearchIndex.ORGANIZATIONS, syncStream)
-            documentsIndexed += syncStream.length
-          }
+        // check if there are remaining syncStreams to process
+        if (isLastSegment && syncStream.length > 0) {
+          await this.openSearchService.bulkIndex(OpenSearchIndex.ORGANIZATIONS, syncStream)
+          documentsIndexed += syncStream.length
         }
       }
     }