// 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. Accepting
// writes was a decision about THIS upstream while tracking: a new
// upstream, or re-attaching after a detach, returns to the safe
// default and the operator allows them again on purpose.
_, 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,
accept_writes = CASE WHEN enabled = 1 AND url = excluded.url THEN accept_writes 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
}