Artifact [6c05af72de]

Artifact 6c05af72def2ae6261c323c449b6eb657db7cd8be1a02c71c3987fe78b6e30aa:

• File cmd/quarry/registrar.go — part of check-in [2c157378a3] at 2026-08-18 01:24:41 on branch trunk — repo: add and remove repositories without restarting

  Adding a repository meant editing the server's command line and restarting, which took every other repository down with it. The registry now lives in OLTP where 0001 §8 always placed it, and repo.add / repo.remove (admin) register and tear down at runtime.

  registrar owns a repository's whole lifetime — index handle, pipeline, sync backend, poll goroutine. The startup loop could not: it relied on defer ix.Close() reaching serve's return and on a poll loop that never stops, and both leak the moment a repository is removed.

  Add persists before it serves, and rolls the record back if opening fails: a registration that serves without being recorded vanishes at the next restart with no trace of why. Remove stops serving and forgets, but leaves the store and index on disk — 'stop serving' and 'destroy' are different intentions and only one is reversible.

  --repo stays as bootstrap and override. (user: dmestas, size: 8911) [annotate] [blame] [check-ins using]

package main

import (
	"context"
	"fmt"
	"log"
	"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
	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
	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
}

// 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 registers a repository AND records it, so it comes back on the
// next start. The order is deliberate: persist first, 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(name, path string) error {
	if rg.odb == nil {
		return fmt.Errorf("no forge database: %w", ops.ErrUnavailable)
	}
	if _, exists := rg.deps.Repos.Get(name); exists {
		return fmt.Errorf("repository %q is already served: %w", name, ops.ErrInvalid)
	}
	if err := rg.odb.AddRepo(name, path); err != nil {
		return fmt.Errorf("%w: %v", ops.ErrInvalid, err)
	}
	if err := rg.Register(name, path); err != nil {
		// Roll the record back rather than leave a registration that
		// only takes effect after a restart.
		if derr := rg.odb.DeleteRepo(name); derr != nil {
			log.Printf("repo.add %s: rollback failed: %v", name, derr)
		}
		return err
	}
	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)
		}
	}
}