Skip to content
quarry Artifact [867ad6404f]
Login

Artifact [867ad6404f]

Check-ins Using Download Hex

Artifact 867ad6404fcbf3fe11dc2790027edb92ef5bac54c98881e172dd13b54638c75b:

// mirror.go stores which upstream each served store tracks and how
// often quarry pulls from it. A mirror is forge configuration, not DAG
// state, so it belongs in the OLTP database beside accounts and webhooks
// (0001 §8) rather than in the disposable index.
//
// The remote deliberately does NOT live in the store via `zit remote`.
// That would put the same fact in two places — quarry's schedule here,
// zit's URL there — with no mechanism to keep them agreeing, and the
// failure mode is a store quietly mirroring somewhere the operator did
// not configure.
package oltp

import (
	"database/sql"
	"errors"
	"fmt"
	"net/url"
	"time"
)

const mirrorSchema = `
CREATE TABLE IF NOT EXISTS mirror (
	repo     TEXT PRIMARY KEY,        -- the forge's name for the store
	url      TEXT NOT NULL,           -- upstream sync endpoint
	every    INTEGER NOT NULL,        -- seconds between pulls
	enabled  INTEGER NOT NULL DEFAULT 1,
	last_at  TEXT NOT NULL DEFAULT '',-- last ATTEMPT, success or not
	last_ok  TEXT NOT NULL DEFAULT '',-- last success
	last_err TEXT NOT NULL DEFAULT '',
	failures INTEGER NOT NULL DEFAULT 0
);`

// mirrorMigrations are additive columns for stores written by an
// earlier schema; a duplicate-column error means the column exists.
//
// accept_writes: a mirror normally refuses local writes — a push or a
// ticket landing on a tracking store makes a fork nobody decided on.
// Set, the store is a fork that still follows its upstream, and that
// is a posture an operator chose, visible in the table.
//
// detached_from: a mirror's enabled flag going to 0 with the URL kept
// is "detached": quarry became the source of truth, but where the
// history came from is provenance, not configuration, and stays.
var mirrorMigrations = []string{
	`ALTER TABLE mirror ADD COLUMN accept_writes INTEGER NOT NULL DEFAULT 0`,
	`ALTER TABLE mirror ADD COLUMN detached_from TEXT NOT NULL DEFAULT ''`,
}

// MinMirrorInterval floors how often quarry will pull. Upstreams here
// are other people's servers: fossil-scm.org, core.tcl-lang.org,
// sqlite.org. A misconfigured schedule is not quarry's problem to
// absorb, it is theirs to absorb, so the floor is a courtesy that the
// configuration surface cannot opt out of.
const MinMirrorInterval = 60 * time.Second

// Mirror is one store's upstream tracking configuration and its recent
// history.
type Mirror struct {
	Repo    string
	URL     string
	Every   time.Duration
	Enabled bool
	// LastAt is the last attempt, LastOK the last success — kept apart
	// so "tried and failed" is distinguishable from "never due yet".
	LastAt   string
	LastOK   string
	LastErr  string
	Failures int
	// AcceptWrites: local pushes and tickets are allowed on this
	// mirror (a fork that still tracks upstream). Off by default.
	AcceptWrites bool
	// DetachedFrom is the former upstream of a store that was detached:
	// Enabled is false, the store is native now, and this is where its
	// history came from.
	DetachedFrom string
}

// Tracking reports whether the store follows an upstream right now.
func (m Mirror) Tracking() bool { return m.Enabled && m.URL != "" }

// Due reports whether this mirror should be pulled now. A mirror that
// has never run is due immediately; a disabled one never is.
func (m Mirror) Due(now time.Time) bool {
	if !m.Enabled {
		return false
	}
	if m.LastAt == "" {
		return true
	}
	last, err := time.Parse(time.RFC3339, m.LastAt)
	if err != nil {
		// An unparseable stamp is corruption, not a reason to stop
		// mirroring — treat it as "overdue" and let the next write fix it.
		return true
	}
	return now.Sub(last) >= m.Every
}

