package spindle
import (
"cmp"
"context"
"errors"
"math/rand/v2"
"sync"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/spindle/db"
)
const (
repoRetryBase = 30 * time.Second
repoRetryMax = time.Hour
repoRetryAttempts = 12
repoRetryTick = 10 * time.Second
repoRetryCap = 5000
)
var errRetryTableFull = errors.New("repo retry table is full")
// the jetstream cursor moves past a failed event, so this has to live in the db
type repoRetries struct {
mu sync.Mutex
now func() time.Time
tick time.Duration
jobs repoJobs
}
func (r *repoRetries) clock() time.Time {
if r.now != nil {
return r.now()
}
return time.Now()
}
func (r *repoRetries) fail(d *db.DB, did syntax.DID) (bool, error) {
return r.failWith(
func() (int, error) {
attempts, exists, err := d.RepoRetryRow(did)
if err != nil || exists {
return attempts, err
}
// anyone can send a fake did, and new rows only fit while there's room
rows, err := d.CountRepoRetries()
if err == nil && rows >= repoRetryCap {
err = errRetryTableFull
}
return 0, err
},
func(attempts int, due time.Time) error { return d.PutRepoRetry(did, attempts, due) },
func() error { return d.DeleteRepoRetry(did) },
)
}
// counts one more failure in whichever store the callbacks reach, false once it's out of tries
func (r *repoRetries) failWith(prior func() (int, error), put func(attempts int, due time.Time) error, drop func() error) (bool, error) {
r.mu.Lock()
defer r.mu.Unlock()
n, err := prior()
if err != nil {
return false, err
}
attempts, due, retained := r.next(n)
if !retained {
return false, drop()
}
return true, put(attempts, due)
}
func (r *repoRetries) next(prior int) (attempts int, due time.Time, retained bool) {
attempts = prior + 1
if attempts > repoRetryAttempts {
return attempts, time.Time{}, false
}
backoff := min(repoRetryBase<<(attempts-1), repoRetryMax)
// jitter stops everything that failed in one outage from coming due together
backoff -= rand.N(backoff / 4)
return attempts, r.clock().Add(backoff), true
}
func (r *repoRetries) clear(d *db.DB, did syntax.DID) error {
r.mu.Lock()
defer r.mu.Unlock()
return d.DeleteRepoRetry(did)
}
func (s *Spindle) repoJobWorkerCount() int {
if n := s.cfg.Server.RepoJobWorkers; n > 0 {
return n
}
return defaultRepoJobWorkers
}
func (s *Spindle) priorityJobWorkerCount() int {
if len(s.priorityKnots()) == 0 {
return 0
}
if n := s.cfg.Server.PriorityRepoJobWorkers; n > 0 {
return n
}
return max(1, s.repoJobWorkerCount()/4)
}
// the config can list a did or a host, and both get reduced to a host here
func (s *Spindle) priorityKnots() map[string]bool {
return priorityKnotHosts(s.cfg.Server.PriorityKnots)
}
// refresh jobs run here too, and returning waits for them
func (s *Spindle) retryFailedRepos(ctx context.Context) {
workers := s.repoJobWorkerCount()
s.retries.jobs.configure(workers, s.jobKnot, s.priorityKnots())
var pool sync.WaitGroup
for range workers {
pool.Go(func() { s.retries.jobs.work(ctx, s.runRepoJob) })
}
for range s.priorityJobWorkerCount() {
pool.Go(func() { s.retries.jobs.workPriority(ctx, s.runRepoJob) })
}
defer pool.Wait()
ticker := time.NewTicker(cmp.Or(s.retries.tick, repoRetryTick))
defer ticker.Stop()
var lastPrune time.Time
// after a crash there can be untried marks, kick now instead of waiting a tick
s.kickDueRepos()
for {
select {
case <-ctx.Done():
return
case <-ticker.C:
}
s.kickDueRepos()
if time.Since(lastPrune) > time.Hour {
if _, err := s.db.PruneRepoTurns(time.Now().Add(-time.Hour)); err != nil {
s.l.Error("pruning the repo turns", "err", err)
}
if _, err := s.db.PruneRepoPushes(time.Now().Add(-pushMemory)); err != nil {
s.l.Error("pruning the recorded pushes", "err", err)
}
lastPrune = time.Now()
}
}
}
func (r *repoRetries) clearAt(d *db.DB, did syntax.DID, epoch int64) error {
r.mu.Lock()
defer r.mu.Unlock()
return d.ClearRepoRetryAt(did, epoch)
}
func (s *Spindle) failRepoRefresh(did syntax.DID) {
retained, err := s.retries.fail(s.db, did)
switch {
case errors.Is(err, errRetryTableFull):
s.l.Warn("repo retry table is full, not retrying this refresh", "repo", did)
case err != nil:
s.l.Error("saving a repo retry", "repo", did, "err", err)
case !retained:
s.l.Error("giving up on refreshing a repo", "repo", did)
}
}