413 lines
11 KiB
Go
413 lines
11 KiB
Go
// Package manager owns the running set of sync jobs.
|
|
//
|
|
// Each repository gets its own goroutine with its own timer, so a slow or
|
|
// broken repo never delays the others. A shared semaphore caps how many git
|
|
// processes run at once, which is what actually bounds memory use.
|
|
//
|
|
// Reloading is incremental: Apply diffs the new configuration against the
|
|
// running jobs and touches only what changed. Repos whose settings are
|
|
// unchanged keep their timer and their in-flight work.
|
|
package manager
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"log/slog"
|
|
"os"
|
|
"path/filepath"
|
|
"reflect"
|
|
"sort"
|
|
"sync"
|
|
"time"
|
|
|
|
"syncbot/internal/config"
|
|
"syncbot/internal/gitx"
|
|
"syncbot/internal/syncer"
|
|
)
|
|
|
|
// Status is the observable state of one repository.
|
|
type Status struct {
|
|
Name string `json:"name"`
|
|
Src string `json:"src"`
|
|
Dst string `json:"dst"`
|
|
Interval string `json:"interval"`
|
|
Syncing bool `json:"syncing"`
|
|
Refs int `json:"refs"`
|
|
Syncs int64 `json:"syncs"`
|
|
Pushes int64 `json:"pushes"`
|
|
Failures int `json:"consecutive_failures"`
|
|
LastRun time.Time `json:"last_run,omitzero"`
|
|
LastSuccess time.Time `json:"last_success,omitzero"`
|
|
NextRun time.Time `json:"next_run,omitzero"`
|
|
LastError string `json:"last_error,omitempty"`
|
|
LastDurMS int64 `json:"last_duration_ms"`
|
|
}
|
|
|
|
// Manager supervises one goroutine per repository.
|
|
type Manager struct {
|
|
ctx context.Context
|
|
log *slog.Logger
|
|
|
|
mu sync.Mutex
|
|
jobs map[string]*job
|
|
stats map[string]*Status
|
|
sem chan struct{}
|
|
conc int
|
|
workDir string
|
|
sync *syncer.Syncer
|
|
started time.Time
|
|
|
|
// wg covers every goroutine ever started, including ones already replaced
|
|
// by a reload, so shutdown does not leave a git process behind.
|
|
wg sync.WaitGroup
|
|
}
|
|
|
|
type job struct {
|
|
spec config.Job
|
|
cancel context.CancelFunc
|
|
done chan struct{}
|
|
}
|
|
|
|
// New returns a Manager whose jobs all derive from ctx.
|
|
func New(ctx context.Context, log *slog.Logger) *Manager {
|
|
return &Manager{
|
|
ctx: ctx,
|
|
log: log,
|
|
jobs: make(map[string]*job),
|
|
stats: make(map[string]*Status),
|
|
sem: make(chan struct{}, 1),
|
|
conc: 1,
|
|
started: time.Now(),
|
|
}
|
|
}
|
|
|
|
// Apply reconciles the running jobs with cfg. It never blocks on in-flight
|
|
// syncs: a job being replaced is cancelled, and its successor waits for the
|
|
// handover before touching the same mirror directory.
|
|
func (m *Manager) Apply(cfg *config.Config) error {
|
|
home, err := prepareDirs(cfg.WorkDir)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if cfg.Concurrency <= 0 {
|
|
cfg.Concurrency = 1
|
|
}
|
|
|
|
m.mu.Lock()
|
|
|
|
if m.conc != cfg.Concurrency {
|
|
// Callers release into the channel they acquired from, so swapping the
|
|
// semaphore is safe even while syncs are in flight.
|
|
m.log.Info("concurrency changed", "from", m.conc, "to", cfg.Concurrency)
|
|
m.sem = make(chan struct{}, cfg.Concurrency)
|
|
m.conc = cfg.Concurrency
|
|
}
|
|
|
|
workDirChanged := m.workDir != cfg.WorkDir
|
|
if workDirChanged {
|
|
m.workDir = cfg.WorkDir
|
|
m.sync = &syncer.Syncer{Log: m.log, Home: home}
|
|
}
|
|
|
|
want := make(map[string]config.Job, len(cfg.Jobs))
|
|
for _, j := range cfg.Jobs {
|
|
want[j.Name] = j
|
|
}
|
|
|
|
// Stop jobs that disappeared or whose resolved spec changed.
|
|
var stopped []*job
|
|
for name, j := range m.jobs {
|
|
w, keep := want[name]
|
|
if keep && !workDirChanged && reflect.DeepEqual(w, j.spec) {
|
|
continue
|
|
}
|
|
delete(m.jobs, name)
|
|
stopped = append(stopped, j)
|
|
if !keep {
|
|
delete(m.stats, name)
|
|
m.log.Info("repo removed", "repo", name)
|
|
}
|
|
}
|
|
handover := make(map[string]<-chan struct{}, len(stopped))
|
|
for _, j := range stopped {
|
|
j.cancel()
|
|
handover[j.spec.Name] = j.done
|
|
}
|
|
|
|
// Start whatever is now missing.
|
|
var added, restarted int
|
|
for i, spec := range cfg.Jobs {
|
|
if _, running := m.jobs[spec.Name]; running {
|
|
continue
|
|
}
|
|
prev := handover[spec.Name]
|
|
if prev != nil {
|
|
restarted++
|
|
} else {
|
|
added++
|
|
}
|
|
m.startLocked(spec, prev, stagger(i))
|
|
}
|
|
m.mu.Unlock()
|
|
|
|
if added+restarted+len(stopped) > 0 {
|
|
m.log.Info("configuration applied",
|
|
"repos", len(cfg.Jobs), "started", added, "restarted", restarted,
|
|
"stopped", len(stopped)-restarted, "concurrency", cfg.Concurrency)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// startLocked launches a job goroutine. m.mu must be held.
|
|
func (m *Manager) startLocked(spec config.Job, waitFor <-chan struct{}, delay time.Duration) {
|
|
ctx, cancel := context.WithCancel(m.ctx)
|
|
j := &job{spec: spec, cancel: cancel, done: make(chan struct{})}
|
|
m.jobs[spec.Name] = j
|
|
|
|
st, ok := m.stats[spec.Name]
|
|
if !ok {
|
|
st = &Status{Name: spec.Name}
|
|
m.stats[spec.Name] = st
|
|
}
|
|
// Counters survive a restart; the descriptive fields follow the new spec.
|
|
st.Src = gitx.RedactURL(spec.Src.URL)
|
|
st.Dst = gitx.RedactURL(spec.Dst.URL)
|
|
st.Interval = spec.Interval.String()
|
|
|
|
m.wg.Add(1)
|
|
go m.loop(ctx, j, waitFor, delay)
|
|
}
|
|
|
|
func (m *Manager) loop(ctx context.Context, j *job, waitFor <-chan struct{}, delay time.Duration) {
|
|
defer m.wg.Done()
|
|
defer close(j.done)
|
|
|
|
// When replacing a job, let the previous goroutine finish unwinding before
|
|
// operating on its mirror directory.
|
|
if waitFor != nil {
|
|
select {
|
|
case <-waitFor:
|
|
case <-ctx.Done():
|
|
return
|
|
}
|
|
}
|
|
|
|
log := m.log.With("repo", j.spec.Name)
|
|
log.Info("watching", "src", gitx.RedactURL(j.spec.Src.URL),
|
|
"dst", gitx.RedactURL(j.spec.Dst.URL), "interval", j.spec.Interval)
|
|
|
|
timer := time.NewTimer(delay)
|
|
defer timer.Stop()
|
|
|
|
for {
|
|
m.update(j.spec.Name, func(s *Status) { s.NextRun = time.Now().Add(delay) })
|
|
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case <-timer.C:
|
|
}
|
|
|
|
failures := m.runOnce(ctx, j.spec, log)
|
|
if ctx.Err() != nil {
|
|
return
|
|
}
|
|
|
|
delay = j.spec.Interval
|
|
if failures > 0 {
|
|
delay = backoff(j.spec.Interval, j.spec.MaxBackoff, failures)
|
|
if delay > j.spec.Interval {
|
|
log.Warn("backing off", "failures", failures, "retry_in", delay)
|
|
}
|
|
}
|
|
timer.Reset(delay)
|
|
}
|
|
}
|
|
|
|
// runOnce performs a single sync and returns the consecutive failure count.
|
|
func (m *Manager) runOnce(ctx context.Context, spec config.Job, log *slog.Logger) int {
|
|
release, err := m.acquire(ctx)
|
|
if err != nil {
|
|
return 0 // shutting down
|
|
}
|
|
defer release()
|
|
|
|
m.update(spec.Name, func(s *Status) {
|
|
s.Syncing = true
|
|
s.LastRun = time.Now()
|
|
s.Syncs++
|
|
})
|
|
|
|
res, err := m.syncer().Sync(ctx, spec)
|
|
|
|
failures := 0
|
|
m.update(spec.Name, func(s *Status) {
|
|
s.Syncing = false
|
|
s.LastDurMS = res.Duration.Milliseconds()
|
|
if err != nil {
|
|
s.Failures++
|
|
s.LastError = err.Error()
|
|
} else {
|
|
s.Failures = 0
|
|
s.LastError = ""
|
|
s.LastSuccess = time.Now()
|
|
s.Refs = res.Refs
|
|
if res.Pushed {
|
|
s.Pushes++
|
|
}
|
|
}
|
|
failures = s.Failures
|
|
})
|
|
|
|
switch {
|
|
case err != nil && ctx.Err() != nil:
|
|
log.Info("sync cancelled")
|
|
case err != nil:
|
|
log.Error("sync failed", "err", err, "consecutive_failures", failures)
|
|
case res.Pushed:
|
|
log.Info("pushed", "refs", res.Refs, "changes", len(res.Changes),
|
|
"dur", res.Duration.Round(time.Millisecond))
|
|
for _, c := range res.Changes {
|
|
log.Debug("ref updated", "change", c)
|
|
}
|
|
}
|
|
return failures
|
|
}
|
|
|
|
// RunOnce syncs every job exactly once and reports whether all succeeded. It
|
|
// is the engine behind the -once flag, for cron-style or one-shot deployments.
|
|
func (m *Manager) RunOnce(cfg *config.Config) error {
|
|
if err := m.Apply(&config.Config{ // reuse Apply's dir/semaphore setup, no jobs
|
|
WorkDir: cfg.WorkDir, Concurrency: cfg.Concurrency,
|
|
ReloadInterval: cfg.ReloadInterval, LogLevel: cfg.LogLevel, LogFormat: cfg.LogFormat,
|
|
}); err != nil {
|
|
return err
|
|
}
|
|
|
|
var wg sync.WaitGroup
|
|
errs := make([]error, len(cfg.Jobs))
|
|
for i, spec := range cfg.Jobs {
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
release, err := m.acquire(m.ctx)
|
|
if err != nil {
|
|
errs[i] = err
|
|
return
|
|
}
|
|
defer release()
|
|
|
|
log := m.log.With("repo", spec.Name)
|
|
res, err := m.syncer().Sync(m.ctx, spec)
|
|
if err != nil {
|
|
errs[i] = fmt.Errorf("%s: %w", spec.Name, err)
|
|
log.Error("sync failed", "err", err)
|
|
return
|
|
}
|
|
log.Info("synced", "refs", res.Refs, "pushed", res.Pushed,
|
|
"dur", res.Duration.Round(time.Millisecond))
|
|
}()
|
|
}
|
|
wg.Wait()
|
|
return errors.Join(errs...)
|
|
}
|
|
|
|
// Stop cancels every job and waits up to grace for them to unwind. No further
|
|
// Apply may be called afterwards.
|
|
func (m *Manager) Stop(grace time.Duration) {
|
|
m.mu.Lock()
|
|
for _, j := range m.jobs {
|
|
j.cancel()
|
|
}
|
|
m.jobs = make(map[string]*job)
|
|
m.mu.Unlock()
|
|
|
|
// Waiting on the group rather than on the current jobs also covers
|
|
// goroutines a recent reload replaced but that have not finished unwinding.
|
|
done := make(chan struct{})
|
|
go func() {
|
|
m.wg.Wait()
|
|
close(done)
|
|
}()
|
|
|
|
select {
|
|
case <-done:
|
|
case <-time.After(grace):
|
|
m.log.Warn("shutdown grace period expired; abandoning in-flight syncs", "grace", grace)
|
|
}
|
|
}
|
|
|
|
// Statuses returns a snapshot of every known repository, sorted by name.
|
|
func (m *Manager) Statuses() []Status {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
out := make([]Status, 0, len(m.stats))
|
|
for _, s := range m.stats {
|
|
out = append(out, *s)
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].Name < out[j].Name })
|
|
return out
|
|
}
|
|
|
|
// Uptime reports how long the manager has been running.
|
|
func (m *Manager) Uptime() time.Duration { return time.Since(m.started) }
|
|
|
|
// acquire takes a slot from the concurrency semaphore. The release function
|
|
// returns the slot to the same channel it came from, which keeps the count
|
|
// correct even if Apply swapped the semaphore in the meantime.
|
|
func (m *Manager) acquire(ctx context.Context) (func(), error) {
|
|
m.mu.Lock()
|
|
sem := m.sem
|
|
m.mu.Unlock()
|
|
|
|
select {
|
|
case sem <- struct{}{}:
|
|
return func() { <-sem }, nil
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
}
|
|
}
|
|
|
|
func (m *Manager) syncer() *syncer.Syncer {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
return m.sync
|
|
}
|
|
|
|
func (m *Manager) update(name string, fn func(*Status)) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
if s, ok := m.stats[name]; ok {
|
|
fn(s)
|
|
}
|
|
}
|
|
|
|
// prepareDirs creates the mirror store and the isolated HOME git/ssh will use.
|
|
func prepareDirs(workDir string) (home string, err error) {
|
|
home = filepath.Join(workDir, "home")
|
|
for _, d := range []string{filepath.Join(workDir, "mirrors"), home} {
|
|
if err := os.MkdirAll(d, 0o700); err != nil {
|
|
return "", fmt.Errorf("prepare %s: %w", d, err)
|
|
}
|
|
}
|
|
return home, nil
|
|
}
|
|
|
|
// backoff grows the retry delay geometrically, capped at max, so a repo that
|
|
// is down for hours stops hammering it (and our logs) every interval.
|
|
func backoff(interval, max time.Duration, failures int) time.Duration {
|
|
d := interval
|
|
for i := 1; i < failures && d < max; i++ {
|
|
d *= 2
|
|
}
|
|
return min(d, max)
|
|
}
|
|
|
|
// stagger spreads the first run of each repo so a restart with many repos does
|
|
// not fire every git process in the same instant.
|
|
func stagger(index int) time.Duration {
|
|
return min(time.Duration(index)*250*time.Millisecond, 15*time.Second)
|
|
}
|