package archive
import (
"context"
"errors"
"fmt"
"os"
"strings"
"sync"
"time"
)
// Objects is the bucket an archive is kept in, as much of it as
// Bucketed needs.
type Objects interface {
// Where names the bucket. A key known present in one bucket says
// nothing about another bound later under the same repository.
Where() string
Has(ctx context.Context, key string) (bool, error)
// Upload sends the file at path to key, streaming it.
Upload(ctx context.Context, key, path string) error
}
// Bucketed puts each archive in a bucket once and remembers that it is
// there. The bucket is the cache: the forge builds an archive on the
// first request for it and every later reader is sent to the bucket,
// so the compression is paid once per check-in and container rather
// than once per reader, and the bytes never pass through the forge
// again.
//
// Keys are content-addressed, so an object is never stale. A positive
// answer is remembered, and asked again after an hour, since a
// lifecycle rule or a person can remove an object and a reader sent to
// a missing one gets the bucket's 404, which the forge never sees. A
// negative answer is not remembered: another node may have built it.
type Bucketed struct {
mu sync.Mutex
flights map[place]*flight
known map[place]time.Time
// failed is when each bucket last failed. For a minute after, the
// bucket is not asked and the caller serves from the forge, so a
// broken binding costs one failure a minute and not one a request.
failed map[string]time.Time
scopes map[string]*scope
swept map[string]bool
}
type place struct{ where, key string }
type flight struct {
done chan struct{}
res Result
err error
}
// scope is one Forget-able prefix. Uploads into it hold mu for reading
// across their check of gen and the upload itself; Forget takes it for
// writing. So Forget waits for an upload already under way, and an
// upload that starts after a Forget sees the new gen and is refused.
type scope struct {
mu sync.RWMutex
gen uint64
}
// How long a remembered object is trusted, and a failed bucket rested.
const (
recheckAfter = time.Hour
restFor = time.Minute
)
// Job is one archive to make present.
type Job struct {
// Scope is the prefix Forget names for this key: an upload into a
// scope forgotten since the job began is refused.
Scope string
Key string
Build func(dst string) error
// Spill is handed the built file when it could not be uploaded,
// so the reader who waited for the build is served by it rather
// than by a second one. It returns true to keep the file, which is
// then the caller's to remove. nil discards it.
Spill func(path string) (keep bool)
}
// Result says what one Ensure did, for the line the caller logs.
type Result struct {
// Built is true for the one call that ran the build; every other
// caller, waiting on it or arriving after, sees false.
Built bool
Size int64
}
// ErrBucketDown is the answer while a bucket rests after a failure.
var ErrBucketDown = errors.New("archive: the bucket failed recently")
// ErrForgotten is an upload refused because its scope was forgotten
// while the build ran: the repository was deleted.
var ErrForgotten = errors.New("archive: the repository was deleted during the build")
// BuildError is a failure of the build itself, as opposed to the
// bucket's: a caller may serve the archive another way when the bucket
// fails, and should not when the build did.
type BuildError struct{ Err error }
func (e *BuildError) Error() string { return e.Err.Error() }
func (e *BuildError) Unwrap() error { return e.Err }
// UploadError is a good build the bucket did not take. The file went
// to the job's Spill.
type UploadError struct{ Err error }
func (e *UploadError) Error() string { return "archive: upload: " + e.Err.Error() }
func (e *UploadError) Unwrap() error { return e.Err }
func NewBucketed() *Bucketed {
return &Bucketed{flights: map[place]*flight{}, known: map[place]time.Time{},
failed: map[string]time.Time{}, scopes: map[string]*scope{}, swept: map[string]bool{}}
}
// Ensure makes job.Key present in objs, running the build (into a
// temporary file, then uploaded) only if the bucket does not already
// hold it.
//
// Concurrent callers for the same key share one build: the first runs
// it and the rest wait for its answer. A leader whose own request went
// away hands the build on rather than failing every reader behind it.
func (b *Bucketed) Ensure(ctx context.Context, objs Objects, job Job) (Result, error) {
p := place{objs.Where(), job.Key}
for {
b.mu.Lock()
if at, ok := b.failed[p.where]; ok && nowFunc().Sub(at) < restFor {
b.mu.Unlock()
return Result{}, ErrBucketDown
}
if at, ok := b.known[p]; ok {
if nowFunc().Sub(at) < recheckAfter {
b.mu.Unlock()
return Result{}, nil
}
delete(b.known, p)
}
if f, ok := b.flights[p]; ok {
b.mu.Unlock()
select {
case <-f.done:
case <-ctx.Done():
return Result{}, ctx.Err()
}
if errors.Is(f.err, context.Canceled) || errors.Is(f.err, context.DeadlineExceeded) {
continue
}
return Result{}, f.err
}
f := &flight{done: make(chan struct{})}
b.flights[p] = f
sc := b.scopeOf(job.Scope)
b.mu.Unlock()
sc.mu.RLock()
gen := sc.gen
sc.mu.RUnlock()
b.lead(ctx, objs, job, p, f, sc, gen)
return f.res, f.err
}
}
// lead runs one flight. Everything that ends it, a panic included, is
// in defers, so no waiter is left on a flight nobody will close.
func (b *Bucketed) lead(ctx context.Context, objs Objects, job Job, p place, f *flight, sc *scope, gen uint64) {
defer func() {
if r := recover(); r != nil {
f.res, f.err = Result{}, &BuildError{fmt.Errorf("panic: %v", r)}
}
b.mu.Lock()
delete(b.flights, p)
var ue *UploadError
switch {
case f.err == nil:
b.known[p] = nowFunc()
case errors.As(f.err, &ue) && !errors.Is(f.err, ErrForgotten), errors.Is(f.err, errBucket):
b.failed[p.where] = nowFunc()
}
b.mu.Unlock()
f.err = unmark(f.err)
close(f.done)
}()
f.res, f.err = b.fill(ctx, objs, job, sc, gen)
}
// errBucket marks a failure of the bucket's own (a HEAD), for lead to
// rest the bucket; unmark takes it off before anyone sees it.
var errBucket = errors.New("bucket")
type marked struct{ err error }
func (m marked) Error() string { return m.err.Error() }
func (m marked) Is(target error) bool { return target == errBucket }
func (m marked) Unwrap() error { return m.err }
func unmark(err error) error {
if m, ok := err.(marked); ok {
return m.err
}
return err
}
func (b *Bucketed) fill(ctx context.Context, objs Objects, job Job, sc *scope, gen uint64) (Result, error) {
have, err := objs.Has(ctx, job.Key)
if err != nil {
if ctx.Err() != nil {
return Result{}, err
}
return Result{}, marked{err}
}
if have {
return Result{}, nil
}
tmp, err := os.CreateTemp("", "quarry-archive-*")
if err != nil {
return Result{}, err
}
path := tmp.Name()
tmp.Close()
keep := false
defer func() {
if !keep {
os.Remove(path)
}
}()
if err := job.Build(path); err != nil {
return Result{}, &BuildError{err}
}
fi, err := os.Stat(path)
if err != nil {
return Result{}, err
}
err = b.upload(ctx, objs, job.Key, path, sc, gen)
if err != nil {
if job.Spill != nil {
keep = job.Spill(path)
}
return Result{}, &UploadError{err}
}
return Result{Built: true, Size: fi.Size()}, nil
}
func (b *Bucketed) upload(ctx context.Context, objs Objects, key, path string, sc *scope, gen uint64) error {
sc.mu.RLock()
defer sc.mu.RUnlock()
if sc.gen != gen {
return ErrForgotten
}
return objs.Upload(ctx, key, path)
}
// scopeOf is the scope named name, made on first use. The caller holds
// b.mu.
func (b *Bucketed) scopeOf(name string) *scope {
sc, ok := b.scopes[name]
if !ok {
sc = &scope{}
b.scopes[name] = sc
}
return sc
}
// Forget ends a scope: it waits for any upload into it already under
// way, refuses every later upload of a build begun before it, and drops
// what is remembered under prefix in every bucket. A sweep of the
// prefix that follows Forget sees every object an earlier build put
// there.
func (b *Bucketed) Forget(prefix string) {
b.mu.Lock()
sc := b.scopeOf(prefix)
b.mu.Unlock()
sc.mu.Lock()
sc.gen++
sc.mu.Unlock()
b.mu.Lock()
defer b.mu.Unlock()
for p := range b.known {
if strings.HasPrefix(p.key, prefix) {
delete(b.known, p)
}
}
}
// FirstSweep reports true once per name for the life of the process,
// so a sweep of what older builds left runs once rather than per build.
func (b *Bucketed) FirstSweep(name string) bool {
b.mu.Lock()
defer b.mu.Unlock()
if b.swept[name] {
return false
}
b.swept[name] = true
return true
}