git.py: Handle concurrent download completions properly

Use os.rename() to rename the cloned temporary repository into
place in the source cache, and issue a STATUS message when discarding
a duplicate clone, in the case where the same repository is cloned
twice in parallel.

The problem with using shutil.move() is that it will create the source
directory in a subdirectory of the destination when the destination
exists, so it's behavior depends on whether the destination exists.

This shutil.move() behavior has so far hidden the race condition
where a duplicate repo is created in a subdirectory, as you need
to have three concurrent downloads of the same repo in order to
trigger the error.

This fixes issue #503

buildstream/plugins/sources/git.py

@@ -71,6 +71,7 @@ git - stage files from a git repository
 """
 
 import os
+import errno
 import re
 import shutil
 from collections import Mapping
@@ -119,11 +120,21 @@ class GitMirror(SourceFetcher):
                                  fail="Failed to clone git repository {}".format(url),
                                  fail_temporarily=True)
 
+                # Attempt atomic rename into destination, this will fail if
+                # another process beat us to the punch
                 try:
-                    shutil.move(tmpdir, self.mirror)
-                except (shutil.Error, OSError) as e:
-                    raise SourceError("{}: Failed to move cloned git repository {} from '{}' to '{}'"
-                                      .format(self.source, url, tmpdir, self.mirror)) from e
+                    os.rename(tmpdir, self.mirror)
+                except OSError as e:
+
+                    # When renaming and the destination repo already exists, os.rename()
+                    # will fail with ENOTEMPTY, since an empty directory will be silently
+                    # replaced
+                    if e.errno == errno.ENOTEMPTY:
+                        self.source.status("{}: Discarding duplicate clone of {}"
+                                           .format(self.source, url))
+                    else:
+                        raise SourceError("{}: Failed to move cloned git repository {} from '{}' to '{}': {}"
+                                          .format(self.source, url, tmpdir, self.mirror, e)) from e
 
     def _fetch(self, alias_override=None):
         url = self.source.translate_url(self.url, alias_override=alias_override)