fix: paginate organization member sync with heartbeats to prevent timeout loop (#4042)

Signed-off-by: Yeganathan S <63534555+skwowet@users.noreply.github.com>

Author: skwowet

backend/src/services/searchSyncService.ts

@@ -66,11 +66,13 @@ export default class SearchSyncService extends LoggerBase {
   async triggerOrganizationMembersSync(organizationId: string) {
     const client = await this.getSearchSyncClient()
 
-    if (client instanceof SearchSyncApiClient || client instanceof SearchSyncWorkerEmitter) {
+    if (client instanceof SearchSyncApiClient) {
       await this.logExecutionTime(
-        () => client.triggerOrganizationMembersSync(organizationId, false),
+        () => client.syncOrganizationMembers(organizationId),
         `triggerOrganizationMembersSync: organization:${organizationId}`,
       )
+    } else if (client instanceof SearchSyncWorkerEmitter) {
+      await client.triggerOrganizationMembersSync(organizationId, false)
     } else {
       throw new Error('Unexpected search client type!')
     }

pnpm-lock.yaml

@@ -20,6 +20,7 @@
     "@crowd/queue": "workspace:*",
     "@crowd/redis": "workspace:*",
     "@crowd/types": "workspace:*",
+    "@temporalio/activity": "~1.11.8",
     "@temporalio/workflow": "~1.11.8",
     "tsx": "^4.7.1",
     "typescript": "^5.6.3"

services/apps/entity_merging_worker/package.json

@@ -1,3 +1,4 @@
+import { Context, heartbeat } from '@temporalio/activity'
 import { WorkflowIdConflictPolicy, WorkflowIdReusePolicy } from '@temporalio/workflow'
 
 import { DEFAULT_TENANT_ID } from '@crowd/common'
@@ -96,7 +97,19 @@ export async function syncOrganization(organizationId: string, syncStart: Date):
   })
 
   await syncApi.triggerOrganizationSync(organizationId)
-  await syncApi.triggerOrganizationMembersSync(organizationId, null, syncStart)
+
+  const cursor = Context.current().info.heartbeatDetails as string | undefined
+  if (cursor) {
+    svc.log.info({ organizationId, cursor }, 'Resuming organization member sync from heartbeat')
+  }
+
+  const { totalSynced } = await syncApi.syncOrganizationMembers(organizationId, {
+    syncFrom: syncStart,
+    cursor,
+    onPageComplete: (c) => heartbeat(c),
+  })
+
+  svc.log.info({ organizationId, totalSynced }, 'Finished syncing organization members')
 }
 
 export async function notifyFrontendOrganizationUnmergeSuccessful(

services/apps/entity_merging_worker/src/activities/organizations.ts

@@ -14,14 +14,18 @@ const {
   recalculateActivityAffiliationsOfOrganizationAsync,
   setMergeAction,
   syncMember,
-  syncOrganization,
   notifyFrontendMemberMergeSuccessful,
   notifyFrontendMemberUnmergeSuccessful,
   syncRemoveMember,
   finishMemberMergingUpdateActivities,
   finishMemberUnmergingUpdateActivities,
 } = proxyActivities<typeof activities>({
-  startToCloseTimeout: '60 minutes',
+  startToCloseTimeout: '2 hours',
+})
+
+const { syncOrganization } = proxyActivities<typeof activities>({
+  startToCloseTimeout: '2 hours',
+  heartbeatTimeout: '5 minutes',
 })
 
 export async function deleteOrphanMember(memberId: string): Promise<void> {

services/apps/entity_merging_worker/src/workflows/all.ts

@@ -30,17 +30,14 @@ router.post(
 router.post(
   '/sync/organization/members',
   asyncWrap(async (req: ApiRequest, res) => {
-    const memberSyncService = syncService(req)
-
-    const { organizationId, syncFrom } = req.body
+    const { organizationId, lastId, batchSize, syncFrom } = req.body
     try {
-      req.log.trace(
-        `Calling memberSyncService.syncOrganizationMembers for organization ${organizationId}`,
-      )
-      await memberSyncService.syncOrganizationMembers(organizationId, {
+      const result = await syncService(req).syncOrganizationMembers(organizationId, {
+        lastId: lastId ?? undefined,
+        batchSize: batchSize ?? undefined,
         syncFrom: syncFrom ? new Date(syncFrom) : null,
       })
-      res.sendStatus(200)
+      res.json(result)
     } catch (error) {
       req.log.error(error)
       res.status(500).send(error.message)

services/apps/search_sync_api/src/routes/member.ts

@@ -96,9 +96,15 @@ export class WorkerQueueReceiver extends PrioritizedQueueReciever {
           break
         case SearchSyncWorkerQueueMessageType.SYNC_ORGANIZATION_MEMBERS:
           if (data.organizationId) {
-            this.initMemberService()
-              .syncOrganizationMembers(data.organizationId)
-              .catch((err) => this.log.error(err, 'Error while syncing organization members!'))
+            const memberService = this.initMemberService()
+            const orgId = data.organizationId as string
+            ;(async () => {
+              let lastId: string | undefined
+              do {
+                const result = await memberService.syncOrganizationMembers(orgId, { lastId })
+                lastId = result.lastId ?? undefined
+              } while (lastId)
+            })().catch((err) => this.log.error(err, 'Error while syncing organization members!'))
           }
 
           break

services/apps/search_sync_worker/src/queue/index.ts

@@ -4,6 +4,14 @@ export interface ISearchSyncApiConfig {
   baseUrl: string
 }
 
+export interface ISyncOrganizationMembersOptions {
+  syncFrom?: Date | null
+  cursor?: string
+  pageSize?: number
+  maxPages?: number
+  onPageComplete?: (cursor: string, membersSynced: number) => void | Promise<void>
+}
+
 export class SearchSyncApiClient {
   private searchSyncApi
 
@@ -23,19 +31,43 @@ export class SearchSyncApiClient {
     })
   }
 
-  public async triggerOrganizationMembersSync(
+  public async syncOrganizationMembers(
     organizationId: string,
-    onboarding?: boolean,
-    syncFrom: Date | null = null,
-  ): Promise<void> {
+    opts: ISyncOrganizationMembersOptions = {},
+  ): Promise<{ totalSynced: number }> {
     if (!organizationId) {
       throw new Error('organizationId is required!')
     }
 
-    await this.searchSyncApi.post('/sync/organization/members', {
-      organizationId,
-      syncFrom,
-    })
+    const pageSize = opts.pageSize ?? 500
+    const maxPages = opts.maxPages ?? 2000
+
+    let cursor = opts.cursor
+    let totalSynced = 0
+
+    for (let i = 0; i < maxPages; i++) {
+      const { data } = await this.searchSyncApi.post('/sync/organization/members', {
+        organizationId,
+        lastId: cursor ?? null,
+        batchSize: pageSize,
+        syncFrom: opts.syncFrom ?? null,
+      })
+
+      const result: { lastId: string | null; membersSynced: number } = data
+      totalSynced += result.membersSynced
+
+      if (result.lastId === null || result.membersSynced < pageSize) {
+        return { totalSynced }
+      }
+
+      cursor = result.lastId
+      await opts.onPageComplete?.(cursor, result.membersSynced)
+    }
+
+    throw new Error(
+      `syncOrganizationMembers exceeded maxPages (${maxPages}) for organization ${organizationId}. ` +
+        `Synced ${totalSynced} members before aborting.`,
+    )
   }
 
   public async triggerRemoveMember(memberId: string): Promise<void> {

services/libs/opensearch/src/apiClient.ts

@@ -22,7 +22,7 @@ export class MemberRepository extends RepositoryBase<MemberRepository> {
       return JSON.parse(cachedString)
     }
 
-    const results = await this.db().one(
+    const results = await this.db().any(
       `select type, "canDelete", show, label, name, options from "memberAttributeSettings"`,
     )
 
@@ -37,6 +37,10 @@ export class MemberRepository extends RepositoryBase<MemberRepository> {
     lastId?: string,
     syncFrom?: Date,
   ): Promise<string[]> {
+    if (!Number.isInteger(perPage) || perPage <= 0) {
+      throw new Error('perPage must be a positive integer')
+    }
+
     const rows = await this.db().any(
       `
         SELECT
@@ -51,12 +55,13 @@ export class MemberRepository extends RepositoryBase<MemberRepository> {
             m."deletedAt" is null AND
             exists (select 1 from "memberIdentities" where "memberId" = mo."memberId" and "deletedAt" is null)
         ORDER BY mo."memberId"
-        LIMIT ${perPage};
+        LIMIT $(perPage);
       `,
       {
         organizationId,
         lastId,
         syncFrom,
+        perPage,
       },
     )
 

services/libs/opensearch/src/repo/member.repo.ts

@@ -207,64 +207,48 @@ export class MemberSyncService {
 
   public async syncOrganizationMembers(
     organizationId: string,
-    opts: { syncFrom: Date | null } = { syncFrom: null },
-  ): Promise<void> {
-    this.log.debug({ organizationId }, 'Syncing all organization members!')
-    const batchSize = 500
-    let docCount = 0
-    const memberCount = 0
-
-    const now = new Date()
-
-    const loadNextPage = async (lastId?: string): Promise<string[]> => {
-      this.log.info('Loading next page of organization members!', { organizationId, lastId })
-      const memberIds = await logExecutionTimeV2(
-        () =>
-          this.memberRepo.getOrganizationMembersForSync(
-            organizationId,
-            batchSize,
-            lastId,
-            opts.syncFrom,
-          ),
-        this.log,
-        `getOrganizationMembersForSync`,
-      )
-
-      if (memberIds.length === 0) {
-        return []
-      }
+    opts: {
+      lastId?: string
+      batchSize?: number
+      syncFrom?: Date | null
+    } = {},
+  ): Promise<{ lastId: string | null; membersSynced: number }> {
+    const batchSize = opts.batchSize ?? 500
+
+    const memberIds = await logExecutionTimeV2(
+      () =>
+        this.memberRepo.getOrganizationMembersForSync(
+          organizationId,
+          batchSize,
+          opts.lastId,
+          opts.syncFrom ?? undefined,
+        ),
+      this.log,
+      'getOrganizationMembersForSync',
+    )
 
-      return memberIds
+    if (memberIds.length === 0) {
+      return { lastId: null, membersSynced: 0 }
     }
 
-    let memberIds: string[] = await loadNextPage()
-
-    while (memberIds.length > 0) {
-      for (let i = 0; i < memberIds.length; i++) {
-        const memberId = memberIds[i]
-        const { documentsIndexed } = await logExecutionTimeV2(
-          () => this.syncMembers(memberId),
-          this.log,
-          `syncMembers (${i}/${memberIds.length})`,
-        )
-
-        docCount += documentsIndexed
-      }
-
-      const diffInSeconds = (new Date().getTime() - now.getTime()) / 1000
-      this.log.info(
-        { organizationId },
-        `Synced ${memberCount} members! Speed: ${Math.round(
-          memberCount / diffInSeconds,
-        )} members/second!`,
+    let docCount = 0
+    for (let i = 0; i < memberIds.length; i++) {
+      const { documentsIndexed } = await logExecutionTimeV2(
+        () => this.syncMembers(memberIds[i]),
+        this.log,
+        `syncMembers (${i + 1}/${memberIds.length})`,
       )
-      memberIds = await loadNextPage(memberIds[memberIds.length - 1])
+      docCount += documentsIndexed
     }
 
+    const lastId = memberIds[memberIds.length - 1]
+
     this.log.info(
-      { organizationId },
-      `Synced total of ${memberCount} members with ${docCount} documents!`,
+      { organizationId, membersSynced: memberIds.length, docCount, lastId },
+      `Synced page of ${memberIds.length} members (${docCount} docs indexed)`,
     )
+
+    return { lastId, membersSynced: memberIds.length }
   }
 
   public async syncMembers(memberId: string): Promise<IMemberSyncResult> {