Files
hakurei/internal/pkg/pkg.go
Ophestra f712466714
All checks were successful
Test / Create distribution (push) Successful in 47s
Test / Sandbox (push) Successful in 2m50s
Test / ShareFS (push) Successful in 4m46s
Test / Sandbox (race detector) (push) Successful in 5m16s
Test / Hpkg (push) Successful in 5m24s
Test / Hakurei (push) Successful in 5m40s
Test / Hakurei (race detector) (push) Successful in 7m27s
Test / Flake checks (push) Successful in 1m42s
internal/pkg: move dependency flooding to cache
This imposes a hard upper limit to concurrency during dependency satisfaction and moves all dependency-related code out of individual implementations of Artifact. This change also includes ctx and msg as part of Cache.

Signed-off-by: Ophestra <cat@gensokyo.uk>
2026-01-09 05:20:34 +09:00

1183 lines
31 KiB
Go

// Package pkg provides utilities for packaging software.
package pkg
import (
"bytes"
"context"
"crypto/sha512"
"encoding/base64"
"encoding/binary"
"errors"
"io"
"io/fs"
"os"
"path"
"path/filepath"
"runtime"
"slices"
"strings"
"sync"
"syscall"
"hakurei.app/container/check"
"hakurei.app/message"
)
type (
// A Checksum is a SHA-384 checksum computed for a cured [Artifact].
Checksum = [sha512.Size384]byte
// An ID is a unique identifier returned by [Artifact.ID]. This value must
// be deterministically determined ahead of time.
ID Checksum
)
// Encode is abbreviation for base64.URLEncoding.EncodeToString(checksum[:]).
func Encode(checksum Checksum) string {
return base64.URLEncoding.EncodeToString(checksum[:])
}
// Decode is abbreviation for base64.URLEncoding.Decode(checksum[:], []byte(s)).
func Decode(s string) (checksum Checksum, err error) {
var n int
n, err = base64.URLEncoding.Decode(checksum[:], []byte(s))
if err == nil && n != len(Checksum{}) {
err = io.ErrUnexpectedEOF
}
return
}
// MustDecode decodes a string representation of [Checksum] and panics if there
// is a decoding error or the resulting data is too short.
func MustDecode(s string) Checksum {
if checksum, err := Decode(s); err != nil {
panic(err)
} else {
return checksum
}
}
// TContext is passed to [TrivialArtifact.Cure] and provides information and
// methods required for curing the [TrivialArtifact].
//
// Methods of TContext are safe for concurrent use. TContext is valid
// until [TrivialArtifact.Cure] returns.
type TContext struct {
// Address of underlying [Cache], should be zeroed or made unusable after
// [TrivialArtifact.Cure] returns and must not be exposed directly.
cache *Cache
// Populated during [Cache.Cure].
work, temp *check.Absolute
}
// destroy destroys the temporary directory and joins its errors with the error
// referred to by errP. If the error referred to by errP is non-nil, the work
// directory is removed similarly. [Cache] is responsible for making sure work
// is never left behind for a successful [Cache.Cure].
//
// destroy must be deferred by [Cache.Cure] if [TContext] is passed to any Cure
// implementation. It should not be called prior to that point.
func (t *TContext) destroy(errP *error) {
if chmodErr, removeErr := removeAll(t.temp); chmodErr != nil || removeErr != nil {
*errP = errors.Join(*errP, chmodErr, removeErr)
return
}
if *errP != nil {
chmodErr, removeErr := removeAll(t.work)
if chmodErr != nil || removeErr != nil {
*errP = errors.Join(*errP, chmodErr, removeErr)
} else if errors.Is(*errP, os.ErrExist) {
// two artifacts may be backed by the same file
*errP = nil
}
}
}
// Unwrap returns the underlying [context.Context].
func (t *TContext) Unwrap() context.Context { return t.cache.ctx }
// GetMessage returns [message.Msg] held by the underlying [Cache].
func (t *TContext) GetMessage() message.Msg { return t.cache.msg }
// GetWorkDir returns a pathname to a directory which [Artifact] is expected to
// write its output to. This is not the final resting place of the [Artifact]
// and this pathname should not be directly referred to in the final contents.
func (t *TContext) GetWorkDir() *check.Absolute { return t.work }
// GetTempDir returns a pathname which implementations may use as scratch space.
// A directory is not created automatically, implementations are expected to
// create it if they wish to use it, using [os.MkdirAll].
func (t *TContext) GetTempDir() *check.Absolute { return t.temp }
// Open tries to open [Artifact] for reading. If a implements [File], its data
// might be used directly, eliminating the roundtrip to vfs. Otherwise, it must
// cure into a directory containing a single regular file.
//
// If err is nil, the caller is responsible for closing the resulting
// [io.ReadCloser].
func (t *TContext) Open(a Artifact) (r io.ReadCloser, err error) {
if f, ok := a.(File); ok {
return t.cache.openFile(f)
}
var pathname *check.Absolute
if pathname, _, err = t.cache.Cure(a); err != nil {
return
}
var entries []os.DirEntry
if entries, err = os.ReadDir(pathname.String()); err != nil {
return
}
if len(entries) != 1 || !entries[0].Type().IsRegular() {
err = errors.New(
"input directory does not contain a single regular file",
)
return
} else {
return os.Open(pathname.Append(entries[0].Name()).String())
}
}
// FContext is passed to [FloodArtifact.Cure] and provides information and
// methods required for curing the [FloodArtifact].
//
// Methods of FContext are safe for concurrent use. FContext is valid
// until [FloodArtifact.Cure] returns.
type FContext struct {
TContext
// Cured top-level dependencies looked up by Pathname.
deps map[ID]*check.Absolute
}
// InvalidLookupError is the identifier of non-dependency [Artifact] looked up
// via [FContext.Pathname] by a misbehaving [Artifact] implementation.
type InvalidLookupError ID
func (e InvalidLookupError) Error() string {
return "attempting to look up non-dependency artifact " + Encode(e)
}
var _ error = InvalidLookupError{}
// Pathname returns the identifier pathname of an [Artifact]. Calling Pathname
// with an [Artifact] not part of the slice returned by [Artifact.Dependencies]
// panics.
func (f *FContext) Pathname(a Artifact) *check.Absolute {
id := Ident(a)
if p, ok := f.deps[id]; ok {
return p
} else {
panic(InvalidLookupError(id))
}
}
// An Artifact is a read-only reference to a piece of data that may be created
// deterministically but might not currently be available in memory or on the
// filesystem.
type Artifact interface {
// Kind returns the [Kind] of artifact. This is usually unique to the
// concrete type but two functionally identical implementations of
// [Artifact] is allowed to return the same [Kind] value.
Kind() Kind
// Params returns opaque bytes that describes [Artifact]. Implementations
// must guarantee that these values are unique among differing instances
// of the same implementation with the same dependencies.
//
// Callers must not modify the retuned byte slice.
//
// Result must remain identical across multiple invocations.
Params() []byte
// Dependencies returns a slice of [Artifact] that the current instance
// depends on to produce its contents.
//
// Callers must not modify the retuned slice.
//
// Result must remain identical across multiple invocations.
Dependencies() []Artifact
}
// FloodArtifact refers to an [Artifact] requiring its entire dependency graph
// to be cured prior to curing itself.
type FloodArtifact interface {
// Cure cures the current [Artifact] to the working directory obtained via
// [TContext.GetWorkDir] embedded in [FContext].
//
// Implementations must not retain c.
Cure(f *FContext) (err error)
Artifact
}
// TrivialArtifact refers to an [Artifact] that cures without requiring that
// any other [Artifact] is cured before it. Its dependency tree is ignored after
// computing its identifier.
//
// TrivialArtifact is unable to cure any other [Artifact] and it cannot access
// pathnames. This type of [Artifact] is primarily intended for dependency-less
// artifacts or direct dependencies that only consists of [File].
type TrivialArtifact interface {
// Cure cures the current [Artifact] to the working directory obtained via
// [TContext.GetWorkDir].
//
// Implementations must not retain c.
Cure(t *TContext) (err error)
Artifact
}
// KnownIdent is optionally implemented by [Artifact] and is used instead of
// [Kind.Ident] when it is available.
//
// This is very subtle to use correctly. The implementation must ensure that
// this value is globally unique, otherwise [Cache] can enter an inconsistent
// state. This should not be implemented outside of testing.
type KnownIdent interface {
// ID returns a globally unique identifier referring to the current
// [Artifact]. This value must be known ahead of time and guaranteed to be
// unique without having obtained the full contents of the [Artifact].
ID() ID
}
// KnownChecksum is optionally implemented by [Artifact] for an artifact with
// output known ahead of time.
type KnownChecksum interface {
// Checksum returns the address of a known checksum.
//
// Callers must not modify the [Checksum].
//
// Result must remain identical across multiple invocations.
Checksum() Checksum
}
// A File refers to an [Artifact] backed by a single file.
type File interface {
// Cure returns the full contents of [File]. If [File] implements
// [KnownChecksum], Cure is responsible for validating any data it produces
// and must return [ChecksumMismatchError] if validation fails.
//
// Callers must not modify the returned byte slice.
//
// Result must remain identical across multiple invocations.
Cure() ([]byte, error)
Artifact
}
// Ident returns the identifier of an [Artifact].
func Ident(a Artifact) ID {
if ki, ok := a.(KnownIdent); ok {
return ki.ID()
}
return a.Kind().Ident(a.Params(), a.Dependencies()...)
}
// Kind corresponds to the concrete type of [Artifact] and is used to create
// identifier for an [Artifact] with dependencies.
type Kind uint64
const (
// KindHTTPGet is the kind of [Artifact] returned by [NewHTTPGet].
KindHTTPGet Kind = iota
// KindTar is the kind of [Artifact] returned by [NewTar].
KindTar
// KindExec is the kind of [Artifact] returned by [NewExec].
KindExec
// KindExecNet is the kind of [Artifact] returned by [NewExec] but with a
// non-nil checksum.
KindExecNet
// KindFile is the kind of [Artifact] returned by [NewFile].
KindFile
// KindCustomOffset is the first [Kind] value reserved for implementations
// not from this package.
KindCustomOffset = 1 << 31
)
// Ident returns a deterministic identifier for the supplied params and
// dependencies. The caller is responsible for ensuring params uniquely and
// deterministically describes the current [Artifact].
func (k Kind) Ident(params []byte, deps ...Artifact) ID {
type extIdent [len(ID{}) + wordSize]byte
identifiers := make([]extIdent, len(deps))
for i, a := range deps {
id := Ident(a)
copy(identifiers[i][wordSize:], id[:])
binary.LittleEndian.PutUint64(identifiers[i][:], uint64(a.Kind()))
}
slices.SortFunc(identifiers, func(a, b extIdent) int {
return bytes.Compare(a[:], b[:])
})
slices.Compact(identifiers)
h := sha512.New384()
h.Write(binary.LittleEndian.AppendUint64(nil, uint64(k)))
h.Write(params)
for _, e := range identifiers {
h.Write(e[:])
}
return ID(h.Sum(nil))
}
const (
// dirIdentifier is the directory name appended to Cache.base for storing
// artifacts named after their [ID].
dirIdentifier = "identifier"
// dirChecksum is the directory name appended to Cache.base for storing
// artifacts named after their [Checksum].
dirChecksum = "checksum"
// dirWork is the directory name appended to Cache.base for working
// pathnames set up during [Cache.Cure].
dirWork = "work"
// dirTemp is the directory name appended to Cache.base for scratch space
// pathnames allocated during [Cache.Cure].
dirTemp = "temp"
// checksumLinknamePrefix is prepended to the encoded [Checksum] value
// of an [Artifact] when creating a symbolic link to dirChecksum.
checksumLinknamePrefix = "../" + dirChecksum + "/"
)
// A pendingArtifactDep is a dependency [Artifact] pending concurrent curing,
// subject to the cures limit. Values pointed to by result addresses are safe
// to access after the [sync.WaitGroup] associated with this pendingArtifactDep
// is done. pendingArtifactDep must not be reused or modified after it is sent
// to Cache.cureDep.
type pendingArtifactDep struct {
// Dependency artifact populated during [Cache.Cure].
a Artifact
// Address of result pathname populated during [Cache.Cure] and dereferenced
// if curing succeeds.
resP **check.Absolute
// Address of result error slice populated during [Cache.Cure], dereferenced
// after acquiring errsMu if curing fails. No additional action is taken,
// [Cache] and its caller are responsible for further error handling.
errs *[]error
// Address of mutex synchronising access to errs.
errsMu *sync.Mutex
// For synchronising access to result buffer.
*sync.WaitGroup
}
// Cache is a support layer that implementations of [Artifact] can use to store
// cured [Artifact] data in a content addressed fashion.
type Cache struct {
// Work for curing dependency [Artifact] is sent here and cured concurrently
// while subject to the cures limit. Invalid after the context is canceled.
cureDep chan<- *pendingArtifactDep
// [context.WithCancel] over caller-supplied context, used by [Artifact] and
// all dependency curing goroutines.
ctx context.Context
// Cancels ctx.
cancel context.CancelFunc
// For waiting on dependency curing goroutines.
wg sync.WaitGroup
// Reports new cures and passed to [Artifact].
msg message.Msg
// Directory where all [Cache] related files are placed.
base *check.Absolute
// Whether to validate [File.Data] for a [KnownChecksum] file. This
// significantly reduces performance.
strict bool
// Synchronises access to dirChecksum.
checksumMu sync.RWMutex
// Identifier to content pair cache.
ident map[ID]Checksum
// Identifier to error pair for unrecoverably faulted [Artifact].
identErr map[ID]error
// Pending identifiers, accessed through Cure for entries not in ident.
identPending map[ID]<-chan struct{}
// Synchronises access to ident and corresponding filesystem entries.
identMu sync.RWMutex
}
// IsStrict returns whether the [Cache] strictly verifies checksums.
func (c *Cache) IsStrict() bool { return c.strict }
// SetStrict sets whether the [Cache] strictly verifies checksums, even when
// the implementation promises to validate them internally. This significantly
// reduces performance and is not recommended outside of testing.
//
// This method is not safe for concurrent use with any other method.
func (c *Cache) SetStrict(strict bool) { c.strict = strict }
// A ChecksumMismatchError describes an [Artifact] with unexpected content.
type ChecksumMismatchError struct {
// Actual and expected checksums.
Got, Want Checksum
}
func (e *ChecksumMismatchError) Error() string {
return "got " + Encode(e.Got) +
" instead of " + Encode(e.Want)
}
// ScrubError describes the outcome of a [Cache.Scrub] call where errors were
// found and removed from the underlying storage of [Cache].
type ScrubError struct {
// Content-addressed entries not matching their checksum. This can happen
// if an incorrect [File] implementation was cured against a non-strict
// [Cache].
ChecksumMismatches []ChecksumMismatchError
// Dangling identifier symlinks. This can happen if the content-addressed
// entry was removed while scrubbing due to a checksum mismatch.
DanglingIdentifiers []ID
// Miscellaneous errors, including [os.ReadDir] on checksum and identifier
// directories, [Decode] on entry names and [os.RemoveAll] on inconsistent
// entries.
Errs []error
}
// Unwrap returns a concatenation of ChecksumMismatches and Errs.
func (e *ScrubError) Unwrap() []error {
s := make([]error, 0, len(e.ChecksumMismatches)+len(e.Errs))
for _, err := range e.ChecksumMismatches {
s = append(s, &err)
}
for _, err := range e.Errs {
s = append(s, err)
}
return s
}
// Error returns a multi-line representation of [ScrubError].
func (e *ScrubError) Error() string {
var segments []string
if len(e.ChecksumMismatches) > 0 {
s := "checksum mismatches:\n"
for _, m := range e.ChecksumMismatches {
s += m.Error() + "\n"
}
segments = append(segments, s)
}
if len(e.DanglingIdentifiers) > 0 {
s := "dangling identifiers:\n"
for _, id := range e.DanglingIdentifiers {
s += Encode(id) + "\n"
}
segments = append(segments, s)
}
if len(e.Errs) > 0 {
s := "errors during scrub:\n"
for _, err := range e.Errs {
s += err.Error() + "\n"
}
segments = append(segments, s)
}
return strings.Join(segments, "\n")
}
// Scrub frees internal in-memory identifier to content pair cache, verifies all
// cached artifacts against their checksums, checks for dangling identifier
// symlinks and removes them if found.
//
// This method is not safe for concurrent use with any other method.
func (c *Cache) Scrub() error {
c.identMu.Lock()
defer c.identMu.Unlock()
c.checksumMu.Lock()
defer c.checksumMu.Unlock()
c.ident = make(map[ID]Checksum)
c.identErr = make(map[ID]error)
var se ScrubError
var (
ent os.DirEntry
dir *check.Absolute
)
condemnEntry := func() {
chmodErr, removeErr := removeAll(dir.Append(ent.Name()))
if chmodErr != nil {
se.Errs = append(se.Errs, chmodErr)
}
if removeErr != nil {
se.Errs = append(se.Errs, removeErr)
}
}
dir = c.base.Append(dirChecksum)
if entries, err := os.ReadDir(dir.String()); err != nil {
se.Errs = append(se.Errs, err)
} else {
var got, want Checksum
for _, ent = range entries {
if want, err = Decode(ent.Name()); err != nil {
se.Errs = append(se.Errs, err)
condemnEntry()
continue
}
if ent.IsDir() {
if got, err = HashDir(dir.Append(ent.Name())); err != nil {
se.Errs = append(se.Errs, err)
continue
}
} else if ent.Type().IsRegular() {
h := sha512.New384()
var r *os.File
r, err = os.Open(dir.Append(ent.Name()).String())
if err != nil {
se.Errs = append(se.Errs, err)
continue
}
_, err = io.Copy(h, r)
closeErr := r.Close()
if closeErr != nil {
se.Errs = append(se.Errs, closeErr)
}
if err != nil {
se.Errs = append(se.Errs, err)
continue
}
h.Sum(got[:0])
} else {
se.Errs = append(se.Errs, InvalidFileModeError(ent.Type()))
condemnEntry()
continue
}
if got != want {
se.ChecksumMismatches = append(se.ChecksumMismatches, ChecksumMismatchError{
Got: got,
Want: want,
})
condemnEntry()
}
}
}
dir = c.base.Append(dirIdentifier)
if entries, err := os.ReadDir(dir.String()); err != nil {
se.Errs = append(se.Errs, err)
} else {
var (
id ID
linkname string
)
for _, ent = range entries {
if id, err = Decode(ent.Name()); err != nil {
se.Errs = append(se.Errs, err)
condemnEntry()
continue
}
if linkname, err = os.Readlink(
dir.Append(ent.Name()).String(),
); err != nil {
se.Errs = append(se.Errs, err)
se.DanglingIdentifiers = append(se.DanglingIdentifiers, id)
condemnEntry()
continue
}
if _, err = Decode(path.Base(linkname)); err != nil {
se.Errs = append(se.Errs, err)
se.DanglingIdentifiers = append(se.DanglingIdentifiers, id)
condemnEntry()
continue
}
if _, err = os.Stat(dir.Append(ent.Name()).String()); err != nil {
if !errors.Is(err, os.ErrNotExist) {
se.Errs = append(se.Errs, err)
}
se.DanglingIdentifiers = append(se.DanglingIdentifiers, id)
condemnEntry()
continue
}
}
}
if len(c.identPending) > 0 {
se.Errs = append(se.Errs, errors.New(
"scrub began with pending artifacts",
))
} else {
chmodErr, removeErr := removeAll(c.base.Append(dirWork))
if chmodErr != nil {
se.Errs = append(se.Errs, chmodErr)
}
if removeErr != nil {
se.Errs = append(se.Errs, removeErr)
}
if err := os.Mkdir(
c.base.Append(dirWork).String(),
0700,
); err != nil {
se.Errs = append(se.Errs, err)
}
chmodErr, removeErr = removeAll(c.base.Append(dirTemp))
if chmodErr != nil {
se.Errs = append(se.Errs, chmodErr)
}
if removeErr != nil {
se.Errs = append(se.Errs, removeErr)
}
}
if len(se.ChecksumMismatches) > 0 ||
len(se.DanglingIdentifiers) > 0 ||
len(se.Errs) > 0 {
return &se
} else {
return nil
}
}
// loadOrStoreIdent attempts to load a cached [Artifact] by its identifier or
// wait for a pending [Artifact] to cure. If neither is possible, the current
// identifier is stored in identPending and a non-nil channel is returned.
func (c *Cache) loadOrStoreIdent(id *ID) (
done chan<- struct{},
checksum Checksum,
err error,
) {
var ok bool
c.identMu.Lock()
if checksum, ok = c.ident[*id]; ok {
c.identMu.Unlock()
return
}
if err, ok = c.identErr[*id]; ok {
c.identMu.Unlock()
return
}
var notify <-chan struct{}
if notify, ok = c.identPending[*id]; ok {
c.identMu.Unlock()
<-notify
c.identMu.RLock()
if checksum, ok = c.ident[*id]; !ok {
err = c.identErr[*id]
}
c.identMu.RUnlock()
return
}
d := make(chan struct{})
c.identPending[*id] = d
c.identMu.Unlock()
done = d
return
}
// finaliseIdent commits a checksum or error to ident for an identifier
// previously submitted to identPending.
func (c *Cache) finaliseIdent(
done chan<- struct{},
id *ID,
checksum *Checksum,
err error,
) {
c.identMu.Lock()
if err != nil {
c.identErr[*id] = err
} else {
c.ident[*id] = *checksum
}
delete(c.identPending, *id)
c.identMu.Unlock()
close(done)
}
// openFile tries to load [File] from [Cache], and if that fails, obtains it via
// [File.Cure] instead. Notably, it does not cure [File]. If err is nil, the
// caller is responsible for closing the resulting [io.ReadCloser].
func (c *Cache) openFile(f File) (r io.ReadCloser, err error) {
if kc, ok := f.(KnownChecksum); ok {
c.checksumMu.RLock()
r, err = os.Open(c.base.Append(
dirChecksum,
Encode(kc.Checksum()),
).String())
c.checksumMu.RUnlock()
} else {
c.identMu.RLock()
r, err = os.Open(c.base.Append(
dirIdentifier,
Encode(Ident(f)),
).String())
c.identMu.RUnlock()
}
if err != nil {
if !errors.Is(err, os.ErrNotExist) {
return
}
var data []byte
if data, err = f.Cure(); err != nil {
return
}
r = io.NopCloser(bytes.NewReader(data))
}
return
}
// InvalidFileModeError describes an [Artifact.Cure] that did not result in
// a regular file or directory located at the work pathname.
type InvalidFileModeError fs.FileMode
// Error returns a constant string.
func (e InvalidFileModeError) Error() string {
return "artifact did not produce a regular file or directory"
}
// NoOutputError describes an [Artifact.Cure] that did not populate its
// work pathname despite completing successfully.
type NoOutputError struct{}
// Unwrap returns [os.ErrNotExist].
func (NoOutputError) Unwrap() error { return os.ErrNotExist }
// Error returns a constant string.
func (NoOutputError) Error() string {
return "artifact cured successfully but did not produce any output"
}
// removeAll is similar to [os.RemoveAll] but is robust against any permissions.
func removeAll(pathname *check.Absolute) (chmodErr, removeErr error) {
chmodErr = filepath.WalkDir(pathname.String(), func(
path string,
d fs.DirEntry,
err error,
) error {
if err != nil {
return err
}
if d.IsDir() {
return os.Chmod(path, 0700)
}
return nil
})
if errors.Is(chmodErr, os.ErrNotExist) {
chmodErr = nil
}
removeErr = os.RemoveAll(pathname.String())
return
}
// overrideFileInfo overrides the permission bits of [fs.FileInfo] to 0500 and
// is the concrete type returned by overrideFile.Stat.
type overrideFileInfo struct{ fs.FileInfo }
// Mode returns [fs.FileMode] with its permission bits set to 0500.
func (fi overrideFileInfo) Mode() fs.FileMode {
return fi.FileInfo.Mode()&(^fs.FileMode(0777)) | 0500
}
// Sys returns nil to avoid passing the original permission bits.
func (fi overrideFileInfo) Sys() any { return nil }
// overrideFile overrides the permission bits of [fs.File] to 0500 and is the
// concrete type returned by dotOverrideFS for calls with "." passed as name.
type overrideFile struct{ fs.File }
func (f overrideFile) Stat() (fi fs.FileInfo, err error) {
fi, err = f.File.Stat()
if err != nil {
return
}
fi = overrideFileInfo{fi}
return
}
// dirFS is implemented by the concrete type of the return value of [os.DirFS].
type dirFS interface {
fs.StatFS
fs.ReadFileFS
fs.ReadDirFS
fs.ReadLinkFS
}
// dotOverrideFS overrides the permission bits of "." to 0500 to avoid the extra
// system calls to add and remove write bit from the target directory.
type dotOverrideFS struct{ dirFS }
// Open wraps the underlying [fs.FS] with "." special case.
func (fsys dotOverrideFS) Open(name string) (f fs.File, err error) {
f, err = fsys.dirFS.Open(name)
if err != nil || name != "." {
return
}
f = overrideFile{f}
return
}
// Stat wraps the underlying [fs.FS] with "." special case.
func (fsys dotOverrideFS) Stat(name string) (fi fs.FileInfo, err error) {
fi, err = fsys.dirFS.Stat(name)
if err != nil || name != "." {
return
}
fi = overrideFileInfo{fi}
return
}
// InvalidArtifactError describes an artifact that does not implement a
// supported Cure method.
type InvalidArtifactError ID
func (e InvalidArtifactError) Error() string {
return "artifact " + Encode(e) + " cannot be cured"
}
// Cure cures the [Artifact] and returns its pathname and [Checksum]. Direct
// calls to Cure are not subject to the cures limit.
func (c *Cache) Cure(a Artifact) (
pathname *check.Absolute,
checksum Checksum,
err error,
) {
id := Ident(a)
ids := Encode(id)
pathname = c.base.Append(
dirIdentifier,
ids,
)
defer func() {
if err != nil {
pathname = nil
checksum = Checksum{}
}
}()
var done chan<- struct{}
done, checksum, err = c.loadOrStoreIdent(&id)
if done == nil {
return
} else {
defer func() { c.finaliseIdent(done, &id, &checksum, err) }()
}
_, err = os.Lstat(pathname.String())
if err == nil {
var name string
if name, err = os.Readlink(pathname.String()); err != nil {
return
}
checksum, err = Decode(path.Base(name))
return
}
if !errors.Is(err, os.ErrNotExist) {
return
}
var checksums string
defer func() {
if err == nil && checksums != "" {
err = os.Symlink(
checksumLinknamePrefix+checksums,
pathname.String(),
)
}
}()
var checksumPathname *check.Absolute
var checksumFi os.FileInfo
if kc, ok := a.(KnownChecksum); ok {
checksum = kc.Checksum()
checksums = Encode(checksum)
checksumPathname = c.base.Append(
dirChecksum,
checksums,
)
c.checksumMu.RLock()
checksumFi, err = os.Stat(checksumPathname.String())
c.checksumMu.RUnlock()
if err != nil {
if !errors.Is(err, os.ErrNotExist) {
return
}
checksumFi, err = nil, nil
}
}
if c.msg.IsVerbose() {
c.msg.Verbosef("curing %s...", Encode(id))
}
// cure File outside type switch to skip TContext initialisation
if f, ok := a.(File); ok {
if checksumFi != nil {
if !checksumFi.Mode().IsRegular() {
// unreachable
err = InvalidFileModeError(checksumFi.Mode())
}
return
}
var data []byte
data, err = f.Cure()
if err != nil {
return
}
if checksumPathname == nil {
h := sha512.New384()
h.Write(data)
h.Sum(checksum[:0])
checksums = Encode(checksum)
checksumPathname = c.base.Append(
dirChecksum,
checksums,
)
} else if c.IsStrict() {
h := sha512.New384()
h.Write(data)
if got := Checksum(h.Sum(nil)); got != checksum {
err = &ChecksumMismatchError{
Got: got,
Want: checksum,
}
return
}
}
c.checksumMu.Lock()
var w *os.File
w, err = os.OpenFile(
checksumPathname.String(),
os.O_CREATE|os.O_EXCL|os.O_WRONLY,
0400,
)
if err != nil {
c.checksumMu.Unlock()
if errors.Is(err, os.ErrExist) {
err = nil
}
return
}
_, err = w.Write(data)
closeErr := w.Close()
if err == nil {
err = closeErr
}
c.checksumMu.Unlock()
return
}
if checksumFi != nil {
if !checksumFi.Mode().IsDir() {
// unreachable
err = InvalidFileModeError(checksumFi.Mode())
}
return
}
t := TContext{c, c.base.Append(dirWork, ids), c.base.Append(dirTemp, ids)}
switch ca := a.(type) {
case TrivialArtifact:
defer t.destroy(&err)
if err = ca.Cure(&t); err != nil {
return
}
break
case FloodArtifact:
deps := a.Dependencies()
f := FContext{t, make(map[ID]*check.Absolute, len(deps))}
var wg sync.WaitGroup
wg.Add(len(deps))
res := make([]*check.Absolute, len(deps))
errs := make([]error, 0, len(deps))
var errsMu sync.Mutex
for i, d := range deps {
pending := pendingArtifactDep{d, &res[i], &errs, &errsMu, &wg}
select {
case c.cureDep <- &pending:
break
case <-c.ctx.Done():
err = c.ctx.Err()
return
}
}
wg.Wait()
if len(errs) > 0 {
err = errors.Join(errs...)
if err == nil {
// unreachable
err = syscall.ENOTRECOVERABLE
}
return
}
for i, p := range res {
f.deps[Ident(deps[i])] = p
}
defer f.destroy(&err)
if err = ca.Cure(&f); err != nil {
return
}
break
default:
err = InvalidArtifactError(id)
return
}
t.cache = nil
var fi os.FileInfo
if fi, err = os.Lstat(t.work.String()); err != nil {
if errors.Is(err, os.ErrNotExist) {
err = NoOutputError{}
}
return
}
if !fi.IsDir() {
if !fi.Mode().IsRegular() {
err = InvalidFileModeError(fi.Mode())
} else {
err = errors.New("non-file artifact produced regular file")
}
return
}
var gotChecksum Checksum
if gotChecksum, err = HashFS(
dotOverrideFS{os.DirFS(t.work.String()).(dirFS)},
".",
); err != nil {
return
}
if checksumPathname == nil {
checksum = gotChecksum
checksums = Encode(checksum)
checksumPathname = c.base.Append(
dirChecksum,
checksums,
)
} else {
if gotChecksum != checksum {
err = &ChecksumMismatchError{
Got: gotChecksum,
Want: checksum,
}
return
}
}
if err = os.Chmod(t.work.String(), 0700); err != nil {
return
}
c.checksumMu.Lock()
if err = os.Rename(
t.work.String(),
checksumPathname.String(),
); err != nil {
if !errors.Is(err, os.ErrExist) {
c.checksumMu.Unlock()
return
}
// err is zeroed during deferred cleanup
} else {
err = os.Chmod(checksumPathname.String(), 0500)
}
c.checksumMu.Unlock()
return
}
// cure cures the pending [Artifact], stores its result and notifies the caller.
func (pending *pendingArtifactDep) cure(c *Cache) {
defer pending.Done()
pathname, _, err := c.Cure(pending.a)
if err == nil {
*pending.resP = pathname
return
}
pending.errsMu.Lock()
*pending.errs = append(*pending.errs, err)
pending.errsMu.Unlock()
}
// Close cancels all pending cures and waits for them to clean up.
func (c *Cache) Close() { c.cancel(); c.wg.Wait() }
// 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.
//
// A cures value of 0 or lower is equivalent to the value returned by
// [runtime.NumCPU].
func New(
ctx context.Context,
msg message.Msg,
cures int,
base *check.Absolute,
) (*Cache, error) {
for _, name := range []string{
dirIdentifier,
dirChecksum,
dirWork,
} {
if err := os.MkdirAll(base.Append(name).String(), 0700); err != nil &&
!errors.Is(err, os.ErrExist) {
return nil, err
}
}
c := Cache{
msg: msg,
base: base,
ident: make(map[ID]Checksum),
identErr: make(map[ID]error),
identPending: make(map[ID]<-chan struct{}),
}
c.ctx, c.cancel = context.WithCancel(ctx)
cureDep := make(chan *pendingArtifactDep, cures)
c.cureDep = cureDep
if cures < 1 {
cures = runtime.NumCPU()
}
for i := 0; i < cures; i++ {
c.wg.Go(func() {
for {
select {
case <-c.ctx.Done():
return
case pending := <-cureDep:
pending.cure(&c)
break
}
}
})
}
return &c, nil
}