Artifact eadba08e5fe51f000f99268c6c96ad009cdf878b656b6fa96b88f207917c4880:
- File
cmd/quarry/registrar.go
— part of check-in
[58c34af8f6]
at
2026-08-30 13:18:09
on branch trunk
— imports are visible: a cloning repository exists from the first second
repo.add registers the row, returns, and copies behind it. State is cloning until the store lands, failed with the reason if it does not, and a restart settles rows it interrupted rather than leaving them claiming to clone forever. The fleet page grows an Importing section. (user: dmestas, size: 42649) [annotate] [blame] [check-ins using]
package main import ( "archive/tar" "bytes" "compress/gzip" "context" "fmt" "io" "log" "net/http" "os" "path" "path/filepath" "strings" "sync" "sync/atomic" "time" "unicode/utf8" "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 // ciTagsDefault is the fleet's verdict-tag default (RFC 0009); a // repository's own mode overrides it either way. ciTagsDefault bool // 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 { // Startup binds the port only after every repository is registered, // so each repository's cost is the whole forge's downtime. Timed so // a slow one is nameable instead of merely suspected. regStart := time.Now() defer func() { if d := time.Since(regStart); d > 500*time.Millisecond { log.Printf("register %s: took %s", name, d.Round(time.Millisecond)) } }() 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: 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. // // BOUNDED, because this runs before the server binds its port and // the cost is a function of the store's size, not a constant: the // note here used to say ~250ms, measured on small stores. On tcl it // is 116 SECONDS and on sqlite 16, so a fleet with either in it // could not answer a request for minutes after starting — the whole // forge held hostage by a detail about one clone URL. Past the // deadline the probe is simply an unanswered question, which the // false above already handles correctly. fossilClone := false probeCtx, cancelProbe := context.WithTimeout(context.Background(), 2*time.Second) if seen, err := cli.SeenSubmanifest(probeCtx); err == nil { fossilClone = !seen } else if probeCtx.Err() != nil { log.Printf("register %s: submanifest probe timed out; fossil clone withheld", name) } else { log.Printf("register %s: submanifest probe: %v", name, err) } cancelProbe() 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. // One warm pass per repository at a time. warmReads and indexTree // each spawn SEVERAL zit subprocesses — ls, ls-under, a cat per // file, an archive — and on a store tcl's size each takes seconds. // Launched unguarded from every round that moved the tip, the next // round started another set before the last had finished and they // stacked: 43 zit children on the demo fleet, three of them the // same ls against the same hash, with load average 546 on 14 cores. // A warm pass already running IS the warm pass; a second one is // duplicate work that makes the first one slower. var warming, indexingTree atomic.Bool 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 } // The default branch is zit's resolution of "trunk" — fossil's // main-branch setting and tag propagation live there, so a repo // that renamed trunk to main answers "main". Memoised per ingest // generation like the leaf set: it changes only when the DAG does. var dbMu sync.Mutex var dbName, dbTip string var dbGen uint64 dbGen = ^uint64(0) entry.DefaultBranch = func() (string, string) { gen := ingestGen.Load() dbMu.Lock() defer dbMu.Unlock() if dbGen == gen { return dbName, dbTip } tip, name, err := cli.Info(context.Background(), "trunk") if err != nil { return "", "" // caller falls back to the index heuristic } dbName, dbTip, dbGen = name, tip, gen return dbName, dbTip } 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) // Reset drops every derived table, and the doc rows are one of // them. Without this a rebuild leaves documentation silently // unsearchable until the next commit happens to move the tip — // which on a mirror of a quiet project is never. if err == nil { if indexingTree.CompareAndSwap(false, true) { go func() { defer indexingTree.Store(false) indexTree(cli, ix) }() } } 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 { // CompareAndSwap claims the slot: false means nobody is // warming, and the winner clears it on the way out. if warming.CompareAndSwap(false, true) { go func() { defer warming.Store(false) warmReads(cli, ix) }() } if indexingTree.CompareAndSwap(false, true) { go func() { defer indexingTree.Store(false) indexTree(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)) } // warmedTip is the version whose listing has already been warmed, // warmingTip whether a warm is in flight. Both belong to this poll // loop: one repository, one warm at a time, and never twice for the // same tip. var warmingTip atomic.Bool var warmedTip atomic.Value warmedTip.Store("") // 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. // // ONLY when the tip has actually moved, and only one at a time. // This fired on EVERY poll tick as a bare goroutine whose result // was discarded — so on a store where the call takes seconds the // next tick started another before the last finished. Measured // on the demo fleet: 134 concurrent `zit ls --long`, all against // the SAME hash, load average 546 on 14 cores. A listing that is // immutable per version needs computing once per version, and // the memo behind it already knows that; what was missing was // anyone asking whether the work was already in flight. if b, ok, err := ix.TipOf("trunk"); err == nil && ok { if warmedTip.Load() != b.Tip && warmingTip.CompareAndSwap(false, true) { tip := b.Tip go func() { defer warmingTip.Store(false) if _, err := cli.Ls(context.Background(), tip); err == nil { warmedTip.Store(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) } } rg.gateSweep(name, ix) rg.carryForward(name, ix) } } // gateSweep is the bypass tripwire (RFC 0007): a protected branch that // moved without a forge accept behind it is named, loudly, within a // poll tick — because a stock zit client can accept against the store // and the store cannot refuse it; noticing is the forge's half. func (rg *registrar) gateSweep(name string, ix *index.Index) { if rg.odb == nil { return } rules, err := rg.odb.Protections(name) if err != nil { return } for _, rule := range rules { if strings.ContainsAny(rule.Branch, "*?[") { continue // a glob names no single tip to watch; a rung for later } b, ok, err := ix.TipOf(rule.Branch) if err != nil || !ok { continue } watch, seen := rg.odb.GateWatch(name, rule.Branch) if !seen { // First sight: baseline at the tip. Retroactively flagging // history would bury the signal under old noise. _ = rg.odb.SetGateWatch(name, rule.Branch, b.Tip) continue } if watch == b.Tip { continue } fresh, err := ix.CheckinsAfter(watch, 50) if err != nil { continue } for _, e := range fresh { if e.Branch != rule.Branch { continue } if ok, _ := rg.odb.ForgeAccepted(name, e.Hash); ok { continue } if err := rg.odb.AddGateBypass(name, rule.Branch, e.Hash); err == nil { log.Printf("gate: %s %s moved past its checks at %.12s — no forge accept behind it", name, rule.Branch, e.Hash) } } _ = rg.odb.SetGateWatch(name, rule.Branch, b.Tip) } } // carryForward writes a premerge pair's verdict onto the merge // check-in an accept created from that pair (RFC 0009 §premerge): the // portable record says this merge was tested before it existed. Runs // after every fold; a merge already carrying a verdict is skipped, so // the write happens once. func (rg *registrar) carryForward(name string, ix *index.Index) { switch rg.odb.CITagsMode(name) { case "off": return case "on": default: if !rg.ciTagsDefault { return } } repo, ok := rg.deps.Repos.Get(name) if !ok || repo.Index == nil { return } landed, err := ix.Landings("accepted", 10) if err != nil { return } ids := make([]string, 0, len(landed)) for _, l := range landed { ids = append(ids, l.ID) } merges, err := ix.MergesForLandings(ids) if err != nil { return } for _, l := range landed { m := merges[l.ID] if m.Checkin == "" { continue } if v, verr := ix.VerdictOf(m.Checkin); verr != nil || v != "" { continue } parents, perr := ix.ParentsOf(m.Checkin) if perr != nil || len(parents) != 2 { continue } // The tested pair was branchTip:targetTip; parent order is not // recorded, so ask both spellings. verdict := rg.deps.PairVerdict(name, "pair:"+parents[0]+":"+parents[1], l.Target) if verdict == "" { verdict = rg.deps.PairVerdict(name, "pair:"+parents[1]+":"+parents[0], l.Target) } if verdict == "" { continue } if err := repo.WriteTag(verdict, m.Checkin, "premerge"); err != nil { log.Printf("ci: carry-forward %s %.12s: %v", name, m.Checkin, err) continue } log.Printf("ci: %s %.12s %s (carried from premerge)", name, m.Checkin, verdict) } } // TriggerCI launches a repository's pipelines at a check-in — the // re-run path (RFC 0009 §triggers). Empty checkin means the tip. func (rg *registrar) TriggerCI(name, checkin string) error { if !rg.engine.Enabled() { return fmt.Errorf("no CI engine configured") } repo, ok := rg.deps.Repos.Get(name) if !ok || repo.Index == nil || repo.Zit == nil { return fmt.Errorf("no repository %q", name) } return ci.RunAt(rg.engine, ciStore{db: rg.odb}, name, ciRepo{ix: repo.Index, zit: repo.Zit}, rg.ciCloneBase+name, checkin) } // 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 } // An import is REGISTERED FIRST and copied afterwards. Cloning // inline meant the repository did not exist anywhere until the // last byte arrived: no row, no page, no progress, and an // operator watching the fleet saw nothing at all for the // minutes a large repository takes. Now the row is written // immediately in state Cloning, this call returns, and the copy // runs behind it — so the repository can be looked at while it // fills, and a failure has somewhere to be reported. if err := rg.odb.AddImportingRepo(spec.Name, spec.Path, spec.Description, spec.Visibility, spec.Clone); err != nil { return "", fmt.Errorf("%w: %v", ops.ErrInvalid, err) } if spec.Mirror { rg.recordMirror(spec) } go rg.importClone(spec) return spec.Path, nil } 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 != "" { rg.recordMirror(spec) } return spec.Path, nil } // recordMirror attaches the upstream a clone came from. 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. func (rg *registrar) recordMirror(spec api.RepoSpec) { every := time.Duration(spec.MirrorEvery) * time.Second if every == 0 { every = 24 * time.Hour } if err := rg.odb.SetMirror(spec.Name, spec.Clone, every); err != nil { log.Printf("repo.add %s: mirror not recorded: %v", spec.Name, err) } } // importClone copies the upstream into an already-registered row, then // moves it from Cloning to Ready. Every exit updates the row: a failure // that only logged would leave a repository claiming to be importing // with nothing importing it. func (rg *registrar) importClone(spec api.RepoSpec) { // The bulk lane, as before: a clone is minutes of network and // memory, and admins asking for three at once should queue rather // than run unbounded. err := rg.bulk.Do(context.Background(), func() error { return zitcli.Clone(context.Background(), spec.Clone, spec.Path) }) if err != nil { why := err.Error() if errors.Is(err, work.ErrBusy) { why = "too many imports in flight; nothing was copied" } os.RemoveAll(spec.Path) if serr := rg.odb.SetRepoState(spec.Name, oltp.Failed, why); serr != nil { log.Printf("import %s: recording failure: %v", spec.Name, serr) } log.Printf("import %s from %s: %v", spec.Name, spec.Clone, err) return } if err := rg.Register(spec.Name, spec.Path); err != nil { os.RemoveAll(spec.Path) if serr := rg.odb.SetRepoState(spec.Name, oltp.Failed, err.Error()); serr != nil { log.Printf("import %s: recording failure: %v", spec.Name, serr) } log.Printf("import %s: registering: %v", spec.Name, err) return } if err := rg.odb.SetRepoState(spec.Name, oltp.Ready, ""); err != nil { log.Printf("import %s: marking ready: %v", spec.Name, err) } log.Printf("import %s from %s: complete", spec.Name, spec.Clone) } // Imports reports what is copying now and what failed copying. Bytes // are measured from the store on disk rather than tracked in memory, so // the number survives a reader arriving mid-import and needs no // bookkeeping the copy would have to keep up to date. func (rg *registrar) Imports() ([]api.RepoImport, error) { if rg.odb == nil { return nil, nil } rows, err := rg.odb.Repos() if err != nil { return nil, err } var out []api.RepoImport for _, r := range rows { if r.State == oltp.Ready { continue } out = append(out, api.RepoImport{ Name: r.Name, URL: r.ImportURL, State: r.State, Error: r.ImportError, Started: r.Created, Bytes: dirBytes(r.Path), }) } return out, nil } // dirBytes totals a store's files. A missing directory is zero, not an // error: a failed import has already had its partial store removed. func dirBytes(path string) int64 { if path == "" { return 0 } var total int64 filepath.WalkDir(path, func(_ string, e os.DirEntry, err error) error { if err != nil || e.IsDir() { return nil //nolint:nilerr // an unreadable entry contributes nothing } if info, ierr := e.Info(); ierr == nil { total += info.Size() } return nil }) return total } // ReconcileImports settles rows left mid-import by a restart. A clone // runs as a CHILD of this server, so stopping the server kills it and // the row would otherwise claim to be cloning forever — which is // exactly what a deploy during an import produced. Marking them failed // is the honest state: the copy is not running, and the partial store // is removed so a retry starts clean. func (rg *registrar) ReconcileImports() { if rg.odb == nil { return } names, err := rg.odb.ImportingRepos() if err != nil { log.Printf("imports: listing interrupted: %v", err) return } for _, name := range names { row, ok, err := rg.odb.RepoByName(name) if err != nil || !ok { continue } if row.Path != "" { os.RemoveAll(row.Path) } if err := rg.odb.SetRepoState(name, oltp.Failed, "interrupted by a server restart before the copy finished"); err != nil { log.Printf("imports: settling %s: %v", name, err) continue } log.Printf("imports: %s was interrupted by a restart; marked failed", name) } } // 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 { // A repository whose import failed has a ROW but was never served, // so there is nothing to unregister — and refusing on that basis // would make a failed import impossible to clear from anywhere but // the database. Forgetting a registration that does not exist is // the outcome the caller asked for either way. if err := rg.Unregister(name); err != nil && !errors.Is(err, ops.ErrNotFound) { 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() } // The "" key is the site-admin marker the dashboard reads ("a site // admin's forge IS the fleet"). RoleOf already grants site admins // RepoAdmin everywhere; without this map-shaped statement of the // same fact, an admin with no explicit rows saw an empty dashboard. if acct, found := a.db.AccountByName(account); found && acct.Admin { out[""] = oltp.RepoAdmin.String() } return out, nil } // GuardGrant asks the RFC 0008 guard about a grant before the API // makes it. Callers hold s, so the guard warns rather than refuses — // the last-admin removal happens, and the caller is told. func (a accountStore) GuardGrant(actor, account, repo, role string) (string, error) { r, err := oltp.ParseRole(role) if err != nil { return "", err } return a.db.GuardCaps(actor, account, repo, r.Caps()) } // GuardRevoke is GuardGrant for taking the string away. func (a accountStore) GuardRevoke(actor, account, repo string) (string, error) { return a.db.GuardCaps(actor, account, repo, "") } // CapsAt is account.explain's seam. An unknown account is an error, // not an empty string: the op explains people, not names. func (a accountStore) CapsAt(account, repo string) (string, string, error) { if _, found := a.db.AccountByName(account); !found { return "", "", fmt.Errorf("no such account %q", account) } stored, err := a.db.CapsOf(account, repo) if err != nil { return "", "", err } return stored, string(a.db.EffectiveCaps(account, repo)), 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) } } } // indexTree files the default branch tip's FILES into the search // index: fossil's search scope "d" for documentation, plus source, // which fossil does not search at all. // // quarry rendered both as first-class surfaces — the docs tab, the file // browser — and could find neither. A reader could open // www/quickstart.wiki and get nothing searching for it. // // One tarball, not a read per file. `cat` per path costs a subprocess // each: 2,199 files on sqlite is ~31 seconds and 2,199 store opens, // against 2.2 seconds for one archive. The tarball is the whole tree in // one pass, which is exactly the shape of the question. // // Asynchronous, on the triggers a rebuild and a moved tip already // carry. Not through dagnats: dagnats is optional — a forge without one // "simply has no CI rather than a broken one" — and search must not // inherit that. This is local derivation from the store, the same class // of work as the fold, which Reindex already routes around the bus on // purpose. func indexTree(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 } // The tip has not moved since the last pass: the rows already // describe it. This is what makes the boot trigger cheap on a // restart that ingested nothing. if at, verr := ix.TreeVersion(); verr == nil && at == tip.Tip { return } ctx, cancel := context.WithTimeout(context.Background(), 15*time.Minute) defer cancel() tmp, err := os.CreateTemp("", "quarry-tree-*.tar.gz") if err != nil { return } defer os.Remove(tmp.Name()) tmp.Close() if err := cli.ArchiveTo(ctx, "tar", tip.Tip, "", tmp.Name(), treeArchiveMaxBytes); err != nil { log.Printf("tree index: archive %.12s: %v", tip.Tip, err) return } entries, skipped, err := readTreeArchive(tmp.Name()) if err != nil { log.Printf("tree index: read %.12s: %v", tip.Tip, err) return } if err := ix.ReplaceTree(tip.Tip, entries); err != nil { log.Printf("tree index: %v", err) return } var docs int for _, e := range entries { if e.Doc { docs++ } } log.Printf("tree index: %d files (%d documentation) at %.12s, %d skipped", len(entries), docs, tip.Tip, skipped) } // readTreeArchive turns the tip tarball into index rows, skipping what // full-text search cannot use: binaries, and files too large to be // prose or source anybody reads. func readTreeArchive(path string) (entries []index.TreeText, skipped int, err error) { f, err := os.Open(path) if err != nil { return nil, 0, err } defer f.Close() gz, err := gzip.NewReader(f) if err != nil { return nil, 0, err } defer gz.Close() tr := tar.NewReader(gz) buf := make([]byte, treeFileMaxBytes+1) for { h, rerr := tr.Next() if rerr == io.EOF { break } if rerr != nil { return nil, skipped, rerr } if h.Typeflag != tar.TypeReg || h.Size > treeFileMaxBytes { if h.Typeflag == tar.TypeReg { skipped++ } continue } // The archive names files under a top-level directory; the // index wants the repository-relative path a link uses. name := h.Name if i := strings.IndexByte(name, '/'); i >= 0 { name = name[i+1:] } if name == "" { continue } n, rerr := io.ReadFull(tr, buf[:h.Size]) if rerr != nil && rerr != io.ErrUnexpectedEOF { skipped++ continue } body := buf[:n] if !utf8.Valid(body) || bytes.IndexByte(body, 0) >= 0 { skipped++ // binary: a full-text index cannot use it continue } entries = append(entries, index.TreeText{ Path: name, Body: string(body), Doc: api.IsDocPath(name), }) } return entries, skipped, nil } // treeFileMaxBytes bounds one file, and treeArchiveMaxBytes the whole // tarball. Generous for prose and source, small against a vendored // amalgamation or a checked-in binary. const ( treeFileMaxBytes = 1 << 20 treeArchiveMaxBytes = 512 << 20 ) // 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 } // RFC 0004's rule for reading at all: any read letter in the // effective set, categories included. return rg.odb.EffectiveCaps(acct.Name, repo).HasAny("gjoh2r") } // 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 }