Skip to content
quarry Artifact [3c7960d727]
Login

Artifact [3c7960d727]

Check-ins Using Download Hex

Artifact 3c7960d7271df78b7caba7188ce3fbac1029e094571997c3b9eff882eaee7241:

package main

import (
	"context"
	"fmt"
	"log"
	"net/http"
	"os"
	"path/filepath"
	"sync"
	"sync/atomic"
	"time"

	"craftdesign.group/quarry/internal/api"
	"craftdesign.group/quarry/internal/ci"
	"craftdesign.group/quarry/internal/events"
	"craftdesign.group/quarry/internal/index"
	"craftdesign.group/quarry/internal/ingest"
	"craftdesign.group/quarry/internal/lease"
	"craftdesign.group/quarry/internal/oltp"
	"craftdesign.group/quarry/internal/ops"
	"craftdesign.group/quarry/internal/work"
	"craftdesign.group/quarry/internal/xfer"
	"craftdesign.group/quarry/internal/zitcli"

	"errors"
)

// registrar owns a served repository's whole lifetime: its index handle,
// its ingest pipeline, its sync backend, and the goroutine polling it.
//
// This was the body of serve's startup loop. It is a type now because a
// repository can be registered while the server runs, and the loop's
// shape did not survive that: it relied on `defer ix.Close()` reaching
// serve's return, and on a poll goroutine that never stops. Neither is
// available to an admin call, and both are leaks if a repository is ever
// removed.
type registrar struct {
	dir         string // derived-data root
	reposRoot   string // where repositories created through the API are placed
	node        string // this node's id, for durable consumer names
	interval    time.Duration
	ciCloneBase string

	ev         *events.Server
	leases     *lease.Manager
	deps       *api.Deps
	fronter    *xfer.Fronter
	ingestLane *work.Lane
	bulk       *work.Lane
	engine     *ci.Engine
	odb        *oltp.DB

	// codes is shared with the auth verifier's cache, so a project code
	// learned here is the one the push gate checks.
	codes  map[string]string
	codeMu *sync.Mutex

	mu      sync.Mutex
	running map[string]*served

	privMu  sync.RWMutex
	private map[string]bool
}

// served is what a registration leaves behind that must be released.
type served struct {
	ix   *index.Index
	stop context.CancelFunc
	done chan struct{} // closed when the poll goroutine has exited
}

func (rg *registrar) track(name string, s *served) {
	rg.mu.Lock()
	defer rg.mu.Unlock()
	if rg.running == nil {
		rg.running = map[string]*served{}
	}
	rg.running[name] = s
}

// Register opens a repository and starts serving it. Safe to call while
// the server is running.
func (rg *registrar) Register(name, path string) error {
	if _, exists := rg.deps.Repos.Get(name); exists {
		return fmt.Errorf("repository %q is already served: %w", name, ops.ErrInvalid)
	}
	if err := os.MkdirAll(filepath.Join(rg.dir, name), 0o755); err != nil {
		return err
	}
	ix, err := index.Open(filepath.Join(rg.dir, name, "index.db"))
	if err != nil {
		return err
	}
	cli := zitcli.New(path)
	// The lease keys on the store's project code, learned via the same
	// boundary-clean handshake user-add uses. Doing it before anything
	// is published means a path that is not a repository fails here,
	// with the index closed again, rather than half-registering.
	code, cerr := xfer.ProjectCode(path)
	if cerr != nil {
		ix.Close()
		// A path that is not a repository is a bad REQUEST, not a server
		// fault. Without this it surfaced as "internal error", which
		// tells an operator to read the server log for what is in fact
		// their own typo.
		return fmt.Errorf("%w: %s is not a zit repository: %v", ops.ErrInvalid, path, cerr)
	}
	rg.codeMu.Lock()
	rg.codes[name] = code
	rg.codeMu.Unlock()

	indexing := new(atomic.Bool)
	indexing.Store(true)
	entry := &api.Repo{
		Index: ix, Zit: cli, Code: code,
		Indexing: indexing.Load,
		Guard: func(fn func() error) error {
			err := rg.leases.Guard(code, fn)
			if errors.Is(err, lease.ErrNotHolder) || errors.Is(err, lease.ErrUnavailable) {
				return fmt.Errorf("%w: %v", ops.ErrUnavailable, err)
			}
			return err
		},
	}
	entry.Mirror = func(url string) (bool, error) {
		var changed bool
		err := entry.Guard(func() error {
			var perr error
			changed, perr = cli.Pull(url)
			return perr
		})
		return changed, err
	}
	pipe := ingest.NewPipeline(cli, rg.ev, ix, name, rg.node)
	entry.Sync = pipe.Sync
	entry.Health = func() api.RepoHealth {
		s := pipe.Stats()
		h := api.RepoHealth{
			RoundSeconds: s.LastRound.Seconds(),
			LastCount:    s.LastCount,
			Total:        s.Total,
			Fallbacks:    s.Fallbacks,
			LastErr:      s.LastErr,
			Failures:     s.Failures,
		}
		if !s.LastOK.IsZero() {
			h.LastOK = s.LastOK.Format(time.RFC3339)
		}
		if !s.LastErrAt.IsZero() {
			h.LastErrAt = s.LastErrAt.Format(time.RFC3339)
		}
		return h
	}

	rg.fronter.Register(name, path)
	// Published complete or not at all.
	rg.deps.Repos.Put(name, entry)

	ctx, stop := context.WithCancel(context.Background())
	done := make(chan struct{})
	rg.track(name, &served{ix: ix, stop: stop, done: done})
	go rg.poll(ctx, done, name, path, cli, ix, pipe, indexing)
	return nil
}

