diff --git a/lib/Zef/Repository/Ecosystems.rakumod b/lib/Zef/Repository/Ecosystems.rakumod index 811a1c49..c9573bc3 100644 --- a/lib/Zef/Repository/Ecosystems.rakumod +++ b/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; diff --git a/lib/Zef/Utils/FileSystem.rakumod b/lib/Zef/Utils/FileSystem.rakumod index 162f80a1..d5a3346d 100644 --- a/lib/Zef/Utils/FileSystem.rakumod +++ b/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(); } diff --git a/t/repository-ecosystems.rakutest b/t/repository-ecosystems.rakutest new file mode 100644 index 00000000..19f3ccc7 --- /dev/null +++ b/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, :$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, :$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('504 Gateway Time-out'); + 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, :$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, :$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; diff --git a/t/utils-filesystem.rakutest b/t/utils-filesystem.rakutest index f781e478..58253e29 100644 --- a/t/utils-filesystem.rakutest +++ b/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 = Proc::Async.new( + $*EXECUTABLE.absolute, '-I', $lib, '-e', $worker, + $lock.absolute, %worker.absolute, %worker.absolute, + ).start; + return %worker; + } + my sub took-lock(%worker, $seconds = 60) { + my $give-up = now + $seconds; + sleep 0.01 until %worker.e || now > $give-up; + return %worker.e; + } + my sub release(%worker) { + %worker.spurt(''); + await %worker; + } + + my %first = start-worker('first'); + ok took-lock(%first), 'the first process takes the lock'; + + my %second = start-worker('second'); + sleep 2; + nok %second.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.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);