// 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 (
"craftdesign.group/quarry/internal/outbound"
"database/sql"
"errors"
"fmt"
"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 ''`,
// The last pulls per mirror, not only the last one. "When did it
// stop?" and "has it always been slow?" had no answer: the row kept
// one last_at and one last_err, and 29 silent failures were how an
// outage went unnoticed (a06c9ad3).
`CREATE TABLE IF NOT EXISTS mirror_pull (
repo TEXT NOT NULL,
at TEXT NOT NULL,
ms INTEGER NOT NULL,
changed INTEGER NOT NULL,
err TEXT NOT NULL DEFAULT ''
)`,
`CREATE INDEX IF NOT EXISTS mirror_pull_repo ON mirror_pull(repo, at)`,
// Divergence: check-ins this tracking store holds that its upstream
// does not. zit has no probe for "leaves the upstream lacks"; the
// honest signal is quarry's own — a push admitted onto a tracking
// store, which the sync gate knows at the moment it admits it. It
// means someone pushed before writes were refused, or the fork
// decision has been made de facto and the store should be detached
// (5d40f151).
`ALTER TABLE mirror ADD COLUMN diverged_at TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE mirror ADD COLUMN diverged_last TEXT NOT NULL DEFAULT ''`,
}
// MirrorPullKeep is how many pulls each mirror remembers.
const MirrorPullKeep = 20
// MirrorPull is one attempt: when, how long, whether anything arrived,
// and the error if it failed.
type MirrorPull struct {
At string `json:"at"`
Ms int `json:"ms"`
Changed bool `json:"changed"`
Err string `json:"err,omitempty"`
}
// 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
// DivergedAt is when a push was last admitted onto this tracking
// store (RFC3339, empty when never): the store now holds check-ins
// the upstream lacks. It is a fact with a time, not a count — one
// push is several rounds at the gate. DivergedLast names a check-in
// when the gate knew one, else stays as it was.
DivergedAt string
DivergedLast 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.
// And https: fossil manifests are unsigned, so an http upstream can
// be rewritten in transit into check-ins nobody made (#225).
_, err := outbound.Check(endpoint)
if err != nil {
return fmt.Errorf("oltp: mirror url %q: %v", endpoint, err)
}
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.
// Whether this call makes the store a tracking mirror, as opposed
// to adjusting one that already is: no row, or a row a detach left
// disabled. Read before the upsert, which erases the difference.
var enabled int
becoming := true
if err := d.db.QueryRow(`SELECT enabled FROM mirror WHERE repo = ?`, repo).Scan(&enabled); err == nil {
becoming = enabled == 0
}
_, 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()))
if err != nil {
return err
}
// A tracking mirror starts with chat OFF. Its conversation is
// wherever the upstream keeps it, and a channel here that nobody
// asked for is an empty room on every mirror of the fleet — which
// is what the fleet showed once the tab stopped hiding on mirrors.
// An admin turns it on from Settings, or a chat bridge turns it on
// by syndicating the upstream's chat (SetChatBridge). Re-attaching
// after a detach starts over, as accept_writes does above: a
// decision about the old posture is not a decision about this one.
// Adjusting an interval or URL leaves an admin's choice alone.
if becoming {
_, err = d.db.Exec(`UPDATE repo SET chat = 0 WHERE name = ?`, repo)
}
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, diverged_at = '', diverged_last = ''
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, diverged_at, diverged_last
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, &m.DivergedAt, &m.DivergedLast); 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, diverged_at, diverged_last
FROM mirror WHERE repo = ?`, repo).
Scan(&m.Repo, &m.URL, &every, &enabled,
&m.LastAt, &m.LastOK, &m.LastErr, &m.Failures, &accept, &m.DetachedFrom, &m.DivergedAt, &m.DivergedLast)
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.
// RecordMirrorRun stamps an attempt on the mirror row and appends it
// to the mirror's pull history, keeping the last MirrorPullKeep. It
// answers the failure streak after this attempt — 0 on success — so
// the caller can decide whether this is the attempt worth telling
// somebody about.
func (d *DB) RecordMirrorRun(repo string, at time.Time, took time.Duration, changed bool, runErr error) (int, error) {
stamp := at.UTC().Format(time.RFC3339)
msg := ""
if runErr != nil {
msg = runErr.Error()
if _, err := d.db.Exec(`
UPDATE mirror SET last_at = ?, last_err = ?, failures = failures + 1
WHERE repo = ?`, stamp, msg, repo); err != nil {
return 0, err
}
} else if _, err := d.db.Exec(`
UPDATE mirror SET last_at = ?, last_ok = ?, last_err = '', failures = 0
WHERE repo = ?`, stamp, stamp, repo); err != nil {
return 0, err
}
ch := 0
if changed {
ch = 1
}
if _, err := d.db.Exec(`INSERT INTO mirror_pull (repo, at, ms, changed, err) VALUES (?, ?, ?, ?, ?)`,
repo, at.UTC().Format(time.RFC3339Nano), int(took.Milliseconds()), ch, msg); err != nil {
return 0, err
}
// Trim to the last MirrorPullKeep for this repo. Cheap: one mirror
// pulls once an hour, so this runs a few dozen times a day.
if _, err := d.db.Exec(`DELETE FROM mirror_pull WHERE repo = ? AND at NOT IN (
SELECT at FROM mirror_pull WHERE repo = ? ORDER BY at DESC LIMIT ?)`, repo, repo, MirrorPullKeep); err != nil {
return 0, err
}
var streak int
err := d.db.QueryRow(`SELECT failures FROM mirror WHERE repo = ?`, repo).Scan(&streak)
if errors.Is(err, sql.ErrNoRows) {
return 0, nil
}
return streak, err
}
// MirrorPulls lists a mirror's recent attempts, newest first.
func (d *DB) MirrorPulls(repo string, n int) ([]MirrorPull, error) {
if n <= 0 || n > MirrorPullKeep {
n = MirrorPullKeep
}
rows, err := d.db.Query(`SELECT at, ms, changed, err FROM mirror_pull WHERE repo = ? ORDER BY at DESC LIMIT ?`, repo, n)
if err != nil {
return nil, err
}
defer rows.Close()
var out []MirrorPull
for rows.Next() {
var p MirrorPull
var ch int
if err := rows.Scan(&p.At, &p.Ms, &ch, &p.Err); err != nil {
return nil, err
}
p.Changed = ch != 0
out = append(out, p)
}
return out, rows.Err()
}
// MirrorDiverged records that a push was admitted onto a tracking
// store at that moment: it now holds check-ins the upstream lacks.
// Called by the sync gate, the only place that fact is known. A
// check-in hash is kept when given; the gate usually has none.
func (d *DB) MirrorDiverged(repo, checkin string, at time.Time) error {
_, err := d.db.Exec(`UPDATE mirror SET diverged_at = ?, diverged_last = CASE WHEN ? = '' THEN diverged_last ELSE ? END WHERE repo = ?`,
at.UTC().Format(time.RFC3339), checkin, checkin, repo)
return err
}