// poll runs the store's initial ingest and then its round loop until the
// context is cancelled.
func (rg *registrar) poll(ctx context.Context, done chan struct{}, name, path string,
	cli *zitcli.Client, ix *index.Index, pipe *ingest.Pipeline, indexing *atomic.Bool) {
	defer close(done)

	start := time.Now()
	var n int
	err := rg.ingestLane.Do(ctx, func() error {
		var serr error
		n, serr = pipe.Sync()
		return serr
	})
	indexing.Store(false)
	if err != nil {
		log.Printf("initial ingest of %s: %v", name, err)
	} else {
		log.Printf("ingested %d artifacts from %s (%s) in %s",
			n, name, path, time.Since(start).Round(time.Second))
	}

	// time.Tick leaks its ticker; a Ticker can be stopped, which matters
	// now that a repository can be unregistered.
	tick := time.NewTicker(rg.interval)
	defer tick.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-tick.C:
		}
		if err := rg.ingestLane.Do(ctx, func() error {
			_, serr := pipe.Sync()
			return serr
		}); err != nil {
			if ctx.Err() != nil {
				return
			}
			log.Printf("ingest %s: %v", name, err)
			continue
		}
		// Warm the tip's listing so the repo home is never the request
		// that pays for it: `ls --long` costs ~4ms per file, which is
		// 8.5s on a 2,198-path tree. The listing is immutable per
		// version, so it is computed once and reused until the tip moves.
		if b, ok, err := ix.TipOf("trunk"); err == nil && ok {
			go cli.Ls(context.Background(), b.Tip)
		}
		// CI looks right after a fold, which is the moment a new
		// check-in becomes visible to quarry whatever brought it in.
		if rg.engine.Enabled() {
			if err := ci.Trigger(rg.engine, ciStore{db: rg.odb}, name,
				ciRepo{ix: ix, zit: cli}, rg.ciCloneBase+name); err != nil {
				log.Printf("ci: %s: %v", name, err)
			}
		}
	}
}

// Unregister stops serving a repository and releases what it held.
//
// Order matters: drop it from the registry FIRST so no new request can
// find it, then stop the poll loop and wait for it to exit, then close
// the index. Closing while a fold is mid-transaction is how a "removed"
// repository takes the process down with it.
func (rg *registrar) Unregister(name string) error {
	if _, ok := rg.deps.Repos.Get(name); !ok {
		return fmt.Errorf("no repository %q: %w", name, ops.ErrNotFound)
	}
	rg.deps.Repos.Delete(name)
	rg.fronter.Unregister(name)

	rg.mu.Lock()
	s := rg.running[name]
	delete(rg.running, name)
	rg.mu.Unlock()

	if s != nil {
		s.stop()
		<-s.done
		s.ix.Close()
	}
	rg.codeMu.Lock()
	delete(rg.codes, name)
	rg.codeMu.Unlock()
	return nil
}

