internal/pkg: lock on-filesystem cache
All checks were successful
Test / Create distribution (push) Successful in 47s
Test / Sandbox (push) Successful in 2m55s
Test / ShareFS (push) Successful in 4m52s
Test / Hpkg (push) Successful in 5m11s
Test / Sandbox (race detector) (push) Successful in 5m17s
Test / Hakurei (race detector) (push) Successful in 7m53s
Test / Hakurei (push) Successful in 4m6s
Test / Flake checks (push) Successful in 1m41s

Any fine-grained file-based locking here significantly hurts performance and is not part of the use case of the package. This change guarantees exclusive access to prevent inconsistent state on the filesystem.

Signed-off-by: Ophestra <cat@gensokyo.uk>
This commit is contained in:
2026-01-16 18:09:10 +09:00
parent 5936e6a4aa
commit 610ee13ab3
2 changed files with 77 additions and 15 deletions

View File

@@ -22,10 +22,12 @@ import (
"strings"
"sync"
"syscall"
"testing"
"unique"
"unsafe"
"hakurei.app/container/check"
"hakurei.app/internal/lockedfile"
"hakurei.app/message"
)
@@ -353,6 +355,10 @@ const (
)
const (
// fileLock is the file name appended to Cache.base for guaranteeing
// exclusive access to the cache directory.
fileLock = "lock"
// dirIdentifier is the directory name appended to Cache.base for storing
// artifacts named after their [ID].
dirIdentifier = "identifier"
@@ -438,6 +444,11 @@ type Cache struct {
identPending map[unique.Handle[ID]]<-chan struct{}
// Synchronises access to ident and corresponding filesystem entries.
identMu sync.RWMutex
// Unlocks the on-filesystem cache. Must only be called from Close.
unlock func()
// Synchronises calls to Close.
closeOnce sync.Once
}
// IsStrict returns whether the [Cache] strictly verifies checksums.
@@ -1405,19 +1416,44 @@ func (pending *pendingArtifactDep) cure(c *Cache) {
}
// Close cancels all pending cures and waits for them to clean up.
func (c *Cache) Close() { c.cancel(); c.wg.Wait() }
func (c *Cache) Close() {
c.closeOnce.Do(func() {
c.cancel()
c.wg.Wait()
c.unlock()
})
}
// New returns the address to a new instance of [Cache]. Concurrent cures of
// dependency [Artifact] is limited to the caller-supplied value, however direct
// calls to [Cache.Cure] is not subject to this limitation.
// Open returns the address of a newly opened instance of [Cache].
//
// Concurrent cures of a [FloodArtifact] dependency graph is limited to the
// caller-supplied value, however direct calls to [Cache.Cure] is not subject
// to this limitation.
//
// A cures value of 0 or lower is equivalent to the value returned by
// [runtime.NumCPU].
func New(
//
// A successful call to Open guarantees exclusive access to the on-filesystem
// cache for the resulting instance of [Cache]. The [Cache.Close] method cancels
// and waits for pending cures on [Cache] before releasing this lock and must be
// called once the [Cache] is no longer needed.
func Open(
ctx context.Context,
msg message.Msg,
cures int,
base *check.Absolute,
) (*Cache, error) {
return open(ctx, msg, cures, base, true)
}
// open implements Open but allows omitting the [lockedfile] lock when called
// from a test. This is used to simulate invalid states in the test suite.
func open(
ctx context.Context,
msg message.Msg,
cures int,
base *check.Absolute,
lock bool,
) (*Cache, error) {
for _, name := range []string{
dirIdentifier,
@@ -1443,6 +1479,18 @@ func New(
c.cureDep = cureDep
c.identPool.New = func() any { return new(extIdent) }
if lock || !testing.Testing() {
if unlock, err := lockedfile.MutexAt(
base.Append(fileLock).String(),
).Lock(); err != nil {
return nil, err
} else {
c.unlock = unlock
}
} else {
c.unlock = func() {}
}
if cures < 1 {
cures = runtime.NumCPU()
}