fix: possible race condition where multiple organizations could be created

controlplane/src/core/controllers/auth.ts

@@ -18,6 +18,7 @@ import { AuthenticationError } from '../errors/errors.js';
 import { OrganizationInvitationRepository } from '../repositories/OrganizationInvitationRepository.js';
 import { DefaultNamespace, NamespaceRepository } from '../repositories/NamespaceRepository.js';
 import { OrganizationGroupRepository } from '../repositories/OrganizationGroupRepository.js';
+import { runLocking } from '../util.js';
 
 export type AuthControllerOptions = {
   db: PostgresJsDatabase<typeof schema>;
@@ -255,11 +256,15 @@ const plugin: FastifyPluginCallback<AuthControllerOptions> = function Auth(fasti
           return insertedSessions[0];
         });
 
-        const orgs = await opts.organizationRepository.memberships({
-          userId,
-        });
+        const userOrganizations = await runLocking(userId, async () => {
+          const orgs = await opts.organizationRepository.memberships({
+            userId,
+          });
+
+          if (orgs.length > 0) {
+            return orgs;
+          }
 
-        if (orgs.length === 0) {
           await opts.keycloakClient.authenticateClient();
 
           const organizationSlug = uid(8);
@@ -319,7 +324,9 @@ const plugin: FastifyPluginCallback<AuthControllerOptions> = function Auth(fasti
             user_first_name: firstName,
             user_last_name: lastName,
           });
-        }
+
+          return [];
+        });
 
         // Create a JWT token containing the session id and user id.
         const jwt = await encrypt<UserSession>({
@@ -343,7 +350,7 @@ const plugin: FastifyPluginCallback<AuthControllerOptions> = function Auth(fasti
           } else {
             res.redirect(opts.webBaseUrl);
           }
-        } else if (orgs.length === 0) {
+        } else if (userOrganizations.length === 0) {
           res.redirect(opts.webBaseUrl + '?migrate=true');
         } else {
           res.redirect(opts.webBaseUrl);

controlplane/src/core/util.ts

@@ -42,6 +42,8 @@ const schemaTagRegex = /^(?![/-])[\d/A-Za-z-]+(?<![/-])$/;
 const graphNameRegex = /^[\dA-Za-z]+(?:[./@_-][\dA-Za-z]+)*$/;
 const pluginVersionRegex = /^v\d+$/;
 
+const lockedPromises = new Map<string, Promise<unknown>>();
+
 /**
  * Wraps a function with a try/catch block and logs any errors that occur.
  * If the error is a public error, it is returned as a response message.
@@ -645,3 +647,23 @@ export function newCompositionOptions(disableResolvabilityValidation?: boolean):
     disableResolvabilityValidation,
   };
 }
+
+export function runLocking<TResult>(key: string, action: () => Promise<TResult>): Promise<TResult> {
+  const existingPromise = lockedPromises.get(key);
+  if (existingPromise) {
+    return existingPromise as Promise<TResult>;
+  }
+
+  const promise = (async () => await action())();
+  lockedPromises.set(key, promise);
+
+  promise.finally(() => {
+    const other = lockedPromises.get(key);
+    if (other === promise) {
+      // Remove only if we're still pointing at the same promise (avoid race).
+      lockedPromises.delete(key);
+    }
+  });
+
+  return promise;
+}

controlplane/test/test-util.ts

@@ -975,3 +975,17 @@ export const resolvabilitySDLTwo = `
     name: String!
   }
 `;
+
+export function deferred<T = void>() {
+  let resolve!: (v: T | PromiseLike<T>) => void;
+  let reject!: (e?: unknown) => void;
+
+  const promise = new Promise<T>((_resolve, _reject) => { resolve = _resolve; reject = _reject; });
+  return { promise, resolve, reject };
+}
+
+export async function expectPending(p: Promise<unknown>) {
+  const timeout = new Promise((resolve) => setTimeout(() => resolve("timeout"), 20));
+  const race = await Promise.race([p.then(() => "resolved" as const), timeout]);
+  expect(race).toBe("timeout");
+}

controlplane/test/utils.test.ts

@@ -1,5 +1,12 @@
-import { describe, expect, test } from 'vitest';
-import { isValidLabelMatchers, mergeUrls, normalizeLabelMatchers, isGoogleCloudStorageUrl } from '../src/core/util.js';
+import { afterEach, beforeEach, describe, expect, test, vi } from 'vitest';
+import {
+  isValidLabelMatchers,
+  mergeUrls,
+  normalizeLabelMatchers,
+  isGoogleCloudStorageUrl,
+  runLocking
+} from '../src/core/util.js';
+import { deferred, expectPending } from "./test-util.js";
 
 describe('Utils', () => {
   test('isValidLabelMatchers', () => {
@@ -49,4 +56,97 @@ describe('Utils', () => {
       expect(isGoogleCloudStorageUrl('https://storage.googleapis.com.evil.com')).toBe(false);
     });
   });
+
+  describe('runLocking', () => {
+    beforeEach(() => vi.useFakeTimers());
+    afterEach(() => vi.useRealTimers());
+
+    test('that single call returns without locking', async () => {
+      const result = await runLocking('test', () => Promise.resolve('hello world'));
+      expect(result).toBe('hello world');
+    });
+
+    test('coalesces concurrent calls for the SAME key', async () => {
+      const gate = deferred<void>();
+
+      const worker = vi.fn(async () => {
+        await gate.promise; // hold until we release the gate
+        return "OK";
+      });
+
+      // Start two concurrent calls with the same key
+      const p1 = runLocking("user:42", worker);
+      const p2 = runLocking("user:42", worker);
+
+      // Both callers must share the exact same Promise
+      expect(p1).toBe(p2);
+
+      // Underlying work should have started only once
+      expect(worker).toHaveBeenCalledTimes(1);
+
+      // Let it finish and verify both callers get the same result
+      gate.resolve();
+
+      const [r1, r2] = await Promise.all([p1, p2]);
+      expect(r1).toBe("OK");
+      expect(r2).toBe("OK");
+    });
+
+    test('does NOT lock across DIFFERENT keys (independent execution)', async () => {
+      const gateA = deferred<void>();
+      const gateB = deferred<void>();
+
+      const worker = vi.fn(async (val: string, gate: ReturnType<typeof deferred<void>>) => {
+        await gate.promise;
+        return val;
+      });
+
+      const pA = runLocking("user:A", () => worker("A", gateA));
+      const pB = runLocking("user:B", () => worker("B", gateB));
+
+      // Different keys should produce different Promises and start separate work
+      expect(pA).not.toBe(pB);
+      expect(worker).toHaveBeenCalledTimes(2);
+
+      // Resolve B first; A should still be pending
+      gateB.resolve();
+
+      const rB = await pB;
+      expect(rB).toBe("B");
+
+      const p = expectPending(pA); // A hasn't been released yet
+      await vi.advanceTimersByTimeAsync(100);
+
+      await p;
+
+      // Now resolve A
+      gateA.resolve();
+      const rA = await pA;
+      expect(rA).toBe("A");
+    })
+  });
+
+  test('starts fresh after a call completes (entry cleanup)', async () => {
+    // One worker whose gate we can swap between runs
+    let gate = deferred<void>();
+    const worker = vi.fn(async () => {
+      await gate.promise;
+      return "done";
+    });
+
+    // First run
+    const p1 = runLocking("user:99", worker);
+    gate.resolve();
+    await p1;
+
+    // Swap to a new gate and call again — should start a NEW underlying run
+    gate = deferred<void>();
+
+    const p2 = runLocking("user:99", worker);
+    expect(p2).not.toBe(p1); // new promise after cleanup
+    expect(worker).toHaveBeenCalledTimes(2);
+
+    gate.resolve();
+    await p2;
+  })
 });