Fix: Concurrency issues when cloning repository (#45)

package-lock.json

@@ -681,6 +681,12 @@
       "integrity": "sha512-2cd51m3i0yeY1i3dKxcqJKeS5Q4jZnjP37OseoNeIX1OM0AhmGPuuYmwJ9OqtsU35YrREQxdb2VeX5sM3cwGMQ==",
       "dev": true
     },
+    "@types/events": {
+      "version": "3.0.0",
+      "resolved": "https://registry.npmjs.org/@types/events/-/events-3.0.0.tgz",
+      "integrity": "sha512-EaObqwIvayI5a8dCzhFrjKzVwKLxjoG9T6Ppd5CEo07LRKfQ8Yokw54r5+Wq7FaBQ+yXRvQAYPrHwya1/UFt9g==",
+      "dev": true
+    },
     "@types/express": {
       "version": "4.16.1",
       "resolved": "https://registry.npmjs.org/@types/express/-/express-4.16.1.tgz",
@@ -702,6 +708,17 @@
         "@types/range-parser": "*"
       }
     },
+    "@types/glob": {
+      "version": "7.1.1",
+      "resolved": "https://registry.npmjs.org/@types/glob/-/glob-7.1.1.tgz",
+      "integrity": "sha512-1Bh06cbWJUHMC97acuD6UMG29nMt0Aqz1vF3guLfG+kHHJhy3AyohZFFxYk2f7Q1SQIrNwvncxAE0N/9s70F2w==",
+      "dev": true,
+      "requires": {
+        "@types/events": "*",
+        "@types/minimatch": "*",
+        "@types/node": "*"
+      }
+    },
     "@types/helmet": {
       "version": "0.0.43",
       "resolved": "https://registry.npmjs.org/@types/helmet/-/helmet-0.0.43.tgz",
@@ -757,6 +774,12 @@
       "integrity": "sha512-FwI9gX75FgVBJ7ywgnq/P7tw+/o1GUbtP0KzbtusLigAOgIgNISRK0ZPl4qertvXSIE8YbsVJueQ90cDt9YYyw==",
       "dev": true
     },
+    "@types/minimatch": {
+      "version": "3.0.3",
+      "resolved": "https://registry.npmjs.org/@types/minimatch/-/minimatch-3.0.3.tgz",
+      "integrity": "sha512-tHq6qdbT9U1IRSGf14CL0pUlULksvY9OZ+5eEgl1N7t+OA3tGvNpxJCzuKQlsNgCVwbAs670L1vcVQi8j9HjnA==",
+      "dev": true
+    },
     "@types/node": {
       "version": "12.0.2",
       "resolved": "https://registry.npmjs.org/@types/node/-/node-12.0.2.tgz",
@@ -786,6 +809,16 @@
       "integrity": "sha512-ewFXqrQHlFsgc09MK5jP5iR7vumV/BYayNC6PgJO2LPe8vrnNFyjQjSppfEngITi0qvfKtzFvgKymGheFM9UOA==",
       "dev": true
     },
