package spindle
import (
"context"
"errors"
"slices"
"sync"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"tangled.org/core/api/org_tangled"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/knotclient"
"tangled.org/core/tapc"
)
const (
defaultRepoJobWorkers = 256
// unserved repos can only have a refresh pending, which keeps the cap low
recordMarksPerRepo = 1000
recordMarksPerUnservedRepo = 20
)
type jobState uint8
const (
jobQueued jobState = iota + 1
jobRunning
jobAgain
)
type queuedJob struct {
did syntax.DID
knot string
}
// one job per repo, a kick during a run folds into one more run
// knots share the pool, and a reserve stays back so new knots can start
type repoJobs struct {
mu sync.Mutex
state map[syntax.DID]jobState
queue []queuedJob
// two channels so one kind of worker can't eat the other's wakeup
wake chan struct{}
wakePriority chan struct{}
// zero until configure runs, and zero means no limits
pool int
knotOf func(syntax.DID) string
priority map[string]bool
queued map[string]int
running map[string]int
// the slot goes back under the knot it was taken from, not whatever the repo says now
runningKnot map[syntax.DID]runningJob
}
type runningJob struct {
knot string
shared bool
}
// priority is keyed by knot, the same string knotOf returns
func (j *repoJobs) configure(pool int, knotOf func(syntax.DID) string, priority map[string]bool) {
j.mu.Lock()
defer j.mu.Unlock()
j.pool, j.knotOf, j.priority = pool, knotOf, priority
}
func (j *repoJobs) wakeLocked() chan struct{} {
if j.wake == nil {
j.wake = make(chan struct{}, 1)
j.wakePriority = make(chan struct{}, 1)
j.state = map[syntax.DID]jobState{}
j.queued = map[string]int{}
j.running = map[string]int{}
j.runningKnot = map[syntax.DID]runningJob{}
}
return j.wake
}
func nudge(wake chan struct{}) {
select {
case wake <- struct{}{}:
default:
}
}
func (j *repoJobs) nudgeAllLocked() {
nudge(j.wake)
nudge(j.wakePriority)
}
func (j *repoJobs) enqueueLocked(did syntax.DID, knot string) {
j.state[did] = jobQueued
j.queue = append(j.queue, queuedJob{did, knot})
j.queued[knot]++
}
func (j *repoJobs) kick(did syntax.DID) {
j.mu.Lock()
j.wakeLocked()
switch j.state[did] {
case jobRunning:
j.state[did] = jobAgain
fallthrough
case jobQueued, jobAgain:
j.mu.Unlock()
return
}
knotOf := j.knotOf
j.mu.Unlock()
// knotOf hits the db, call it outside the lock
knot := ""
if knotOf != nil {
knot = knotOf(did)
}
j.mu.Lock()
defer j.mu.Unlock()
switch j.state[did] {
case 0:
j.enqueueLocked(did, knot)
j.nudgeAllLocked()
case jobRunning:
j.state[did] = jobAgain
}
}
func (j *repoJobs) limitLocked(knot string) int {
limit := j.pool
if others := len(j.queued) - 1; others > 0 {
limit = max(2, j.pool/(others+1))
}
if knot == "" {
limit = max(2, limit/2)
}
return min(limit, max(1, j.pool-max(2, j.pool/8)))
}
func (j *repoJobs) pickLocked(dedicated bool) int {
for i, q := range j.queue {
switch {
case dedicated:
if j.priority[q.knot] {
return i
}
case j.pool == 0 || j.running[q.knot] < j.limitLocked(q.knot):
return i
}
}
return -1
}
func (j *repoJobs) next(ctx context.Context) (syntax.DID, bool) {
return j.take(ctx, false)
}
func (j *repoJobs) nextPriority(ctx context.Context) (syntax.DID, bool) {
return j.take(ctx, true)
}
func (j *repoJobs) take(ctx context.Context, dedicated bool) (syntax.DID, bool) {
for {
j.mu.Lock()
j.wakeLocked()
wake := j.wake
if dedicated {
wake = j.wakePriority
}
if i := j.pickLocked(dedicated); i >= 0 {
q := j.queue[i]
j.queue = slices.Delete(j.queue, i, i+1)
if j.queued[q.knot]--; j.queued[q.knot] <= 0 {
delete(j.queued, q.knot)
}
j.state[q.did] = jobRunning
if !dedicated {
j.running[q.knot]++
}
j.runningKnot[q.did] = runningJob{q.knot, !dedicated}
if len(j.queue) > 0 {
j.nudgeAllLocked()
}
j.mu.Unlock()
return q.did, true
}
j.mu.Unlock()
select {
case <-wake:
case <-ctx.Done():
return "", false
}
}
}
func (j *repoJobs) finish(did syntax.DID) {
j.mu.Lock()
defer j.mu.Unlock()
j.wakeLocked()
ran := j.runningKnot[did]
delete(j.runningKnot, did)
if ran.shared {
if j.running[ran.knot]--; j.running[ran.knot] <= 0 {
delete(j.running, ran.knot)
}
}
defer j.nudgeAllLocked()
if j.state[did] != jobAgain {
delete(j.state, did)
return
}
j.enqueueLocked(did, ran.knot)
}
func (j *repoJobs) work(ctx context.Context, run func(context.Context, syntax.DID)) {
j.workWith(ctx, run, j.next)
}
func (j *repoJobs) workPriority(ctx context.Context, run func(context.Context, syntax.DID)) {
j.workWith(ctx, run, j.nextPriority)
}
func (j *repoJobs) workWith(ctx context.Context, run func(context.Context, syntax.DID), next func(context.Context) (syntax.DID, bool)) {
for {
did, ok := next(ctx)
if !ok {
return
}
// a panic can't kill the worker, and the marks are still there for the next run
_ = recovered("repo job", func() error { run(ctx, did); return nil })
j.finish(did)
}
}
type markOnlyKey struct{}
func withMarkOnly(ctx context.Context) context.Context {
return context.WithValue(ctx, markOnlyKey{}, true)
}
func markOnly(ctx context.Context) bool {
return ctx.Value(markOnlyKey{}) != nil
}
// a mark is a db row saying the repo needs a refresh, the repo's job picks it up
func (s *Spindle) markRepo(repoDid syntax.DID) error {
known, err := s.knowsRepo(repoDid)
if err != nil {
return err
}
marked, err := s.db.MarkRepoDirty(repoDid, s.retries.clock(), known, repoRetryCap)
if err != nil {
return err
}
if !marked {
s.l.Warn("repo retry table is full, dropping the mark of an unknown repo", "repo", repoDid)
return nil
}
s.retries.jobs.kick(repoDid)
return nil
}
// same as markRepo but for one record, applied by the repo's job
func (s *Spindle) markRecord(rec *tapc.RecordEventData) error {
repo, err := s.servedRepo(rec.Did)
if err != nil {
return err
}
perRepo := recordMarksPerRepo
if repo == nil {
if _, pending, err := s.db.RepoRetryRow(rec.Did); err != nil || !pending {
return err
}
perRepo = recordMarksPerUnservedRepo
}
r := db.RecordRetry{RepoDid: rec.Did, Collection: rec.Collection.String(), Rkey: rec.Rkey.String()}
if err := s.db.MarkRecordDirty(r, s.retries.clock(), perRepo); err != nil {
return err
}
s.retries.jobs.kick(rec.Did)
return nil
}
func (s *Spindle) runDueWipe(ctx context.Context, did syntax.DID) {
w, err := s.db.GetWipe(did)
if err != nil || w.Due.After(s.retries.clock()) {
return
}
if err := s.runWipe(ctx, did); err != nil && ctx.Err() == nil {
s.l.Warn("finishing a wipe failed, it will be retried", "repo", did, "err", err)
}
}
func (s *Spindle) kickDueRepos() {
now := s.retries.clock()
for _, due := range []struct {
what string
read func() ([]syntax.DID, error)
}{
{"wipes", func() ([]syntax.DID, error) { return s.db.DueWipes(now) }},
{"repo retries", func() ([]syntax.DID, error) { return s.db.DueRepoRetries(now) }},
{"record marks", func() ([]syntax.DID, error) { return s.db.DueRecordMarkRepos(now) }},
{"pull events", func() ([]syntax.DID, error) { return s.db.DuePullEventRepos(now) }},
} {
dids, err := due.read()
if err != nil {
s.l.Error("reading what is due", "what", due.what, "err", err)
continue
}
for _, did := range dids {
s.retries.jobs.kick(did)
}
}
}
// a slow knot would pin a worker forever, and the pool is shared
var repoJobBudget = 5 * time.Minute
// past this many marks the job requeues itself, which stops one repo from hogging a worker
const recordMarksPerJob = 200
var errRepoJobBudget = errors.New("the repo job ran out of time")
func (s *Spindle) runRepoJob(ctx context.Context, did syntax.DID) {
// pull events run in this job too, and the one-job-per-repo rule keeps them in order
defer s.runDuePullEvents(ctx, did)
if wiping, err := s.db.WipeExists(did); err != nil {
s.l.Error("reading a wipe", "repo", did, "err", err)
return
} else if wiping {
s.runDueWipe(ctx, did)
return
}
jobCtx, cancel := context.WithTimeoutCause(ctx, repoJobBudget, errRepoJobBudget)
defer cancel()
s.refreshAndRunMarks(jobCtx, did)
// count a timeout as a failure, or the repo comes back every tick and eats a worker
if ctx.Err() == nil && errors.Is(context.Cause(jobCtx), errRepoJobBudget) {
s.l.Warn("a repo job ran out of time, backing the repo off", "repo", did)
s.failRepoRefresh(did)
}
}
func (s *Spindle) jobKnot(did syntax.DID) string {
repo, err := s.servedRepo(did)
if err != nil || repo == nil {
return ""
}
return repo.Knot
}
func (s *Spindle) refreshAndRunMarks(ctx context.Context, did syntax.DID) {
marked, due, err := s.db.RepoRetryDue(did, s.retries.clock())
if err != nil {
s.l.Error("reading a repo retry", "repo", did, "err", err)
return
}
// a kick doesn't skip the backoff, a failed refresh still waits it out
if marked && !due {
return
}
if marked {
covered := s.collaboratorMarks(did)
switch s.refreshForJob(ctx, did) {
case refreshFailed:
return
case refreshSynced:
// a full sync already read every collaborator, their marks can go
for _, m := range covered {
if err := s.db.ClearRecordRetryAt(m.RecordRetry, m.Epoch); err != nil {
s.l.Error("clearing a record mark", "repo", did, "rkey", m.Rkey, "err", err)
}
}
}
}
s.runRecordMarks(ctx, did)
}
type refreshResult int
const (
refreshSynced refreshResult = iota
refreshTooLong
refreshNoKnot
refreshFailed
)
func (s *Spindle) collaboratorMarks(did syntax.DID) []db.RecordMark {
marks, err := s.db.DueRecordMarksOf(did, s.retries.clock())
if err != nil {
s.l.Error("reading the record marks", "repo", did, "err", err)
return nil
}
return slices.DeleteFunc(marks, func(m db.RecordMark) bool { return m.Collection != org_tangled.GraphCollaboratorNSID })
}
func (s *Spindle) refreshForJob(ctx context.Context, did syntax.DID) refreshResult {
epoch, _ := s.db.RepoRetryEpoch(did)
err := s.RefreshRepo(ctx, did)
switch {
case err == nil:
return refreshSynced
case ctx.Err() != nil:
return refreshFailed
case errors.Is(err, knotclient.ErrNoKnot):
// waiting won't give it a knot, the retry gets dropped
if err := s.retries.clear(s.db, did); err != nil {
s.l.Error("clearing a repo retry", "repo", did, "err", err)
}
return refreshNoKnot
case errors.Is(err, knotclient.ErrTooManyRecords):
// retrying won't shorten the list, the collaborator marks work through it one at a time
if err := s.retries.clearAt(s.db, did, epoch); err != nil {
s.l.Error("clearing a repo retry", "repo", did, "err", err)
}
return refreshTooLong
}
s.l.Warn("refreshing a repo failed", "repo", did, "err", err)
s.failRepoRefresh(did)
return refreshFailed
}
func (s *Spindle) runRecordMarks(ctx context.Context, did syntax.DID) {
for worked := 0; ctx.Err() == nil; {
marks, err := s.db.DueRecordMarksOf(did, s.retries.clock())
if err != nil {
s.l.Error("reading the record marks", "repo", did, "err", err)
return
}
if len(marks) == 0 {
return
}
failed := false
for _, m := range marks {
if ctx.Err() != nil {
return
}
if worked++; worked > recordMarksPerJob {
s.retries.jobs.kick(did)
return
}
if !s.runRecordMark(ctx, m) {
failed = true
}
}
// a failed mark is in backoff, looping again would just spin
if failed {
return
}
}
}
func (s *Spindle) runRecordMark(ctx context.Context, m db.RecordMark) bool {
if m.Collection == org_tangled.GraphCollaboratorNSID {
return s.applyCollaboratorMark(ctx, m)
}
rec := &tapc.RecordEventData{Did: m.RepoDid, Collection: syntax.NSID(m.Collection), Rkey: syntax.RecordKey(m.Rkey), Action: tapc.RecordUpdateAction, Live: true}
if err := s.processEvent(ctx, tapc.Event{Type: tapc.EvtRecord, Record: rec}); err != nil {
if ctx.Err() == nil {
s.l.Warn("applying a pull request record failed", "repo", m.RepoDid, "collection", m.Collection, "rkey", m.Rkey, "err", err)
}
return false
}
// clear it here, handlers drop unserved repos without clearing
if err := s.db.ClearRecordRetryAt(m.RecordRetry, m.Epoch); err != nil {
s.l.Error("clearing a record mark", "repo", m.RepoDid, "rkey", m.Rkey, "err", err)
return false
}
return true
}