fix: prevent concurrent mailing list connects from racing ownership (CM-1318)

Signed-off-by: Uroš Marolt <uros@marolt.me>

Author: themarolt

backend/src/services/integrationService.ts

@@ -18,6 +18,7 @@ import { CommonIntegrationService, getGithubInstallationToken } from '@crowd/com
 import { ICreateInsightsProject } from '@crowd/data-access-layer/src/collections'
 import {
   findMailingListsOwnedByOtherIntegration,
+  lockMailingListSourceUrls,
   upsertMailingLists,
 } from '@crowd/data-access-layer/src/mailinglist'
 import {
@@ -1467,6 +1468,14 @@ export default class IntegrationService {
         options,
       )
 
+      // Serialize concurrent connects touching the same sourceUrl(s) so the
+      // ownership check below and the upsert that follows it can't be
+      // straddled by another transaction re-pointing ownership in between.
+      await lockMailingListSourceUrls(
+        qx,
+        lists.map((l) => l.sourceUrl),
+      )
+
       const conflicts = await findMailingListsOwnedByOtherIntegration(qx, integration.id, lists)
       if (conflicts.length > 0) {
         throw new Error400(

services/libs/data-access-layer/src/mailinglist/index.ts

@@ -5,6 +5,30 @@ export interface IMailingListToOnboard {
   sourceUrl: string
 }
 
+/**
+ * Take a transaction-scoped advisory lock per sourceUrl, so two concurrent
+ * connect/update calls touching the same sourceUrl serialize instead of both
+ * passing `findMailingListsOwnedByOtherIntegration`'s check and then racing
+ * in `upsertMailingLists`'s `ON CONFLICT DO UPDATE` (which silently
+ * reassigns ownership to whichever transaction commits last). Locks release
+ * automatically at commit/rollback. Call before the ownership check, on a
+ * `qx` bound to the same transaction as the rest of the connect flow.
+ * @param qx - Query executor (must be transactional)
+ * @param sourceUrls - sourceUrls about to be checked/upserted
+ */
+export async function lockMailingListSourceUrls(
+  qx: QueryExecutor,
+  sourceUrls: string[],
+): Promise<void> {
+  // Sorted so overlapping concurrent requests always acquire locks in the
+  // same order, avoiding a deadlock between two multi-list connects.
+  for (const sourceUrl of [...sourceUrls].sort()) {
+    await qx.selectNone(`SELECT pg_advisory_xact_lock(hashtext($(sourceUrl))::bigint)`, {
+      sourceUrl,
+    })
+  }
+}
+
 /**
  * Find sourceUrls already onboarded under a different, non-deleted
  * integration. Used to reject a connect/update call before it silently