package spindle
import (
"context"
"database/sql"
"errors"
"fmt"
"log/slog"
"net/http"
"strings"
"sync"
"time"
"github.com/bluesky-social/indigo/atproto/syntax"
"github.com/hashicorp/golang-lru/v2/expirable"
"tangled.org/core/api/org_tangled"
"tangled.org/core/gitutil"
"tangled.org/core/knotfeed"
"tangled.org/core/log"
"tangled.org/core/spindle/db"
"tangled.org/core/spindle/knotclient"
"tangled.org/core/spindle/models"
"tangled.org/core/spindle/xrpc"
"tangled.org/core/tid"
"tangled.org/core/workflow"
)
const (
notifyWorkBudget = 5 * time.Minute
defaultNotifySlots = 64
notifyMinShare = 2
notifyWaitWindow = 2 * time.Minute
// unknown repos cost two reads each and get a pool of their own
notifyMaxUnknown = 2
notifyMaxUnknownPerKnot = 1
unknownRepoMemory = time.Minute
unknownRepoMemoryMax = 4096
// keep pushes an hour longer because the knot's outbox gives up after 3 days
pushMemory = 3*24*time.Hour + time.Hour
changedFileOverheadBytes = 48
changedFilesMaxBytes = 524288
changedFilesMaxCount = 8192
)
// xrpc gets the status through HTTPStatus since it can't import spindle
type notifyStatus struct {
status int
err error
}
func (e notifyStatus) Error() string { return e.err.Error() }
func (e notifyStatus) Unwrap() error { return e.err }
func (e notifyStatus) HTTPStatus() int { return e.status }
// every cap keys on this form because the repos table and did docs spell hosts differently
func canonicalKnotHost(host string) string {
name, port, hasPort := strings.Cut(strings.ToLower(host), ":")
name = strings.TrimSuffix(name, ".")
if hasPort {
return name + ":" + port
}
return name
}
type slotLimits struct {
pool int
// a new knot always finds a slot because one knot can't use up the reserve
reserve int
dedicated int
priority map[string]bool
}
func slotReserve(pool int) int { return max(2, pool/8) }
// only knots with work get entries, which keeps this from growing with every knot seen
type notifySlots struct {
mu sync.Mutex
held map[string]int
total int
prioHeld map[string]int
prioTotal int
waiting map[string]time.Time
swept time.Time
now func() time.Time
}
func (n *notifySlots) clock() time.Time {
if n.now != nil {
return n.now()
}
return time.Now()
}
// a refused knot still goes in waiting, which shrinks everyone else's share
func (n *notifySlots) acquire(host string, lim slotLimits) bool {
host = canonicalKnotHost(host)
n.mu.Lock()
defer n.mu.Unlock()
now := n.clock()
n.forgetStale(now)
if lim.priority[host] && n.prioTotal < lim.dedicated {
if n.prioHeld == nil {
n.prioHeld = map[string]int{}
}
n.prioHeld[host]++
n.prioTotal++
return true
}
limit := max(1, lim.pool-lim.reserve)
if others := n.othersWaiting(host); others > 0 {
limit = max(notifyMinShare, lim.pool/(others+1))
}
if n.total >= lim.pool || n.held[host] >= limit {
if n.waiting == nil {
n.waiting = map[string]time.Time{}
}
n.waiting[host] = now
return false
}
if n.held == nil {
n.held = map[string]int{}
}
n.held[host]++
n.total++
delete(n.waiting, host)
return true
}
func (n *notifySlots) release(host string) {
host = canonicalKnotHost(host)
n.mu.Lock()
defer n.mu.Unlock()
if n.held[host] == 0 {
n.prioTotal--
if n.prioHeld[host]--; n.prioHeld[host] == 0 {
delete(n.prioHeld, host)
}
return
}
n.total--
if n.held[host]--; n.held[host] == 0 {
delete(n.held, host)
}
}
func (n *notifySlots) inflight() int {
n.mu.Lock()
defer n.mu.Unlock()
return n.total + n.prioTotal
}
func (n *notifySlots) othersWaiting(host string) int {
count := len(n.waiting)
if _, ok := n.waiting[host]; ok {
count--
}
return count
}
func (n *notifySlots) forgetStale(now time.Time) {
if now.Sub(n.swept) < notifyWaitWindow {
return
}
n.swept = now
for host, at := range n.waiting {
if now.Sub(at) >= notifyWaitWindow {
delete(n.waiting, host)
}
}
}
func priorityKnotHosts(knots []string) map[string]bool {
hosts := map[string]bool{}
for _, knot := range knots {
host := strings.TrimSpace(knot)
if rest, ok := strings.CutPrefix(strings.ToLower(host), "did:web:"); ok {
host = strings.ReplaceAll(rest, "%3a", ":")
}
if host != "" {
hosts[canonicalKnotHost(host)] = true
}
}
return hosts
}
type notifyAdmission struct {
mu sync.Mutex
held map[string]int
total int
once sync.Once
missed *expirable.LRU[syntax.DID, struct{}]
}
func (a *notifyAdmission) acquire(host string) bool {
host = canonicalKnotHost(host)
a.mu.Lock()
defer a.mu.Unlock()
if a.total >= notifyMaxUnknown || a.held[host] >= notifyMaxUnknownPerKnot {
return false
}
if a.held == nil {
a.held = map[string]int{}
}
a.held[host]++
a.total++
return true
}
func (a *notifyAdmission) release(host string) {
host = canonicalKnotHost(host)
a.mu.Lock()
defer a.mu.Unlock()
a.total--
if a.held[host]--; a.held[host] == 0 {
delete(a.held, host)
}
}
func (a *notifyAdmission) memory() *expirable.LRU[syntax.DID, struct{}] {
a.once.Do(func() {
a.missed = expirable.NewLRU[syntax.DID, struct{}](unknownRepoMemoryMax, nil, unknownRepoMemory)
})
return a.missed
}
func (a *notifyAdmission) missedRecently(did syntax.DID) bool {
_, ok := a.memory().Get(did)
return ok
}
func (a *notifyAdmission) miss(did syntax.DID) {
a.memory().Add(did, struct{}{})
}
// an error makes the knot retry, an empty id means nothing ran
func (s *Spindle) NotifyEvent(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, event *org_tangled.EventNotifyEvent_Input_Event) (models.PipelineId, error) {
push := event.EventPush
if push == nil {
return "", errors.New("only push events come from a knot")
}
key := fmt.Sprintf("%s %s %s %s", repoDid, push.Ref, push.OldSha, push.NewSha)
// the work outlives a knot hanging up, so its retry joins it instead of redoing it
res := s.pushGroup.DoChan(key, func() (any, error) {
s.notifyWork.Add(1)
defer s.notifyWork.Done()
work, cancel := context.WithTimeout(context.WithoutCancel(ctx), notifyWorkBudget)
defer cancel()
// shutdown has to cancel this, or git keeps writing after the db closes
if s.rootCtx != nil {
defer context.AfterFunc(s.rootCtx, cancel)()
}
return s.runPush(work, k, repoDid, push)
})
select {
case r := <-res:
pipeline, _ := r.Val.(models.PipelineId)
return pipeline, r.Err
case <-ctx.Done():
return "", ctx.Err()
}
}
func (s *Spindle) acquireNotifySlot(host string) (release func(), ok bool) {
if !s.notifySlots.acquire(host, s.notifyLimits()) {
return nil, false
}
return func() { s.notifySlots.release(host) }, true
}
func (s *Spindle) notifySlotCount() int {
if n := s.cfg.Server.NotifySlots; n > 0 {
return n
}
return defaultNotifySlots
}
func (s *Spindle) notifyLimits() slotLimits {
pool := s.notifySlotCount()
dedicated := s.cfg.Server.PriorityNotifySlots
if dedicated <= 0 {
dedicated = pool / 4
}
s.notifyPriorityOnce.Do(func() { s.notifyPriorityHosts = s.priorityKnots() })
return slotLimits{pool: pool, reserve: slotReserve(pool), dedicated: dedicated, priority: s.notifyPriorityHosts}
}
func (s *Spindle) runPush(ctx context.Context, k knotclient.Knot, repoDid syntax.DID, push *org_tangled.EventPush) (pipeline models.PipelineId, err error) {
defer func() {
if err != nil && !errors.Is(err, xrpc.ErrNotifyBusy) {
log.FromContext(ctx).Error("notify event failed", "repo", repoDid, "err", err)
}
}()
defer func() { err = refusedAsStatus(err) }()
repo, err := s.repoForPush(ctx, k, repoDid)
if err != nil || repo == nil {
return "", err
}
release, ok := s.acquireNotifySlot(k.Host)
if !ok {
return "", xrpc.ErrNotifyBusy
}
defer release()
return s.handlePush(ctx, k, repo, push)
}
// a miss asks the knot because db only holds repos that passed ensureRepo
func (s *Spindle) repoForPush(ctx context.Context, k knotclient.Knot, repoDid syntax.DID) (*db.Repo, error) {
repo, err := s.db.GetRepoByDid(repoDid)
if err == nil {
return repo, nil
}
if !errors.Is(err, sql.ErrNoRows) {
return nil, fmt.Errorf("looking up repo %s: %w", repoDid, err)
}
if s.notifyAdmission.missedRecently(repoDid) {
return nil, nil
}
if !s.notifyAdmission.acquire(k.Host) {
return nil, xrpc.ErrNotifyBusy
}
defer s.notifyAdmission.release(k.Host)
repo, err = s.ensureRepo(ctx, k, repoDid)
if err != nil {
return nil, fmt.Errorf("looking up repo %s: %w", repoDid, err)
}
if repo == nil {
s.notifyAdmission.miss(repoDid)
log.FromContext(ctx).Info("ignoring event for a repo this spindle does not serve", "repo", repoDid)
}
return repo, nil
}
// the knot drops an event on any 4xx, keep those for refusals a retry can't fix
func refusedAsStatus(err error) error {
if errors.Is(err, errRevRefused) || errors.Is(err, errTooManyWorkflowFiles) {
return notifyStatus{http.StatusUnprocessableEntity, err}
}
return err
}
func (s *Spindle) handlePush(ctx context.Context, k knotclient.Knot, repo *db.Repo, push *org_tangled.EventPush) (models.PipelineId, error) {
l := log.FromContext(ctx).With("repo", repo.RepoDid, "ref", push.Ref)
if !isMaterializedRef(push.Ref) {
l.Info("skipping push outside heads and tags")
return "", nil
}
oldSha, oldOk := knotfeed.ParseObjectID(push.OldSha)
newSha, newOk := knotfeed.ParseObjectID(push.NewSha)
if !oldOk || !newOk {
// this 422 is final since a retry sends the same shas
return "", notifyStatus{http.StatusUnprocessableEntity, fmt.Errorf("push shas %q and %q are not object ids", push.OldSha, push.NewSha)}
}
if newSha.IsZero() {
l.Info("ref deleted, nothing to trigger")
return "", nil
}
// bans apply even when the repo isn't banned, because the knot names the pusher
if ban, err := s.db.GetBan(syntax.DID(*push.Actor)); err != nil {
return "", fmt.Errorf("checking bans: %w", err)
} else if ban != nil {
l.Warn("push by a banned user, running nothing", "actor", *push.Actor)
return "", nil
}
// the ban check runs before the fetch, so a banned repo costs no git work
if gate, err := s.db.Gate(&s.cfg.Server, repo.RepoDid); err != nil {
return "", fmt.Errorf("checking admission: %w", err)
} else if gate.Banned || gate.Paused {
l.Info("repo is banned or its owner isn't admitted, running nothing for the push")
return "", nil
}
fire := func() {
s.wh.FirePush(context.WithoutCancel(ctx), repo, *push.Actor, push.Ref, oldSha.String(), newSha.String())
}
if id, done, err := s.admitPush(l, repo, push, oldSha, newSha, fire); done || err != nil {
return id, err
}
repoCloneUri := s.knots.CloneURL(k, repo.RepoDid)
repoPath := s.newRepoPath(repo.RepoDid)
if err := s.syncRepo(ctx, repoCloneUri, repoPath, newSha.String()); err != nil {
return "", err
}
l.Info("synced git repo")
changedFiles, _ := changedFilesUnderBudget(ctx, l, repoPath, oldSha, newSha)
triggerRepo, err := s.buildTriggerRepo(ctx, repo)
if err != nil {
return "", fmt.Errorf("building trigger repo: %w", err)
}
if s.scheduler != nil && push.Ref == "refs/heads/"+triggerRepo.DefaultBranch {
if err := s.scheduler.RefreshRepo(ctx, *repo); err != nil {
l.Warn("failed to refresh schedules after default-branch push", "err", err)
}
}
trigger := pushTrigger(triggerRepo, push, oldSha, newSha)
pipelineId, err := s.runPipeline(ctx, models.PipelineId(tid.TID()), repo.RepoDid, trigger, changedFiles, repoCloneUri, repoPath, newSha.String(), nil, triggerRepo, false)
if errors.Is(err, db.ErrBannedSubject) {
l.Warn("dropping push for banned repo")
return "", nil
}
if err != nil {
return "", err
}
if pipelineId == "" {
l.Info("no workflow matched 'push' trigger, skipping the event")
return "", nil
}
l.Info("pipeline triggered", "pipeline", pipelineId)
return pipelineId, nil
}
// the lock covers only db reads and the webhook since a fetch can hang on a knot
func (s *Spindle) admitPush(l *slog.Logger, repo *db.Repo, push *org_tangled.EventPush, oldSha, newSha knotfeed.ObjectID, fire func()) (id models.PipelineId, done bool, err error) {
defer s.lockRepo(repo.RepoDid).Unlock()
if served, err := s.servedRepo(repo.RepoDid); err != nil {
return "", true, fmt.Errorf("looking up repo %s: %w", repo.RepoDid, err)
} else if served == nil {
l.Info("repo went away before its push ran, dropping the push")
return "", true, nil
}
// webhook fires on the first record, even if the push fails later
if first, err := s.db.RecordPush(repo.RepoDid, push.Ref, oldSha.String(), newSha.String(), time.Now()); err != nil {
return "", true, fmt.Errorf("recording the push: %w", err)
} else if first {
fire()
}
if existing, found, err := s.db.PushPipeline(repo.RepoDid.String(), push.Ref, oldSha.String(), newSha.String()); err != nil {
return "", true, fmt.Errorf("looking up the push: %w", err)
} else if found {
l.Info("push already has a pipeline, skipping the replay", "pipeline", existing)
return existing, true, nil
}
// skip ci only skips the pipeline, the webhook already fired
if hasSkipCIPushOption(push.PushOptions) {
l.Info("push requested ci skip, skipping the event")
return "", true, nil
}
return "", false, nil
}
func hasSkipCIPushOption(pushOptions []string) bool {
for _, opt := range pushOptions {
switch opt {
case "skip-ci", "ci-skip":
return true
}
}
return false
}
func pushTrigger(repo *workflow.TriggerRepo, push *org_tangled.EventPush, oldSha, newSha knotfeed.ObjectID) workflow.TriggerMetadata {
return workflow.TriggerMetadata{
Kind: string(workflow.TriggerKindPush),
Push: &workflow.PushTriggerData{
Ref: push.Ref,
OldSha: oldSha.String(),
NewSha: newSha.String(),
},
Repo: repo,
Actor: push.Actor,
}
}
func isMaterializedRef(refname string) bool {
return strings.HasPrefix(refname, "refs/heads/") || strings.HasPrefix(refname, "refs/tags/")
}
// unknown changes run the path-gated workflows too, since missing ci is worse
func changedFilesUnderBudget(ctx context.Context, l *slog.Logger, repoPath string, oldSha, newSha knotfeed.ObjectID) (files []string, known bool) {
if oldSha.IsZero() {
return nil, false
}
paths, err := changedFiles(ctx, repoPath, oldSha, newSha)
if err != nil {
l.Warn("changed files unavailable between revisions", "oldSha", oldSha, "newSha", newSha, "err", err)
return nil, false
}
admitted, ok := admitChangedFiles(paths)
if !ok {
l.Info("too many changed files to filter by path, running the workflows that have a paths filter", "oldSha", oldSha, "newSha", newSha, "files", len(paths))
return nil, false
}
return admitted, true
}
// sync is shallow so oldSha may be missing, and go-git can't read sha256 repos
func changedFiles(ctx context.Context, repoPath string, oldSha, newSha knotfeed.ObjectID) ([]string, error) {
diff := func() ([]byte, error) {
return gitutil.Output(ctx, "-C", repoPath, "diff", "--name-only", "--no-renames", "-z", oldSha.String(), newSha.String())
}
out, err := diff()
if err != nil {
if fetchErr := gitutil.Run(ctx, "-C", repoPath, "fetch", "--depth=1", "--filter=tree:0", "origin", oldSha.String()); fetchErr != nil {
return nil, fmt.Errorf("fetching the old commit: %w", fetchErr)
}
if out, err = diff(); err != nil {
return nil, err
}
}
return strings.FieldsFunc(string(out), func(r rune) bool { return r == 0 }), nil
}
func admitChangedFiles(paths []string) (admitted []string, ok bool) {
spent := 0
admitted = make([]string, 0, min(len(paths), changedFilesMaxCount))
for _, path := range paths {
spent += changedFileOverheadBytes + len(path)
if spent > changedFilesMaxBytes || len(admitted) == changedFilesMaxCount {
return nil, false
}
admitted = append(admitted, path)
}
return admitted, true
}