// SetMirror configures (or reconfigures) a store's upstream. Reconfiguring
// preserves the history: an operator correcting a typo in the URL has not
// made the previous failures untrue.
func (d *DB) SetMirror(repo, endpoint string, every time.Duration) error {
	if repo == "" {
		return errors.New("oltp: mirror needs a repository")
	}
	// Only http(s), for the same reason a webhook URL is restricted: the
	// server fetches this on its own initiative, so any other scheme is a
	// request to make quarry speak a protocol on someone else's behalf.
	u, err := url.Parse(endpoint)
	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
		return fmt.Errorf("oltp: mirror url must be http or https, got %q", endpoint)
	}
	if every < MinMirrorInterval {
		every = MinMirrorInterval
	}
	// A changed URL makes the recorded failures about a peer that no
	// longer exists (the commonest repair is a moved project), so they
	// are cleared with it; the same URL keeps them, because correcting
	// the interval has not made the previous failures untrue.
	_, err = d.db.Exec(`
		INSERT INTO mirror (repo, url, every, enabled) VALUES (?, ?, ?, 1)
		ON CONFLICT(repo) DO UPDATE SET
			last_err = CASE WHEN url = excluded.url THEN last_err ELSE '' END,
			failures = CASE WHEN url = excluded.url THEN failures ELSE 0 END,
			url = excluded.url, every = excluded.every, enabled = 1, detached_from = ''`,
		repo, endpoint, int64(every.Seconds()))
	return err
}

// SetMirrorWrites sets whether a mirror accepts local writes.
func (d *DB) SetMirrorWrites(repo string, accept bool) error {
	res, err := d.db.Exec(`UPDATE mirror SET accept_writes = ? WHERE repo = ?`, boolInt(accept), repo)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return fmt.Errorf("oltp: %s is not a mirror", repo)
	}
	return nil
}

// DetachMirror stops tracking and makes the store native, keeping the
// upstream as provenance. Pushes are accepted from here on; the
// header can still say where the history came from.
func (d *DB) DetachMirror(repo string) error {
	res, err := d.db.Exec(`
		UPDATE mirror SET enabled = 0, detached_from = url, last_err = '', failures = 0
		WHERE repo = ?`, repo)
	if err != nil {
		return err
	}
	if n, _ := res.RowsAffected(); n == 0 {
		return fmt.Errorf("oltp: %s is not a mirror", repo)
	}
	return nil
}

// DisableMirror stops scheduled pulls without forgetting the URL, so
// re-enabling does not mean retyping it.
func (d *DB) DisableMirror(repo string) error {
	_, err := d.db.Exec(`UPDATE mirror SET enabled = 0 WHERE repo = ?`, repo)
	return err
}

// DeleteMirror forgets a store's upstream entirely.
func (d *DB) DeleteMirror(repo string) error {
	_, err := d.db.Exec(`DELETE FROM mirror WHERE repo = ?`, repo)
	return err
}

// Mirrors lists every configured mirror, repo-ordered.
func (d *DB) Mirrors() ([]Mirror, error) {
	rows, err := d.db.Query(`
		SELECT repo, url, every, enabled, last_at, last_ok, last_err, failures,
		       accept_writes, detached_from
		FROM mirror ORDER BY repo`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	var out []Mirror
	for rows.Next() {
		var m Mirror
		var every int64
		var enabled, accept int
		if err := rows.Scan(&m.Repo, &m.URL, &every, &enabled,
			&m.LastAt, &m.LastOK, &m.LastErr, &m.Failures, &accept, &m.DetachedFrom); err != nil {
			return nil, err
		}
		m.Every = time.Duration(every) * time.Second
		m.Enabled, m.AcceptWrites = enabled != 0, accept != 0
		out = append(out, m)
	}
	return out, rows.Err()
}

// MirrorFor returns one store's mirror, if it has one.
func (d *DB) MirrorFor(repo string) (Mirror, bool, error) {
	var m Mirror
	var every int64
	var enabled, accept int
	err := d.db.QueryRow(`
		SELECT repo, url, every, enabled, last_at, last_ok, last_err, failures,
		       accept_writes, detached_from
		FROM mirror WHERE repo = ?`, repo).
		Scan(&m.Repo, &m.URL, &every, &enabled,
			&m.LastAt, &m.LastOK, &m.LastErr, &m.Failures, &accept, &m.DetachedFrom)
	if errors.Is(err, sql.ErrNoRows) {
		return Mirror{}, false, nil
	}
	if err != nil {
		return Mirror{}, false, err
	}
	m.Every = time.Duration(every) * time.Second
	m.Enabled, m.AcceptWrites = enabled != 0, accept != 0
	return m, true, nil
}

// RecordMirrorRun stamps an attempt. A success clears the error and the
// failure streak; a failure leaves last_ok alone, so "last succeeded"
// stays answerable however long a store has been broken.
func (d *DB) RecordMirrorRun(repo string, at time.Time, runErr error) error {
	stamp := at.UTC().Format(time.RFC3339)
	if runErr != nil {
		_, err := d.db.Exec(`
			UPDATE mirror SET last_at = ?, last_err = ?, failures = failures + 1
			WHERE repo = ?`, stamp, runErr.Error(), repo)
		return err
	}
	_, err := d.db.Exec(`
		UPDATE mirror SET last_at = ?, last_ok = ?, last_err = '', failures = 0
		WHERE repo = ?`, stamp, stamp, repo)
	return err
}