// Add creates or adopts a repository, records it, and starts serving it.
//
// The order is deliberate: make the store, then persist, then serve. A
// recorded repository that failed to open is visible and fixable; one
// that serves without being recorded vanishes at the next restart and
// leaves no trace of why.
func (rg *registrar) Add(spec api.RepoSpec) (string, error) {
	if rg.odb == nil {
		return "", fmt.Errorf("no forge database: %w", ops.ErrUnavailable)
	}
	if _, exists := rg.deps.Repos.Get(spec.Name); exists {
		return "", fmt.Errorf("repository %q is already served: %w", spec.Name, ops.ErrInvalid)
	}
	// The name is validated BEFORE it is joined to a path, because that
	// is the only thing standing between a name and the filesystem.
	if !oltp.RepoName.MatchString(spec.Name) {
		return "", fmt.Errorf("%w: repository name %q: use letters, digits, dot, dash, underscore",
			ops.ErrInvalid, spec.Name)
	}
	if spec.Path == "" {
		if rg.reposRoot == "" {
			return "", fmt.Errorf("%w: this node has no repositories root, so path is required "+
				"(start it with --repos-root DIR)", ops.ErrInvalid)
		}
		if !spec.Create && spec.Clone == "" {
			return "", fmt.Errorf("%w: path is required when adopting an existing store", ops.ErrInvalid)
		}
		spec.Path = filepath.Join(rg.reposRoot, spec.Name)
	}

	// Making a store is the one step that leaves something behind if a
	// later one fails, so it is undone explicitly rather than left for
	// an operator to find.
	made := false
	switch {
	case spec.Create:
		if err := rg.mustBeEmpty(spec.Path); err != nil {
			return "", err
		}
		if err := zitcli.Init(context.Background(), spec.Path); err != nil {
			return "", fmt.Errorf("%w: %v", ops.ErrInvalid, err)
		}
		made = true
	case spec.Clone != "":
		if err := rg.mustBeEmpty(spec.Path); err != nil {
			return "", err
		}
		// A clone is minutes of network for a large repository, so it
		// takes the bulk lane like any other long, memory-hungry job
		// rather than running unbounded because an admin asked.
		err := rg.bulk.Do(context.Background(), func() error {
			return zitcli.Clone(context.Background(), spec.Clone, spec.Path)
		})
		if errors.Is(err, work.ErrBusy) {
			return "", fmt.Errorf("too many clones in flight, retry shortly: %w", ops.ErrUnavailable)
		}
		if err != nil {
			os.RemoveAll(spec.Path)
			return "", fmt.Errorf("%w: %v", ops.ErrInvalid, err)
		}
		made = true
	}

	undo := func() {
		if made {
			os.RemoveAll(spec.Path)
		}
	}
	if err := rg.odb.AddRepo(spec.Name, spec.Path, spec.Description, spec.Visibility); err != nil {
		undo()
		return "", fmt.Errorf("%w: %v", ops.ErrInvalid, err)
	}
	if err := rg.Register(spec.Name, spec.Path); err != nil {
		if derr := rg.odb.DeleteRepo(spec.Name); derr != nil {
			log.Printf("repo.add %s: rollback failed: %v", spec.Name, derr)
		}
		undo()
		return "", err
	}
	if spec.Mirror && spec.Clone != "" {
		every := time.Duration(spec.MirrorEvery) * time.Second
		if every == 0 {
			every = 24 * time.Hour
		}
		// A failed mirror record is not worth unwinding a working
		// repository over — it is one admin call to fix, and the
		// repository is already serving.
		if err := rg.odb.SetMirror(spec.Name, spec.Clone, every); err != nil {
			log.Printf("repo.add %s: mirror not recorded: %v", spec.Name, err)
		}
	}
	return spec.Path, nil
}

// mustBeEmpty refuses to create or clone over existing content. Both zit
// verbs create the directory themselves, so anything already there is
// somebody else's, and the failure mode of guessing wrong is destroying
// it on the rollback path.
func (rg *registrar) mustBeEmpty(path string) error {
	if path == "" {
		return fmt.Errorf("%w: empty path", ops.ErrInvalid)
	}
	entries, err := os.ReadDir(path)
	if os.IsNotExist(err) {
		return nil
	}
	if err != nil {
		return fmt.Errorf("%w: %v", ops.ErrInvalid, err)
	}
	if len(entries) > 0 {
		return fmt.Errorf("%w: %s already exists and is not empty", ops.ErrInvalid, path)
	}
	return nil
}

