package main
import (
"context"
"fmt"
"log"
"net/http"
"os"
"path"
"path/filepath"
"strings"
"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()
// Probed once here, not per render: it costs ~250ms and the answer
// changes only when an S-bearing artifact first arrives. A probe
// that fails leaves it false, which withholds the fossil clone
// rather than offering one that might not work.
fossilClone := false
if seen, err := cli.SeenSubmanifest(context.Background()); err == nil {
fossilClone = !seen
} else {
log.Printf("register %s: submanifest probe: %v", name, err)
}
indexing := new(atomic.Bool)
indexing.Store(true)
// Bumped after every ingest round so memoised answers that depend
// on the DAG — the leaf set below — recompute exactly when the DAG
// can have changed, and never otherwise.
ingestGen := new(atomic.Uint64)
entry := &api.Repo{
FossilClone: fossilClone,
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
},
}
// Leaf status is zit's answer, memoised per ingest: the set only
// changes when a check-in arrives, and the call is ~10ms warm even
// on the largest store because zit's own cache holds it.
var leafMu sync.Mutex
var leafSet map[string]bool
var leafErr error
var leafGen uint64
entry.Leaves = func() (map[string]bool, error) {
gen := ingestGen.Load()
leafMu.Lock()
defer leafMu.Unlock()
if leafSet != nil && leafGen == gen {
return leafSet, leafErr
}
leafSet, leafErr = cli.Leaves(context.Background())
leafGen = gen
return leafSet, leafErr
}
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)
// In-process reindex: through the same lane as every other ingest,
// so a fleet-wide repair is staged at --ingest-limit stores at a
// time rather than a stampede; the indexing flag makes the fleet
// page say what is happening; the generation bump drops every memo
// derived from the shape being rebuilt.
entry.Reindex = func() (int, error) {
var n int
indexing.Store(true)
defer indexing.Store(false)
err := rg.ingestLane.Do(context.Background(), func() error {
var rerr error
n, rerr = pipe.Reindex()
return rerr
})
ingestGen.Add(1)
return n, err
}
var warmOnce sync.Once
entry.Sync = func() (int, error) {
n, err := pipe.Sync()
// Only a round that actually ingested something can have moved
// the DAG, so only that invalidates the memos hanging off it.
// Bumping on every round made the poll interval the memo's
// lifetime: with a two-second poll, essentially every page load
// paid the cold path — 84ms against 9ms on fossil.
if n > 0 {
ingestGen.Add(1)
}
// Warm the reader's path before a reader pays for it: the memo
// caches are per-process and keyed on the tip, so both a restart
// and a new check-in leave the next files-page visitor holding
// four cold zit subprocesses — 5.6 seconds on a store tcl's
// size. After the round that moved the tip (and once at boot),
// the same reads run here instead, in the background.
if err == nil {
fresh := n > 0
warmOnce.Do(func() { fresh = true })
if fresh {
go warmReads(cli, ix)
}
}
return n, err
}
entry.Health = func() api.RepoHealth {
s := pipe.Stats()
h := api.RepoHealth{
RoundSeconds: s.LastRound.Seconds(),
LastCount: s.LastCount,
Total: s.Total,
Fallbacks: s.Fallbacks,
Gaps: s.Gaps,
Stalled: s.Stalled,
RetryAt: retryAt(s.RetryAt),
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, entry.Sync, indexing)
return nil
}
// poll runs the store's initial ingest and then its round loop until the
// context is cancelled.
// retryAt formats the stall's next attempt, empty when not stalled.
func retryAt(t time.Time) string {
if t.IsZero() {
return ""
}
return t.UTC().Format(time.RFC3339)
}
// sync is entry.Sync — the round PLUS the generation bump that lets
// memoised DAG answers know they may be stale. The poll must not call
// the pipeline directly, or those memos would outlive the ingest that
// invalidated them.
func (rg *registrar) poll(ctx context.Context, done chan struct{}, name, path string,
cli *zitcli.Client, ix *index.Index, sync func() (int, error), indexing *atomic.Bool) {
defer close(done)
start := time.Now()
var n int
err := rg.ingestLane.Do(ctx, func() error {
var serr error
n, serr = 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 := sync()
return serr
}); err != nil {
if ctx.Err() != nil {
return
}
// The stall already announced itself once, with the cause;
// repeating it every poll interval buries the log line that
// matters under hundreds that do not.
if !errors.Is(err, ingest.ErrStalled) {
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
// Seeded BEFORE the repository is registered, so the ingest
// round that follows registration sees the first check-in
// already there rather than racing it.
if files := seedFiles(spec.Name, spec.Description, spec.License, spec.Ignore); spec.Readme || spec.License != "" || spec.Ignore {
if !spec.Readme {
delete(files, "README.md")
}
if err := zitcli.SeedFiles(context.Background(), spec.Path,
spec.Author, "Initial check-in", files); err != nil {
os.RemoveAll(spec.Path)
return "", fmt.Errorf("%w: %v", ops.ErrInvalid, err)
}
}
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
}
// warmSem bounds concurrent warms fleet-wide: a restart syncs every
// store at once, and 23 parallel zit processes over the biggest repos
// is a memory spike with no reader waiting on it.
var warmSem = make(chan struct{}, 2)
// warmReads performs the reads a repository's files page performs, so
// the per-process memos (keyed on the immutable tip hash) are hot when
// a reader arrives. Failures are irrelevant — the reader's own request
// takes the cold path, exactly as if this had never run.
func warmReads(cli *zitcli.Client, ix *index.Index) {
if cli == nil || ix == nil {
return
}
warmSem <- struct{}{}
defer func() { <-warmSem }()
tip, ok, err := ix.TipOf(ix.DefaultBranch())
if err != nil || !ok {
return
}
ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute)
defer cancel()
files, err := cli.Ls(ctx, tip.Tip)
if err != nil {
return
}
cli.LsUnder(ctx, tip.Tip, "")
for _, f := range files {
base := strings.ToLower(path.Base(f.Name))
if strings.HasPrefix(base, "readme") || strings.HasPrefix(base, "license") ||
strings.HasPrefix(base, "copying") || f.Name == ".quarry/repo.json" {
cli.Cat(ctx, f.Name, tip.Tip)
}
}
}
// Update applies a repository's settings.
func (rg *registrar) Update(name, description, visibility, homepage, topics 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 fmt.Errorf("%w: %v", ops.ErrInvalid, err)
}
if err := rg.odb.SetRepoDescription(name, description); err != nil {
return err
}
if err := rg.odb.SetRepoAbout(name, homepage, topics); err != nil {
return err
}
// Visibility is cached for the read gate, so the cache moves with
// the setting rather than a tick later. A window in which a
// repository is still readable after being taken private is the
// feature failing, not staleness.
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)
}
// Delete stops serving a repository and destroys it: the store on disk,
// its derived index, and everything in the forge database keyed to it.
//
// confirm must equal the repository's name. That is the standard
// pattern because it is the only one that works: a yes/no prompt is
// answered reflexively, while typing the name requires having read
// which repository is about to go.
//
// The store on disk is removed ONLY when quarry created it — that is,
// when it lives under the repositories root. quarry serves stores it
// did not create and does not own, and the corpus this forge runs on is
// exactly that: deleting a repository named by --repo would destroy a
// checkout somebody else manages. Those refuse, and say to unregister
// instead.
func (rg *registrar) Delete(name, confirm string) error {
if rg.odb == nil {
return fmt.Errorf("no forge database: %w", ops.ErrUnavailable)
}
if confirm != name {
return fmt.Errorf("%w: type the repository's name to confirm deletion", ops.ErrInvalid)
}
row, found, err := rg.odb.RepoByName(name)
if err != nil {
return err
}
if !found {
return fmt.Errorf("no repository %q: %w", name, ops.ErrNotFound)
}
ours, err := rg.owns(row.Path)
if err != nil {
return err
}
if !ours {
return fmt.Errorf("%w: %s is outside this node's repositories root, so quarry did not "+
"create it and will not delete it — use remove to stop serving it", ops.ErrInvalid, row.Path)
}
// Stop serving before touching anything: Unregister waits for the
// poll loop to exit and closes the index, and removing files under a
// live fold is how a deletion takes the process with it.
if err := rg.Unregister(name); err != nil && !errors.Is(err, ops.ErrNotFound) {
return err
}
if err := rg.odb.PurgeRepo(name); err != nil {
return err
}
rg.refreshPrivate()
// Derived data first — it is rebuildable, so a failure here is
// recoverable, while a half-removed store is not.
if err := os.RemoveAll(filepath.Join(rg.dir, name)); err != nil {
log.Printf("delete %s: derived data: %v", name, err)
}
return os.RemoveAll(row.Path)
}
// owns reports whether a path lies under this node's repositories root.
func (rg *registrar) owns(path string) (bool, error) {
if rg.reposRoot == "" {
return false, nil
}
root, err := filepath.Abs(rg.reposRoot)
if err != nil {
return false, err
}
abs, err := filepath.Abs(path)
if err != nil {
return false, err
}
return strings.HasPrefix(abs, root+string(os.PathSeparator)), nil
}
// Deletable reports whether this node created the repository's store.
func (rg *registrar) Deletable(name string) bool {
if rg.odb == nil {
return false
}
row, found, err := rg.odb.RepoByName(name)
if err != nil || !found {
return false
}
ours, err := rg.owns(row.Path)
return err == nil && ours
}