fix: prevent deadlocks in member affiliation updates (#3305)

Author: skwowet

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

@@ -27,21 +27,47 @@ export async function deleteMember(memberId: string): Promise<void> {
 export async function recalculateActivityAffiliationsOfMemberAsync(
   memberId: string,
 ): Promise<void> {
-  await svc.temporal.workflow.start('memberUpdate', {
-    taskQueue: 'profiles',
-    workflowId: `${TemporalWorkflowId.MEMBER_UPDATE}/${memberId}`,
-    workflowIdReusePolicy: WorkflowIdReusePolicy.WORKFLOW_ID_REUSE_POLICY_TERMINATE_IF_RUNNING,
-    retry: {
-      maximumAttempts: 10,
-    },
-    args: [
-      {
-        member: {
-          id: memberId,
-        },
+  const workflowId = `${TemporalWorkflowId.MEMBER_UPDATE}/${memberId}`
+
+  try {
+    const handle = svc.temporal.workflow.getHandle(workflowId)
+    const { status } = await handle.describe()
+
+    if (status.name === 'RUNNING') {
+      await handle.result()
+    }
+  } catch (err) {
+    if (err.name !== 'WorkflowNotFoundError') {
+      svc.log.error({ err }, 'Failed to check workflow state')
+      throw err
+    }
+  }
+
+  try {
+    await svc.temporal.workflow.start('memberUpdate', {
+      taskQueue: 'profiles',
+      workflowId,
+      // if the workflow is already running, this policy will throw an error
+      workflowIdReusePolicy: WorkflowIdReusePolicy.REJECT_DUPLICATE,
+      retry: {
+        maximumAttempts: 10,
       },
-    ],
-  })
+      args: [
+        {
+          member: {
+            id: memberId,
+          },
+        },
+      ],
+    })
+  } catch (err) {
+    if (err.name === 'WorkflowExecutionAlreadyStartedError') {
+      svc.log.info({ workflowId }, 'Workflow already started, skipping')
+      return
+    }
+
+    throw err
+  }
 }
 
 export async function syncMember(memberId: string): Promise<void> {

services/apps/profiles_worker/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@crowd/profiles-worker",
   "scripts": {
-    "start": "CROWD_TEMPORAL_TASKQUEUE=profiles SERVICE=profiles-worker tsx src/main.ts",
+    "start": "CROWD_TEMPORAL_TASKQUEUE=profiles SERVICE=profiles-worker LOG_LEVEL=debug tsx src/main.ts",
     "start:debug:local": "set -a && . ../../../backend/.env.dist.local && . ../../../backend/.env.override.local && set +a && CROWD_TEMPORAL_TASKQUEUE=profiles SERVICE=profiles-worker LOG_LEVEL=trace tsx --inspect=0.0.0.0:9232 src/main.ts",
     "start:debug": "CROWD_TEMPORAL_TASKQUEUE=profiles SERVICE=profiles-worker LOG_LEVEL=trace tsx --inspect=0.0.0.0:9232 src/main.ts",
     "dev:local": "nodemon --watch src --watch ../../libs --ext ts --exec pnpm run start:debug:local",