// mirroruv.go is RFC 0020 rung 6: a mirror can bring its upstream's
// unversioned files into the repository's bucket. After each pull that
// succeeds, quarry runs an import round as a fossil client of the
// upstream (internal/xfer/uvclient.go): `pragma uv-hash` with the
// catalog's hash, the `uvigot` catalog back, `uvgimme` for every name
// the upstream holds newer by uv.Replaces, and each `uvfile` written
// through UVWrite as `mirror:<host>`. It never sends a uvfile: a mirror
// only reads, and `y` is never needed upstream.
//
// What the round cannot do is its own record (mirror_pull's uv_changed
// and uv_err), never the pull's: a mirror does not fail because of its
// downloads.
package api
import (
"context"
"fmt"
"log"
"net/url"
"strings"
"sync"
"time"
"unicode/utf8"
"craftdesign.group/quarry/internal/ops"
"craftdesign.group/quarry/internal/uv"
"craftdesign.group/quarry/internal/xfer"
)
// MirrorUVStore is the mirror row's unversioned half
// (internal/oltp/mirroruv.go).
type MirrorUVStore interface {
SetMirrorUV(repo string, on bool) error
// SetMirrorUVCredential stores the round's login, sealing the
// password; an empty user clears it.
SetMirrorUVCredential(repo, user, password string) error
// MirrorUVLogin is the round's login and the URL it belongs to,
// read together; ok is false when repo has no mirror. The round is
// its only caller.
MirrorUVLogin(repo string) (login UVLogin, ok bool, err error)
// RecordMirrorUV stamps a round on the mirror and on the pull it
// followed, named by that pull's at: how many files it applied and
// what it could not do.
RecordMirrorUV(repo, pullAt string, changed int, msg string) error
}
// UVLogin is what an import round signs in with, one row's worth: the
// mirror's URL, whether it still tracks and imports, and the credential
// for that URL (User empty for nobody). Never leaves the round.
type UVLogin struct {
URL string
Enabled, UV bool
User, Password string
}
// UVNotShared is the round's result when the upstream's uv-hash reply
// carries no pragma: fossil answers the catalog only to a login holding
// o (Read), the same letter its pull needs, so this is the round
// signing as somebody other than the pull, usually nobody.
const UVNotShared = "upstream does not share its unversioned files with this account"
// uvNotCarried is a file the bucket's per-file limit keeps out.
const uvNotCarried = "not carried (larger than this bucket's limit)"
// The round's bounds. fossil's server sends at most its max-download
// (5 MB by default) of files a reply, at least one, and answers the
// rest with a uvigot for the client to ask again, so a large catalog
// takes several exchanges; what is left over when a bound is reached
// is asked for on the next pull.
const (
uvImportRounds = 64
uvImportTime = 10 * time.Minute
uvImportBytes = 1 << 30
)
// uvNoMatch is a uv-hash no catalog has: sent when the real one drew
// silence, to tell "already the same" from "not shared".
const uvNoMatch = "0000000000000000000000000000000000000000"
// uvBlocked is why repo cannot import unversioned files, empty when it
// can: the writes the round makes are UVWrite's, which refuses a
// repository with no bucket of its own and a private one whose bucket
// serves a public base.
func (d Deps) uvBlocked(repo string) string {
if d.MirrorUV == nil || d.UV == nil || d.Buckets == nil {
return "this forge keeps no unversioned files"
}
_, b, _, ok, err := d.uvBinding(repo)
switch {
case err != nil:
return err.Error()
case !ok:
return "the repository has no bucket of its own to put them in; bind one in its Settings"
}
if err := d.uvPublicRefusal(repo, b); err != nil {
return plainErr(err).Error()
}
return ""
}
// uvImportSlots is how many import rounds run at once across the
// fleet; uvImporting holds the mirrors with one running, so a mirror
// never has two; uvImports is every round started, for tests to wait
// on. A round runs off the caller's goroutine because the scheduler
// pulls every mirror from one loop, and a slow upstream's downloads
// must not hold up the next mirror's pull.
var (
uvImportSlots = make(chan struct{}, 2)
uvImporting sync.Map
uvImports sync.WaitGroup
// uvClosing is set by CloseImportUV; under uvStarting, so no round
// is added to uvImports once its Wait may have begun, which a
// WaitGroup forbids.
uvStarting sync.Mutex
uvClosing bool
)
// startImportUV starts m's import round when its switch is on, unless
// one is still running for it (the next pull tries again). Called by
// PullMirror after a pull that succeeded and its indexing.
func (d Deps) startImportUV(m Mirror) {
if !m.UV || d.MirrorUV == nil || d.Mirrors == nil {
return
}
// The pull this round follows, which RecordMirrorRun just wrote.
pulls, err := d.Mirrors.MirrorPulls(m.Repo, 1)
if err != nil || len(pulls) == 0 {
return
}
stop := d.Shutdown
if stop == nil {
stop = context.Background()
}
uvStarting.Lock()
defer uvStarting.Unlock()
if uvClosing || stop.Err() != nil {
return
}
if _, running := uvImporting.LoadOrStore(m.Repo, true); running {
return
}
uvImports.Add(1)
go func() {
defer uvImports.Done()
defer uvImporting.Delete(m.Repo)
select {
case uvImportSlots <- struct{}{}:
case <-stop.Done():
return
}
defer func() { <-uvImportSlots }()
// select picks at random when the slot and the stop are both
// ready; a round that won the slot after the forge stopped
// does not start.
if stop.Err() != nil {
return
}
// The round may have waited a long while for its slot. What it
// signs in as, and where it posts, is read now, from one row:
// the switch turned off since the pull cancels it, and a URL
// changed since cancels it too, so a credential is only ever
// sent to the upstream it was saved for.
login, ok, err := d.MirrorUV.MirrorUVLogin(m.Repo)
var changed int
var msg string
switch {
case err != nil:
msg = "the upstream credential: " + err.Error()
case !ok || !login.Enabled || !login.UV:
return
case login.URL != m.URL:
msg = "the upstream changed while this round waited; the next pull imports from the new one"
default:
ctx, cancel := context.WithTimeout(stop, uvImportTime)
changed, msg = d.uvImportRound(ctx, m.Repo, login)
cancel()
}
if err := d.MirrorUV.RecordMirrorUV(m.Repo, pulls[0].At, changed, clipNote(msg, uvMsgMax)); err != nil {
log.Printf("mirror: %s: recording the unversioned round: %v", m.Repo, err)
}
}()
}
// WaitImportUV waits for every import round started so far.
func WaitImportUV() { uvImports.Wait() }
// CloseImportUV refuses every round from now on and waits for those
// running. The forge cancels Deps.Shutdown and then calls this before
// closing the database, so no round writes a bucket object after its
// catalog row can be, and a pull finishing meanwhile starts none.
func CloseImportUV() {
uvStarting.Lock()
uvClosing = true
uvStarting.Unlock()
uvImports.Wait()
}
// uvImportRound is one import: how many files it applied, and what it
// could not do in words, empty when nothing.
func (d Deps) uvImportRound(ctx context.Context, repo string, login UVLogin) (int, string) {
if why := d.uvBlocked(repo); why != "" {
return 0, why
}
_, b, _, _, err := d.uvBinding(repo)
if err != nil {
return 0, err.Error()
}
files, _, err := d.UVCatalog(ctx, repo)
if err != nil {
return 0, "reading the catalog: " + err.Error()
}
have := make(map[string]uv.File, len(files))
for _, f := range files {
have[f.Name] = f
}
// The reply's cap is the client's own (xfer.MaxUVReply); a file the
// bucket would refuse is never asked for.
up := xfer.UVUpstream{URL: login.URL}
if login.User != "" {
if d.ProjectCode == nil {
return 0, "this forge cannot read the repository's project code, which a login is signed with"
}
code, err := d.ProjectCode(repo)
if err != nil {
return 0, "the project code a login is signed with: " + err.Error()
}
up.User, up.Password, up.ProjectCode = login.User, login.Password, code
}
who := "mirror:" + mirrorHost(login.URL)
reply, err := up.Exchange(ctx, "pragma uv-hash "+uv.CatalogHash(files)+"\n")
if err == nil && reply.Err == "" && reply.Pragma == "" {
// Silence is fossil's answer both to a catalog that matches and
// to a login that may not read it; a hash nothing matches tells
// the two apart, and costs one catalog when they match.
reply, err = up.Exchange(ctx, "pragma uv-hash "+uvNoMatch+"\n")
if err == nil && reply.Err == "" && reply.Pragma == "" {
return 0, UVNotShared
}
}
if err != nil {
return 0, "upstream: " + err.Error()
}
if reply.Err != "" {
return 0, "upstream refused: " + reply.Err
}
var notes []string
changed := 0
note := func(name, why string) { notes = append(notes, clipNote(name+": "+why, uvNoteMax)) }
// write applies one version and says what happened to it.
write := func(f uv.File, content []byte) {
f.Who = who
var applied bool
var err error
if content == nil && !f.Deleted {
_, applied, err = d.UVTouch(ctx, repo, f)
} else {
_, applied, err = d.UVWrite(ctx, repo, f, content)
}
switch {
case err != nil:
note(f.Name, plainErr(err).Error())
case applied:
changed++
}
}
var wanted []string
for _, x := range reply.Catalog {
if err := uv.CheckName(x.Name); err != nil {
note(x.Name, "not carried ("+err.Error()+")")
continue
}
h, exists := have[x.Name]
if !uv.Replaces(x, h, exists) {
continue
}
switch {
case x.Deleted:
// A tombstone carries no content: the client half of
// xfer.c applies it from the uvigot without asking.
write(x, nil)
case exists && !h.Deleted && h.Hash == x.Hash:
// Only the mtime moved, which xfer.c also applies from the
// uvigot alone.
write(x, nil)
case x.Size > b.LimitBytes:
note(x.Name, uvNotCarried)
default:
wanted = append(wanted, x.Name)
}
}
got := reply.Bytes
for round := 0; len(wanted) > 0; round++ {
if round >= uvImportRounds || got >= uvImportBytes || ctx.Err() != nil {
notes = append(notes, fmt.Sprintf("%d %s left for the next pull", len(wanted), plural(len(wanted), "file", "files")))
break
}
var cards strings.Builder
for _, name := range wanted {
cards.WriteString("uvgimme " + name + "\n")
}
r, err := up.Exchange(ctx, cards.String())
if err != nil {
notes = append(notes, clipNote("upstream: "+err.Error(), uvNoteMax))
break
}
if r.Err != "" {
notes = append(notes, clipNote("upstream refused: "+r.Err, uvNoteMax))
break
}
got += r.Bytes
asked := make(map[string]bool, len(wanted))
for _, name := range wanted {
asked[name] = true
}
for _, f := range r.Files {
if !asked[f.File.Name] {
continue // not asked for: a mirror takes only what it asked
}
delete(asked, f.File.Name)
switch {
case f.Bad != "":
note(f.File.Name, f.Bad)
case f.Omitted:
note(f.File.Name, "the upstream sent it without its content")
case !f.File.Deleted && int64(len(f.Content)) > b.LimitBytes:
note(f.File.Name, uvNotCarried)
default:
write(f.File, f.Content)
}
}
if len(asked) == len(wanted) {
notes = append(notes, fmt.Sprintf("the upstream sent none of the %d %s asked for", len(wanted), plural(len(wanted), "file", "files")))
break
}
next := wanted[:0]
for _, name := range wanted {
if asked[name] {
next = append(next, name)
}
}
wanted = next
}
return changed, joinNotes(notes)
}
// What one note and a round's whole message may hold. The words come
// from the upstream (an error card, a file's name), so they are bounded
// and stripped of control characters before they reach a row, a page
// or a log.
const (
uvNoteMax = 300
uvMsgMax = 2000
)
// clipNote is s with control characters as spaces, cut to max bytes on
// a character boundary.
func clipNote(s string, max int) string {
s = strings.Map(func(r rune) rune {
if r < 0x20 || r == 0x7f || r == utf8.RuneError {
return ' '
}
return r
}, s)
if len(s) <= max {
return s
}
cut := max - len("…")
for cut > 0 && !utf8.RuneStart(s[cut]) {
cut--
}
return s[:cut] + "…"
}
// joinNotes is a round's notes as one line, bounded: a catalog of a
// thousand oversized files is a count, not a page.
func joinNotes(notes []string) string {
const keep = 5
if len(notes) <= keep {
return strings.Join(notes, "; ")
}
return strings.Join(notes[:keep], "; ") + fmt.Sprintf("; and %d more", len(notes)-keep)
}
// mirrorHost is the host a mirror's imports are recorded as written by.
func mirrorHost(raw string) string {
if u, err := url.Parse(raw); err == nil && u.Hostname() != "" {
return u.Hostname()
}
return raw
}
// checkMirrorUV checks the import's fields against the mirror as it
// will be, before anything is saved, and answers the writes that apply
// them. held is the mirror as stored (zero for a new one); keeps says
// its credential survives the save, which it does only while the
// upstream's URL stays the same (oltp SetMirror drops it otherwise). A
// login given without its password keeps the held password only then.
func (d Deps) checkMirrorUV(repo string, on bool, user, password string, held Mirror, keeps bool) (func() error, error) {
if on {
if why := d.uvBlocked(repo); why != "" {
return nil, fmt.Errorf("unversioned files cannot be imported into %s: %s: %w", repo, why, ops.ErrInvalid)
}
}
setCred := true
switch {
case user == "" && password != "":
return nil, fmt.Errorf("a password needs the login it belongs to: %w", ops.ErrInvalid)
case user != "" && password == "":
if !keeps || user != held.UVUser || !held.UVCredential {
return nil, fmt.Errorf("%s needs its password on the upstream, or clear the login to read as nobody: %w", user, ops.ErrInvalid)
}
setCred = false
}
return func() error {
if setCred {
if err := d.MirrorUV.SetMirrorUVCredential(repo, user, password); err != nil {
return err
}
}
return d.MirrorUV.SetMirrorUV(repo, on)
}, nil
}
type mirrorUVIn struct {
Repo string `json:"repo" validate:"required,max=64"`
// On turns the import round on or off.
On bool `json:"on"`
// User is the upstream login the round signs with; empty clears
// the credential, and the round reads as nobody.
User string `json:"user" validate:"max=64"`
// Password is User's password on the upstream. Empty keeps the one
// held when User is unchanged.
Password string `json:"password" validate:"max=256"`
}
type mirrorUVOut struct {
Mirror Mirror `json:"mirror"`
}
func registerMirrorUV(r *ops.Registry, d Deps) error {
return ops.Register(r, ops.Op[mirrorUVIn, mirrorUVOut]{
Name: "mirror.uv",
Caps: "a",
Scope: ops.ScopeRepo,
Doc: "Turn a mirror's import of its upstream's unversioned files on or off, and set the upstream login the " +
"import signs with. Reading them needs `o` on the upstream, never `y`. The password is sealed and never " +
"returned. Refused on while the repository has no bucket of its own, or is private with a public bucket base.",
Kind: ops.Command,
Auth: ops.Token,
Route: "POST /api/v1/mirrors/uv",
Run: func(ctx context.Context, in mirrorUVIn) (mirrorUVOut, error) {
if d.Mirrors == nil || d.MirrorUV == nil {
return mirrorUVOut{}, fmt.Errorf("mirroring is not configured: %w", ops.ErrUnavailable)
}
if _, err := d.repoFor(ctx, in.Repo); err != nil {
return mirrorUVOut{}, err
}
if err := d.refuseWithout(ctx, in.Repo, "a", false); err != nil {
return mirrorUVOut{}, err
}
m, ok, err := d.Mirrors.MirrorFor(in.Repo)
if err != nil {
return mirrorUVOut{}, err
}
if !ok || !m.Enabled {
return mirrorUVOut{}, fmt.Errorf("%s is not a tracking mirror: %w", in.Repo, ops.ErrNotFound)
}
apply, err := d.checkMirrorUV(in.Repo, in.On, in.User, in.Password, m, true)
if err != nil {
return mirrorUVOut{}, err
}
if err := apply(); err != nil {
return mirrorUVOut{}, err
}
if m, _, err = d.Mirrors.MirrorFor(in.Repo); err != nil {
return mirrorUVOut{}, err
}
m.State, m.UVBlocked = mirrorState(m, time.Now()), d.uvBlocked(in.Repo)
return mirrorUVOut{Mirror: m}, nil
},
})
}