Fix index file race conditions

lib/Zef/Repository/Ecosystems.rakumod

@@ -118,15 +118,27 @@ class Zef::Repository::Ecosystems does PackageRepository {
     #| see role Repository in lib/Zef.rakumod
     has Int $!update-counter; # Keep track if we already did an update during this runtime
     method update(Supplier :$stdout = Supplier.new, Supplier :$stderr = Supplier.new --> Nil) {
+        self!update-package-list(:$stdout, :$stderr);
+        return;
+    }
+
+    #| Fetch mirrors until one of them provides a package list we can parse, and return that parsed list
+    method !update-package-list(Supplier :$stdout = Supplier.new, Supplier :$stderr = Supplier.new) {
         $!update-counter++;
 
-        $!mirrors.first: -> $uri {
+        return $!mirrors.map(-> $uri {
             # TODO: use the logger to send these as events
             $stderr.emit("===> Updating $!name mirror: $uri");
             UNDO $stderr.emit("!!!> Failed to update $!name mirror: $uri");
             KEEP $stderr.emit("===> Updated $!name mirror: $uri");
 
-            my $save-as  = $!cache.IO.child($uri.IO.basename);
+            # Other zef processes may be updating this mirror at the same time, so never fetch to a shared path.
+            # The fetcher leaves a lock file next to what it fetched, so take a directory of our own to remove.
+            my $stage-at = $!cache.IO.child("{time}.{$*PID}.{(^10000).rand}");
+            LEAVE try delete-paths($stage-at) if $stage-at.e;
+            next unless $stage-at.e || mkdir($stage-at);
+
+            my $save-as  = $stage-at.child($uri.IO.basename);
             my $saved-as = try {
                 CATCH { default { $stderr.emit($_) } }
                 $!fetcher.fetch(Candidate.new(:$uri), $save-as, :timeout(180));
@@ -138,10 +150,14 @@ class Zef::Repository::Ecosystems does PackageRepository {
             $saved-as .= child("{$!name}.json") if $saved-as.d;
             next unless $saved-as.e;
 
-            lock-file-protect("{$saved-as}.lock", -> {
-                self!spurt-package-list($saved-as.slurp(:bin))
-            });
-        }
+            # A mirror can respond with something that is not a package list at all, such as an error page
+            my $content = $saved-as.slurp(:bin);
+            my $parsed  = try Zef::from-json($content.decode);
+            next without $parsed;
+
+            next unless self!spurt-package-list($content);
+            $parsed;
+        }).head;
     }
 
     #| see role Repository in lib/Zef.rakumod
@@ -197,22 +213,19 @@ class Zef::Repository::Ecosystems does PackageRepository {
 
     #| Read our package db
     method !slurp-package-list(--> List) {
-        return [ ] unless self!package-list-path.e;
+        my $path = self!package-list-path;
+        return [ ] unless $path.e;
 
-        do given self!package-list-path.open(:r) {
-            LEAVE {.close}
-            .lock: :shared;
-            try |Zef::from-json(.slurp);
-        }
+        CATCH { default { die "Failed to parse the $!name package list at $path: {.message}" } }
+        |Zef::from-json($path.slurp);
     }
 
-    #| Write our package db
+    #| Write our package db. Readers in other processes only ever see the old or the new file, never a partial one
     method !spurt-package-list($content --> Bool) {
-        do given self!package-list-path.open(:w) {
-            LEAVE {.close}
-            .lock;
-            try .spurt($content);
-        }
+        my $path = self!package-list-path;
+        my $temp = $path.sibling("{$path.basename}.{time}.{$*PID}.{(^10000).rand}");
+        LEAVE try $temp.unlink if $temp.e;
+        so try { $temp.spurt($content); $temp.rename($path) }
     }
 
     #| Check if our package list should be updated
@@ -226,10 +239,13 @@ class Zef::Repository::Ecosystems does PackageRepository {
     has $!populate-distributions-lock = Lock.new;
     method !populate-distributions(--> Nil) {
         $!populate-distributions-lock.protect: {
-            self.update if !$!update-counter && self!is-package-list-stale;
+            # An update parses the package list it fetched, so there is no need to read it back here
+            my $package-list = !$!update-counter && self!is-package-list-stale
+                ?? self!update-package-list
+                !! Nil;
             return if +@!distributions;
 
-            for self!slurp-package-list -> $meta {
+            for @($package-list // self!slurp-package-list) -> $meta {
                 with try Zef::Distribution.new(|%($meta)) -> $dist {
                     # If the distribution doesn't have a name or we can't parse the name then just skip it.
                     next unless $dist.name;

lib/Zef/Utils/FileSystem.rakumod

@@ -161,10 +161,9 @@ module Zef::Utils::FileSystem {
 
     sub lock-file-protect(IO() $path, &code, Bool :$shared = False) is export {
         do given ($shared ?? $path.IO.open(:r) !! $path.IO.open(:w)) {
-            LEAVE {
-                .close;
-                try .path.unlink;
-            }
+            # The lock file outlives this call. Deleting it would hand the lock on a file that no longer has a
+            # name to whoever is waiting for it, while the next caller creates a new file and locks that instead.
+            LEAVE .close;
             .lock(:$shared);
             code();
         }

t/repository-ecosystems.rakutest

@@ -0,0 +1,101 @@
+use v6.d;
+use Test;
+plan 3;
+
+use Zef;
+use Zef::Fetch;
+use Zef::Repository::Ecosystems;
+use Zef::Utils::FileSystem;
+
+
+sub make-mirror(IO() $path, Int $dists = 1) {
+    $path.spurt: Zef::to-json([
+        (^$dists).map({ %( name => "Dist$_", version => '1', auth => 'zef:test', provides => %( "Dist$_" => "lib/Dist$_.rakumod" ) ) })
+    ]);
+    return $path;
+}
+
+
+subtest 'Concurrent updates of the same index' => {
+    my $root = $*TMPDIR.child(100000.rand).mkdir;
+    LEAVE try delete-paths($root);
+
+    # A large index keeps each write slow enough to be observed while it happens
+    my $mirror = make-mirror($root.child('index.json'), 20000);
+    my $cache  = $root.child('store').mkdir;
+    my $index  = $cache.child('test').child('test.json');
+
+    my $update = q:to/END/;
+        use Zef::Fetch;
+        use Zef::Repository::Ecosystems;
+        my ($cache, $mirror) = @*ARGS;
+        my $fetcher = Zef::Fetch.new(backends => [{ module => 'Zef::Service::FetchPath' },]);
+        my $repo = Zef::Repository::Ecosystems.new(:name<test>, :$fetcher, :cache($cache.IO), :auto-update(True), :mirrors[$mirror]);
+        $repo.update for ^3;
+        END
+    my $lib = $?FILE.IO.parent.parent.absolute;
+
+    my $updating = True;
+    my $watcher  = start {
+        my %sizes;
+        while $updating {
+            with try $index.s -> $size { %sizes{$size}++ }
+        }
+        %sizes.keys.map(*.Int).sort.List;
+    }
+    await (^4).map: {
+        Proc::Async.new($*EXECUTABLE.absolute, '-I', $lib, '-e', $update, $cache.absolute, $mirror.absolute).start;
+    }
+    $updating = False;
+
+    is-deeply (await $watcher), ($mirror.s,), 'the index is only ever seen with the complete mirror content';
+    is-deeply $index.slurp(:bin), $mirror.slurp(:bin), 'the index matches the mirror after the updates';
+    is-deeply $index.parent.dir.map(*.basename).List, ('test.json',), 'only the index is left in the repository directory';
+    is-deeply $cache.dir.map(*.basename).List, ('test',), 'nothing the update fetched is left in the cache';
+}
+
+
+subtest 'Index that cannot be parsed' => {
+    my $root = $*TMPDIR.child(100000.rand).mkdir;
+    LEAVE try delete-paths($root);
+
+    my $mirror  = make-mirror($root.child('index.json'));
+    my $cache   = $root.child('store').mkdir;
+    my $fetcher = Zef::Fetch.new(backends => [{ module => 'Zef::Service::FetchPath' },]);
+
+    my $index = $cache.child('test').mkdir.child('test.json');
+    $index.spurt('[{"name": "Dist0", "vers');
+    my $repo = Zef::Repository::Ecosystems.new(:name<test>, :$fetcher, :$cache, :auto-update(0), :mirrors[$mirror.absolute]);
+    throws-like { $repo.search('Dist0', :strict) }, Exception,
+        message  => /'Failed to parse the test package list at ' .* 'test.json'/,
+        'a truncated index is reported instead of being searched as an empty ecosystem';
+    is $index.slurp, '[{"name": "Dist0", "vers', 'the truncated index is left alone when auto-update is off';
+}
+
+
+subtest 'Mirror that does not provide a package list' => {
+    my $root = $*TMPDIR.child(100000.rand).mkdir;
+    LEAVE try delete-paths($root);
+
+    my $error-page = $root.child('error.json');
+    $error-page.spurt('<html><body>504 Gateway Time-out</body></html>');
+    my $mirror  = make-mirror($root.child('index.json'));
+    my $fetcher = Zef::Fetch.new(backends => [{ module => 'Zef::Service::FetchPath' },]);
+
+    {
+        my $cache = $root.child('store-error').mkdir;
+        my $repo  = Zef::Repository::Ecosystems.new(:name<test>, :$fetcher, :$cache, :auto-update(True), :mirrors[$error-page.absolute]);
+        is (try $repo.search('Dist0', :strict).elems), 0, 'a mirror that does not provide a package list gives no results';
+        nok $cache.child('test').child('test.json').e, 'a mirror that does not provide a package list is not saved as the index';
+    }
+
+    {
+        my $cache = $root.child('store-first').mkdir;
+        my $repo  = Zef::Repository::Ecosystems.new(:name<test>, :$fetcher, :$cache, :auto-update(True), :mirrors[$mirror.absolute, $error-page.absolute]);
+        is (try $repo.search('Dist0', :strict).elems), 1, 'the first mirror that provides a package list is used';
+        is-deeply $cache.child('test').child('test.json').slurp(:bin), $mirror.slurp(:bin), 'later mirrors do not replace the index of the one that was used';
+    }
+}
+
+
+done-testing;

t/utils-filesystem.rakutest

@@ -1,6 +1,6 @@
 use v6.d;
 use Test;
-plan 4;
+plan 5;
 
 use Zef::Utils::FileSystem;
 
@@ -169,6 +169,64 @@ subtest "list-paths and delete-paths :f :r" => {
     }
 }
 
+
+
+subtest "lock-file-protect only lets one process at a time into the protected code" => {
+    temp $save-to = $save-to.child(++$dir-id);
+    mkdir $save-to;
+    LEAVE try delete-paths($save-to);
+
+    my $lock   = $save-to.child('shared.lock');
+    my $worker = q:to/END/;
+        use Zef::Utils::FileSystem;
+        my ($lock, $acquired, $release) = @*ARGS;
+        lock-file-protect($lock, -> {
+            $acquired.IO.spurt('');
+            sleep 0.01 until $release.IO.e;
+        });
+        END
+    my $lib = $?FILE.IO.parent.parent.absolute;
+
+    my sub start-worker($name) {
+        my %worker =
+            acquired => $save-to.child("$name.acquired"),
+            release  => $save-to.child("$name.release");
+        %worker<done> = Proc::Async.new(
+            $*EXECUTABLE.absolute, '-I', $lib, '-e', $worker,
+            $lock.absolute, %worker<acquired>.absolute, %worker<release>.absolute,
+        ).start;
+        return %worker;
+    }
+    my sub took-lock(%worker, $seconds = 60) {
+        my $give-up = now + $seconds;
+        sleep 0.01 until %worker<acquired>.e || now > $give-up;
+        return %worker<acquired>.e;
+    }
+    my sub release(%worker) {
+        %worker<release>.spurt('');
+        await %worker<done>;
+    }
+
+    my %first = start-worker('first');
+    ok took-lock(%first), 'the first process takes the lock';
+
+    my %second = start-worker('second');
+    sleep 2;
+    nok %second<acquired>.e, 'a second process waits while the first holds the lock';
+
+    release(%first);
+    ok took-lock(%second), 'the second process takes the lock once the first is done';
+
+    my %third = start-worker('third');
+    sleep 2;
+    nok %third<acquired>.e, 'a process that arrives after an earlier holder exited still waits';
+
+    release(%second);
+    ok took-lock(%third), 'the third process takes the lock once the second is done';
+    release(%third);
+}
+
+
 try rmdir($save-to);