From 865291ed892b78c05b794040d23a4d0154424d95 Mon Sep 17 00:00:00 2001 From: Nick Logan Date: Mon, 14 Sep 2026 13:23:15 -0700 Subject: [PATCH 1/4] Avoid partial ecosystem index reads during concurrent updates Previously an update truncated the index file before locking it, and every updating process fetched the mirror to the same path in the store. When several zef processes started on a fresh store at the same time, such as the parallel installs Blin runs, some of them read an empty or partially written index. The parse error was swallowed, so those processes reported "No candidates found matching identity" for distributions that exist. This fetches each mirror to a unique path that is removed afterwards, and writes the index to a unique temporary file that is then renamed over the index. Readers now see either the previous index or the complete new one, so the lock files are no longer needed. --- lib/Zef/Repository/Ecosystems.rakumod | 25 ++++++-------- t/repository-ecosystems.rakutest | 47 +++++++++++++++++++++++++++ 2 files changed, 57 insertions(+), 15 deletions(-) create mode 100644 t/repository-ecosystems.rakutest diff --git a/lib/Zef/Repository/Ecosystems.rakumod b/lib/Zef/Repository/Ecosystems.rakumod index 811a1c49..2ea44159 100644 --- a/lib/Zef/Repository/Ecosystems.rakumod +++ b/lib/Zef/Repository/Ecosystems.rakumod @@ -126,7 +126,9 @@ class Zef::Repository::Ecosystems does PackageRepository { 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 + my $save-as = $!cache.IO.child("{$uri.IO.basename}.{time}.{$*PID}.{(^10000).rand}"); + LEAVE try delete-paths($save-as) if $save-as.e; my $saved-as = try { CATCH { default { $stderr.emit($_) } } $!fetcher.fetch(Candidate.new(:$uri), $save-as, :timeout(180)); @@ -138,9 +140,7 @@ 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)) - }); + next unless self!spurt-package-list($saved-as.slurp(:bin)); } } @@ -199,20 +199,15 @@ class Zef::Repository::Ecosystems does PackageRepository { method !slurp-package-list(--> List) { return [ ] unless self!package-list-path.e; - do given self!package-list-path.open(:r) { - LEAVE {.close} - .lock: :shared; - try |Zef::from-json(.slurp); - } + try |Zef::from-json(self!package-list-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 diff --git a/t/repository-ecosystems.rakutest b/t/repository-ecosystems.rakutest new file mode 100644 index 00000000..642174b0 --- /dev/null +++ b/t/repository-ecosystems.rakutest @@ -0,0 +1,47 @@ +use v6.d; +use Test; +plan 1; + +use Zef; +use Zef::Fetch; +use Zef::Repository::Ecosystems; +use Zef::Utils::FileSystem; + + +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 = $root.child('index.json'); + $mirror.spurt: Zef::to-json([ + (^20000).map({ %( name => "Dist$_", version => '1', auth => 'zef:test', provides => %( "Dist$_" => "lib/Dist$_.rakumod" ) ) }) + ]); + my $cache = $root.child('store').mkdir; + my $fetcher = Zef::Fetch.new(backends => [{ module => 'Zef::Service::FetchPath' },]); + my $index = $cache.child('test').child('test.json'); + + 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: { + start { + my $repo = Zef::Repository::Ecosystems.new(:name, :$fetcher, :$cache, :auto-update(0), :mirrors[$mirror.absolute]); + $repo.update for ^5; + } + } + $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.grep(*.f).List, (), 'no fetched mirror files are left in the cache'; +} + + +done-testing; From ec552dbd407ca3e2b7990ad2f9307cfc10659e27 Mon Sep 17 00:00:00 2001 From: Nick Logan Date: Mon, 14 Sep 2026 13:24:13 -0700 Subject: [PATCH 2/4] Report an ecosystem index that cannot be parsed Previously an index that failed to parse was searched as an empty ecosystem, so an install reported "No candidates found matching identity" for distributions that exist. This reports the index and the parse error instead. --- lib/Zef/Repository/Ecosystems.rakumod | 6 ++++-- t/repository-ecosystems.rakutest | 21 ++++++++++++++++++++- 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/lib/Zef/Repository/Ecosystems.rakumod b/lib/Zef/Repository/Ecosystems.rakumod index 2ea44159..8cb79e57 100644 --- a/lib/Zef/Repository/Ecosystems.rakumod +++ b/lib/Zef/Repository/Ecosystems.rakumod @@ -197,9 +197,11 @@ 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; - try |Zef::from-json(self!package-list-path.slurp); + CATCH { default { die "Failed to parse the $!name package list at $path: {.message}" } } + |Zef::from-json($path.slurp); } #| Write our package db. Readers in other processes only ever see the old or the new file, never a partial one diff --git a/t/repository-ecosystems.rakutest b/t/repository-ecosystems.rakutest index 642174b0..03e1d520 100644 --- a/t/repository-ecosystems.rakutest +++ b/t/repository-ecosystems.rakutest @@ -1,6 +1,6 @@ use v6.d; use Test; -plan 1; +plan 2; use Zef; use Zef::Fetch; @@ -44,4 +44,23 @@ subtest 'Concurrent updates of the same index' => { } +subtest 'Index that cannot be parsed' => { + my $root = $*TMPDIR.child(100000.rand).mkdir; + LEAVE try delete-paths($root); + + my $mirror = $root.child('index.json'); + $mirror.spurt: Zef::to-json([%( name => 'Dist42', version => '1', auth => 'zef:test', provides => %( Dist42 => 'lib/Dist42.rakumod' ) ),]); + 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": "Dist42", "vers'); + my $repo = Zef::Repository::Ecosystems.new(:name, :$fetcher, :$cache, :auto-update(0), :mirrors[$mirror.absolute]); + throws-like { $repo.search('Dist42', :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": "Dist42", "vers', 'the truncated index is left alone when auto-update is off'; +} + + done-testing; From 22be213896213ac713a4c37caca4f6fc9c58e796 Mon Sep 17 00:00:00 2001 From: Nick Logan Date: Thu, 17 Sep 2026 08:06:54 -0700 Subject: [PATCH 3/4] Only replace the index with a package list we can parse Previously an update saved whatever a mirror responded with, so an error page or a truncated response replaced a good index, and every following install failed until the next update. A mirror is now parsed before it replaces the index, and a mirror that does not parse is passed over for the next one. The iteration also kept going after a mirror was saved, leaving the last mirror as the index rather than the first one that worked. It now stops at the first mirror that provides a package list. Populating the distributions reuses the package list that the update parsed instead of reading the index back. --- lib/Zef/Repository/Ecosystems.rakumod | 25 ++++++++-- t/repository-ecosystems.rakutest | 69 ++++++++++++++++++++------- 2 files changed, 72 insertions(+), 22 deletions(-) diff --git a/lib/Zef/Repository/Ecosystems.rakumod b/lib/Zef/Repository/Ecosystems.rakumod index 8cb79e57..510e2aea 100644 --- a/lib/Zef/Repository/Ecosystems.rakumod +++ b/lib/Zef/Repository/Ecosystems.rakumod @@ -118,9 +118,15 @@ 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"); @@ -140,8 +146,14 @@ class Zef::Repository::Ecosystems does PackageRepository { $saved-as .= child("{$!name}.json") if $saved-as.d; next unless $saved-as.e; - next unless 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 @@ -223,10 +235,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/t/repository-ecosystems.rakutest b/t/repository-ecosystems.rakutest index 03e1d520..a5af4cc4 100644 --- a/t/repository-ecosystems.rakutest +++ b/t/repository-ecosystems.rakutest @@ -1,6 +1,6 @@ use v6.d; use Test; -plan 2; +plan 3; use Zef; use Zef::Fetch; @@ -8,18 +8,32 @@ 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 = $root.child('index.json'); - $mirror.spurt: Zef::to-json([ - (^20000).map({ %( name => "Dist$_", version => '1', auth => 'zef:test', provides => %( "Dist$_" => "lib/Dist$_.rakumod" ) ) }) - ]); - my $cache = $root.child('store').mkdir; - my $fetcher = Zef::Fetch.new(backends => [{ module => 'Zef::Service::FetchPath' },]); - my $index = $cache.child('test').child('test.json'); + 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 { @@ -30,10 +44,7 @@ subtest 'Concurrent updates of the same index' => { %sizes.keys.map(*.Int).sort.List; } await (^4).map: { - start { - my $repo = Zef::Repository::Ecosystems.new(:name, :$fetcher, :$cache, :auto-update(0), :mirrors[$mirror.absolute]); - $repo.update for ^5; - } + Proc::Async.new($*EXECUTABLE.absolute, '-I', $lib, '-e', $update, $cache.absolute, $mirror.absolute).start; } $updating = False; @@ -48,18 +59,42 @@ subtest 'Index that cannot be parsed' => { my $root = $*TMPDIR.child(100000.rand).mkdir; LEAVE try delete-paths($root); - my $mirror = $root.child('index.json'); - $mirror.spurt: Zef::to-json([%( name => 'Dist42', version => '1', auth => 'zef:test', provides => %( Dist42 => 'lib/Dist42.rakumod' ) ),]); + 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": "Dist42", "vers'); + $index.spurt('[{"name": "Dist0", "vers'); my $repo = Zef::Repository::Ecosystems.new(:name, :$fetcher, :$cache, :auto-update(0), :mirrors[$mirror.absolute]); - throws-like { $repo.search('Dist42', :strict) }, Exception, + 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": "Dist42", "vers', 'the truncated index is left alone when auto-update is off'; + 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'; + } } From 2a535314377a49a2d190fb742a2a28622fd5305d Mon Sep 17 00:00:00 2001 From: Nick Logan Date: Thu, 17 Sep 2026 08:37:47 -0700 Subject: [PATCH 4/4] Keep the lock file that lock-file-protect locks Previously the lock file was deleted once the protected code finished. A process already waiting for it then took the lock on a file that no longer had a name, while the next process created a new file and locked that instead, so both ran the protected code at once. This keeps the lock file. An ecosystem update now fetches into a directory of its own so that the lock file the fetcher leaves behind is removed along with everything else it fetched. --- lib/Zef/Repository/Ecosystems.rakumod | 10 +++-- lib/Zef/Utils/FileSystem.rakumod | 7 ++-- t/repository-ecosystems.rakutest | 2 +- t/utils-filesystem.rakutest | 60 ++++++++++++++++++++++++++- 4 files changed, 70 insertions(+), 9 deletions(-) diff --git a/lib/Zef/Repository/Ecosystems.rakumod b/lib/Zef/Repository/Ecosystems.rakumod index 510e2aea..c9573bc3 100644 --- a/lib/Zef/Repository/Ecosystems.rakumod +++ b/lib/Zef/Repository/Ecosystems.rakumod @@ -132,9 +132,13 @@ class Zef::Repository::Ecosystems does PackageRepository { UNDO $stderr.emit("!!!> Failed to update $!name mirror: $uri"); KEEP $stderr.emit("===> Updated $!name mirror: $uri"); - # Other zef processes may be updating this mirror at the same time, so never fetch to a shared path - my $save-as = $!cache.IO.child("{$uri.IO.basename}.{time}.{$*PID}.{(^10000).rand}"); - LEAVE try delete-paths($save-as) if $save-as.e; + # 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)); 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 index a5af4cc4..19f3ccc7 100644 --- a/t/repository-ecosystems.rakutest +++ b/t/repository-ecosystems.rakutest @@ -51,7 +51,7 @@ subtest 'Concurrent updates of the same index' => { 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.grep(*.f).List, (), 'no fetched mirror files are left in the cache'; + is-deeply $cache.dir.map(*.basename).List, ('test',), 'nothing the update fetched is left in the cache'; } 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);