// Remove stops serving a repository and forgets it. The store on disk
// and its derived index are left alone: "stop serving this" and
// "destroy this" are different intentions and only one is reversible.
func (rg *registrar) Remove(name string) error {
	if err := rg.Unregister(name); err != nil {
		return err
	}
	if rg.odb != nil {
		if err := rg.odb.DeleteRepo(name); err != nil {
			return err
		}
	}
	return nil
}

// CloseAll releases every registration — serve's shutdown path.
func (rg *registrar) CloseAll() {
	for _, name := range rg.deps.Repos.Names() {
		if err := rg.Unregister(name); err != nil {
			log.Printf("unregister %s: %v", name, err)
		}
	}
}

// accountStore adapts forge OLTP to the ops layer's account seam. The
// role strings cross as text because that is the API's shape; parsing
// them here keeps oltp.Role out of the op surface.
type accountStore struct{ db *oltp.DB }

func (a accountStore) Accounts() ([]api.Account, error) {
	rows, err := a.db.Accounts()
	if err != nil {
		return nil, err
	}
	out := make([]api.Account, 0, len(rows))
	for _, r := range rows {
		out = append(out, api.Account{Name: r.Name, Admin: r.Admin, Created: r.Created})
	}
	return out, nil
}

func (a accountStore) AddAccount(name, password string, admin bool) error {
	return a.db.AddAccount(name, password, admin)
}

func (a accountStore) SetPassword(name, password string) error {
	return a.db.SetPassword(name, password)
}

func (a accountStore) DeleteAccount(name string) error { return a.db.DeleteAccount(name) }

func (a accountStore) Grant(account, repo, role string) error {
	r, err := oltp.ParseRole(role)
	if err != nil {
		return err
	}
	return a.db.Grant(account, repo, r)
}

func (a accountStore) Revoke(account, repo string) error { return a.db.Revoke(account, repo) }

func (a accountStore) GrantsFor(account string) (map[string]string, error) {
	grants, err := a.db.GrantsFor(account)
	if err != nil {
		return nil, err
	}
	out := make(map[string]string, len(grants))
	for repo, role := range grants {
		out[repo] = role.String()
	}
	return out, nil
}

// SetVisibility takes a repository private or public.
func (rg *registrar) SetVisibility(name, visibility string) error {
	if rg.odb == nil {
		return fmt.Errorf("no forge database: %w", ops.ErrUnavailable)
	}
	if err := rg.odb.SetRepoVisibility(name, visibility); err != nil {
		return err
	}
	rg.refreshPrivate()
	return nil
}

// refreshPrivate reloads the private set the read gate consults.
//
// Cached because the gate runs on EVERY request and the answer changes
// only when an admin changes it. Refreshed explicitly on every write
// rather than expired on a timer: a window in which a repository is
// still readable after being taken private is not a staleness problem, it is
// the feature failing.
func (rg *registrar) refreshPrivate() {
	if rg.odb == nil {
		return
	}
	set, err := rg.odb.PrivateRepos()
	if err != nil {
		log.Printf("visibility: reload: %v", err)
		return
	}
	rg.privMu.Lock()
	rg.private = set
	rg.privMu.Unlock()
}

// CanRead reports whether the identity in ctx may read repo.
//
// Public repositories are readable by anyone, which is what keeps an
// anonymous forge working. A private one requires a role on it — and a
// site admin holds every repository, so admins are covered by the same
// lookup rather than by a special case.
func (rg *registrar) CanRead(ctx context.Context, repo string) bool {
	rg.privMu.RLock()
	private := rg.private[repo]
	rg.privMu.RUnlock()
	if !private {
		return true
	}
	user := ops.User(ctx)
	if user == "" {
		return false
	}
	if rg.odb == nil {
		return false
	}
	acct, ok := rg.odb.AccountByName(user)
	if !ok {
		return false
	}
	return rg.odb.RoleOf(acct, repo) >= oltp.Read
}

// identityCtx resolves a sync request's Basic credentials into the
// context the visibility gate reads.
//
// The sync protocol carries its own login card, but that is verified
// deeper in (and only for pushes). For the visibility question — may
// this caller see the repository at all — the HTTP credential is what
// is available before a single byte of the body is parsed, which is
// where the decision has to be made.
func identityCtx(req *http.Request, db *oltp.DB) context.Context {
	ctx := req.Context()
	user, pass, ok := req.BasicAuth()
	if !ok || user == "" || db == nil {
		return ctx
	}
	if _, authed := db.Authenticate(user, pass); !authed {
		return ctx
	}
	return ops.WithUser(ctx, user)
}