+    "@types/rimraf": {
+      "version": "2.0.2",
+      "resolved": "https://registry.npmjs.org/@types/rimraf/-/rimraf-2.0.2.tgz",
+      "integrity": "sha512-Hm/bnWq0TCy7jmjeN5bKYij9vw5GrDFWME4IuxV08278NtU/VdGbzsBohcCUJ7+QMqmUq5hpRKB39HeQWJjztQ==",
+      "dev": true,
+      "requires": {
+        "@types/glob": "*",
+        "@types/node": "*"
+      }
+    },
     "@types/serve-static": {
       "version": "1.13.2",
       "resolved": "https://registry.npmjs.org/@types/serve-static/-/serve-static-1.13.2.tgz",

package.json

@@ -15,7 +15,7 @@
     "test:ci": "jest --ci --reporters=default --reporters=jest-junit",
     "lint": "tslint -p tsconfig.json",
     "test:watch": "jest --watch --coverage=false --config ./config/jest.dev.config.js",
-    "test:e2e": "jest --config ./config/jest.e2e.config.js",
+    "test:e2e": "jest --config ./config/jest.e2e.config.js --runInBand",
     "swagger:gen": "node ./bin/scripts/generate-swagger-specs.js",
     "autorest": "cross-var autorest sdk/config.yaml --package-version=$npm_package_version",
     "sdk:gen": "npm run sdk:gen:ts",
@@ -38,6 +38,7 @@
     "@types/jest": "^24.0.13",
     "@types/node-fetch": "^2.3.4",
     "@types/nodegit": "^0.24.6",
+    "@types/rimraf": "^2.0.2",
     "@types/triple-beam": "^1.3.0",
     "autorest": "^2.0.4283",
     "concurrently": "^4.1.0",
@@ -46,6 +47,7 @@
     "jest": "^24.8.0",
     "jest-junit": "^6.4.0",
     "prettier": "^1.17.1",
+    "rimraf": "^2.6.3",
     "swagger-ui-express": "^4.0.3",
     "ts-jest": "^24.0.2",
     "tslint": "^5.16.0",

src/controllers/branches/branches.controller.e2e.ts

@@ -1,10 +1,50 @@
-import { TEST_REPO, e2eClient } from "../../../test/e2e";
+import { TEST_REPO, UNENCODED_TEST_REPO, deleteLocalRepo, e2eClient } from "../../../test/e2e";
+import { delay } from "../../utils";
 
 describe("Test branch controller", () => {
+  it("doesn't conflict when getting the same repo twice at the same time", async () => {
+    await delay(100);
+    await deleteLocalRepo(UNENCODED_TEST_REPO);
+    await delay(100);
+    const responses = await Promise.all([
+      e2eClient.fetch(`/repos/${TEST_REPO}/branches`),
+      // Delay a little to start the clone
+      delay(10).then(() => e2eClient.fetch(`/repos/${TEST_REPO}/branches`)),
+      delay(20).then(() => e2eClient.fetch(`/repos/${TEST_REPO}/branches`)),
+      delay(30).then(() => e2eClient.fetch(`/repos/${TEST_REPO}/branches`)),
+      delay(40).then(() => e2eClient.fetch(`/repos/${TEST_REPO}/branches`)),
+      delay(50).then(() => e2eClient.fetch(`/repos/${TEST_REPO}/branches`)),
+    ]);
+
+    for (const response of responses) {
+      expect(response.status).toEqual(200);
+    }
+
+    const [first, ...others] = await Promise.all(responses.map(x => x.json() as Promise<any[]>));
+    for (const content of others) {
+      expect(sortBy(content, x => x.name)).toEqual(sortBy(first, x => x.name));
+    }
+  });
+
   it("List branches in the test repo", async () => {
     const response = await e2eClient.fetch(`/repos/${TEST_REPO}/branches`);
     expect(response.status).toEqual(200);
-    const content = await response.json();
-    expect(content).toMatchPayload("branches_list");
+    const content: any[] = await response.json();
+    expect(sortBy(content, x => x.name)).toMatchPayload("branches_list");
   });
 });
+
+function sortBy<T>(array: T[], attr: (item: T) => any): T[] {
+  return array.sort((a, b) => {
+    const aAttr = attr(a);
+    const bAttr = attr(b);
+
+    if (aAttr < bAttr) {
+      return -1;
+    } else if (aAttr > bAttr) {
+      return 1;
+    } else {
+      return 0;
+    }
+  });
+}

src/main.ts

@@ -1,4 +1,5 @@
 import { createApp } from "./app";
+import { Logger } from "./core";
 
 async function bootstrap() {
   const { app } = await createApp();
@@ -6,7 +7,8 @@ async function bootstrap() {
 }
 
 bootstrap().catch(error => {
+  const logger = new Logger("Bootstrap");
   // tslint:disable-next-line: no-console
-  console.error("Error in app", error);
+  logger.error("Error in app", error);
   process.exit(1);
 });

src/services/branch/branch.service.test.ts

@@ -31,7 +31,7 @@ describe("BranchService", () => {
   };
 
   const repoServiceSpy = {
-    get: jest.fn(() => mockRepo),
+    use: jest.fn((_, _1, action) => action(mockRepo)),
   };
 
   beforeEach(() => {
@@ -42,8 +42,8 @@ describe("BranchService", () => {
   it("List the branches", async () => {
     const branches = await service.list("github.com/Azure/git-rest-api");
 
-    expect(repoServiceSpy.get).toHaveBeenCalledTimes(1);
-    expect(repoServiceSpy.get).toHaveBeenCalledWith("github.com/Azure/git-rest-api", {});
+    expect(repoServiceSpy.use).toHaveBeenCalledTimes(1);
+    expect(repoServiceSpy.use).toHaveBeenCalledWith("github.com/Azure/git-rest-api", {}, expect.any(Function));
     expect(mockRepo.getReferences).toHaveBeenCalledTimes(1);
     expect(branches).toEqual([b1, b2]);
   });

src/services/branch/branch.service.ts

@@ -1,5 +1,5 @@
 import { Injectable } from "@nestjs/common";
-import { Reference } from "nodegit";
+import { Reference, Repository } from "nodegit";
 
 import { GitBranch } from "../../dtos";
 import { GitBaseOptions, RepoService } from "../repo";
@@ -13,7 +13,12 @@ export class BranchService {
    * @param remote
    */
   public async list(remote: string, options: GitBaseOptions = {}): Promise<GitBranch[]> {
-    const repo = await this.repoService.get(remote, options);
+    return this.repoService.use(remote, options, async repo => {
+      return this.listGitBranches(repo);
+    });
+  }
+
+  public async listGitBranches(repo: Repository): Promise<GitBranch[]> {
     const refs = await repo.getReferences(Reference.TYPE.LISTALL);
     const branches = refs.filter(x => x.isRemote());
 

src/services/commit/commit.service.ts

@@ -21,7 +21,7 @@ export class CommitService {
     remote: string,
     options: ListCommitsOptions & GitBaseOptions = {},
   ): Promise<PaginatedList<GitCommit> | NotFoundException> {
-    const repo = await this.repoService.get(remote, options);
+    return this.repoService.use(remote, options, async repo => {
       const commits = await this.listCommits(repo, options);
       if (commits instanceof NotFoundException) {
         return commits;
@@ -32,15 +32,17 @@ export class CommitService {
         ...commits,
         items,
       };
+    });
   }
 
   public async get(remote: string, commitSha: string, options: GitBaseOptions = {}): Promise<GitCommit | undefined> {
-    const repo = await this.repoService.get(remote, options);
+    return this.repoService.use(remote, options, async repo => {
       const commit = await this.getCommit(repo, commitSha);
       if (!commit) {
         return undefined;
       }
       return toGitCommit(commit);
+    });
   }
 
   /**

src/services/compare/compare.service.ts

@@ -21,8 +21,7 @@ export class CompareService {
     options: GitBaseOptions = {},
   ): Promise<GitDiff | NotFoundException> {
     const compareRepo = await this.getCompareRepo(remote, base, head, options);
-    const repo = compareRepo.repo;
-
+    return this.repoService.using(compareRepo.repo, async repo => {
       const [baseCommit, headCommit] = await Promise.all([
         this.commitService.getCommit(repo, compareRepo.baseRef),
         this.commitService.getCommit(repo, compareRepo.headRef),
@@ -35,6 +34,7 @@ export class CompareService {
       }
 
       return this.getComparison(repo, baseCommit, headCommit);
+    });
   }
 
   public async getMergeBase(repo: Repository, base: Oid, head: Oid): Promise<Commit | undefined> {

src/services/content/content.service.ts

@@ -1,5 +1,5 @@
 import { Injectable, NotFoundException } from "@nestjs/common";
-import { TreeEntry } from "nodegit";
+import { Repository, TreeEntry } from "nodegit";
 
 import { GitContents } from "../../dtos/git-contents";
 import { GitDirObjectContent } from "../../dtos/git-dir-object-content";
@@ -18,9 +18,13 @@ export class ContentService {
     ref: string | undefined = "master",
     options: GitBaseOptions = {},
   ): Promise<GitContents | NotFoundException> {
-    const repo = await this.repoService.get(remote, options);
-    const commit = await this.commitService.getCommit(repo, ref);
+    return this.repoService.use(remote, options, async repo => {
+      return this.getGitContents(repo, path, ref);
+    });
+  }
 
+  public async getGitContents(repo: Repository, path: string | undefined, ref: string | undefined = "master") {
+    const commit = await this.commitService.getCommit(repo, ref);
     if (!commit) {
       return new NotFoundException(`Ref '${ref}' not found.`);
     }

src/services/repo/repo.service.ts

@@ -25,21 +25,61 @@ export interface RemoteDef {
 
 @Injectable()
 export class RepoService {
+  /**
+   * Map that contains a key and promise when cloning a given repo
+   */
+  private cloningRepos = new Map<string, Promise<Repository>>();
+
   constructor(
     private fs: FSService,
     private fetchService: GitFetchService,
     private permissionService: PermissionService,
   ) {}
 
+  public async use<T>(remote: string, options: GitBaseOptions, action: (repo: Repository) => Promise<T>): Promise<T> {
+    const repo = await this.get(remote, options);
+    return this.using(repo, action);
+  }
+
+  public async using<T>(repo: Repository, action: (repo: Repository) => Promise<T>): Promise<T> {
+    try {
+      const response = await action(repo);
+      repo.cleanup();
+      return response;
+    } catch (error) {
+      repo.cleanup();
+      throw error;
+    }
+  }
+
+  /**
+   * Be carfull with using this one. Repository object needs to be clenup. Make sure its with `using` to ensure it gets cleanup after
+   */
   public async get(remote: string, options: GitBaseOptions = {}): Promise<Repository> {
     await this.validatePermissions([remote], options);
     const repoPath = getRepoMainPath(remote);
 
-    if (await this.fs.exists(repoPath)) {
+    const cloningRepo = this.cloningRepos.get(repoPath);
+    if (cloningRepo) {
+      return cloningRepo;
+    }
+
+    const exists = await this.fs.exists(repoPath);
+    // Check again if the repo didn't start cloning since the last time
+    const isCloningRepo = this.cloningRepos.get(repoPath);
+    if (isCloningRepo) {
+      return isCloningRepo;
+    }
+    if (exists) {
       const repo = await Repository.open(repoPath);
       return this.fetchService.fetch(remote, repo, options);
     } else {
-      return this.fetchService.clone(remote, repoPath, options);
+      const cloneRepoPromise = this.fetchService.clone(remote, repoPath, options).then(repo => {
+        this.cloningRepos.delete(repoPath);
+        return repo;
+      });
+      this.cloningRepos.set(repoPath, cloneRepoPromise);
+      return cloneRepoPromise;
     }
   }
 

src/utils/misc-utils.ts

@@ -1 +1,2 @@
 export const notUndefined = <T>(x: T | undefined): x is T => x !== undefined;
+export const delay = (timeout?: number) => new Promise(r => setTimeout(r, timeout));

test/e2e/index.ts

@@ -1,2 +1,3 @@
 export * from "./e2e-server";
+export * from "./utils";
 import "./custom-matchers";

test/e2e/utils.ts

@@ -0,0 +1,11 @@
+import path from "path";
+import rimraf from "rimraf";
+import { promisify } from "util";
+
+const rm = promisify(rimraf);
+
+const dataFolder = path.resolve(path.join(__dirname, "../../tmp"));
+
+export async function deleteLocalRepo(name: string) {
+  await rm(path.join(dataFolder, "repos", encodeURIComponent(name)));
+}