Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529package spindle
import ( "context" "database/sql" "encoding/json" "errors" "fmt" "slices" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/api/org_tangled" "tangled.org/core/rbac" "tangled.org/core/spindle/db" "tangled.org/core/spindle/knotclient" "tangled.org/core/spindle/models" "tangled.org/core/tapc" "tangled.org/core/tid" "tangled.org/core/workflow")
const pullEventBatch = 16
// a pull request is a ticket with a target branch plus the patches that name ittype knotTicket struct { org_tangled.TrackTicket Editor syntax.DID `json:"x-tngl-editor"`}
type knotPatch struct { org_tangled.ReviewPatch Editor syntax.DID `json:"x-tngl-editor"`}
const ( pullOpen = "open" pullClosed = "closed" pullMerged = "completed")
var pullStates = []string{pullOpen, pullClosed, pullMerged}
var pullActions = map[models.WebhookEvent]string{ models.WebhookEventPullRequestCreated: workflow.PullRequestActionOpened, models.WebhookEventPullRequestResubmitted: workflow.PullRequestActionSynchronize, models.WebhookEventPullRequestReopened: workflow.PullRequestActionReopened, models.WebhookEventPullRequestClosed: workflow.PullRequestActionClosed, models.WebhookEventPullRequestMerged: workflow.PullRequestActionMerged,}
func (s *Spindle) onTicket(ctx context.Context, rec *tapc.RecordEventData) error { return s.onPullRecord(ctx, rec, s.applyTicket)}
func (s *Spindle) onPatch(ctx context.Context, rec *tapc.RecordEventData) error { return s.onPullRecord(ctx, rec, s.applyPatch)}
// a live event won't come again, so failing before anything got stored puts the record in// the retry set, and anything after that is the stored event's to retryfunc (s *Spindle) onPullRecord(ctx context.Context, rec *tapc.RecordEventData, apply func(context.Context, *db.Repo, *tapc.RecordEventData) error) error { if markOnly(ctx) { return s.markRecord(rec) } retry := db.RecordRetry{RepoDid: rec.Did, Collection: rec.Collection.String(), Rkey: rec.Rkey.String()} err := s.applyPullRecord(ctx, rec, retry, apply) if err != nil && rec.Live && ctx.Err() == nil { s.failRecordMark(retry) } return err}
func (s *Spindle) applyPullRecord(ctx context.Context, rec *tapc.RecordEventData, retry db.RecordRetry, apply func(context.Context, *db.Repo, *tapc.RecordEventData) error) error { repo, err := s.servedRepo(rec.Did) if err != nil || repo == nil { return err } // a mark set during apply bumps the epoch, which makes the clear skip it epoch, err := s.db.RecordRetryEpoch(retry) if err != nil { return err } if err := apply(ctx, repo, rec); err != nil { return err } if err := s.db.ClearRecordRetryAt(retry, epoch); err != nil { s.l.Error("clearing a record retry", "repo", rec.Did, "rkey", retry.Rkey, "err", err) } s.retries.jobs.kick(repo.RepoDid) return nil}
// events only say that something changed, so every handler rereads the knot// and derives what happened from the state it stored last timefunc (s *Spindle) applyTicket(ctx context.Context, repo *db.Repo, rec *tapc.RecordEventData) error { rkey := rec.Rkey.String() if rec.Action == tapc.RecordDeleteAction { return s.db.ForgetPullTicket(repo.RepoDid, rkey) } k, err := s.knots.KnotOf(ctx, repo.RepoDid) if err != nil { return fmt.Errorf("finding the knot of %s: %w", repo.RepoDid, err) } _, _, err = s.syncPullTicket(ctx, k, repo, rkey) return err}
func (s *Spindle) applyPatch(ctx context.Context, repo *db.Repo, rec *tapc.RecordEventData) error { rkey := rec.Rkey.String() if rec.Action == tapc.RecordDeleteAction { return s.db.ForgetPullPatch(repo.RepoDid, rkey) } k, err := s.knots.KnotOf(ctx, repo.RepoDid) if err != nil { return fmt.Errorf("finding the knot of %s: %w", repo.RepoDid, err) } var patch knotPatch err = s.knots.GetRecord(ctx, k, repo.RepoDid, org_tangled.ReviewPatchNSID, rkey, &patch) if errors.Is(err, knotclient.ErrNotFound) { return s.db.ForgetPullPatch(repo.RepoDid, rkey) } if err != nil { return fmt.Errorf("reading patch %s/%s: %w", repo.RepoDid, rkey, err) } if patch.Head == nil { s.l.Warn("skipping patch without a head", "repo", repo.RepoDid, "rkey", rkey) return nil }
ticket, row, err := s.syncPullTicket(ctx, k, repo, patch.Ticket) if err != nil || ticket == nil { return err } // the event is created or resubmitted, which only the stored patches can tell ev, err := newPullEvent(repo, patch.Ticket, "", ticket, row.Author, patch.Editor) if err != nil { return err } stored := db.PullPatch{Rkey: rkey, Ticket: patch.Ticket, Head: patch.Head.Oid, CreatedAt: patch.CreatedAt} if patch.Source != nil { stored.SourceRepo, stored.SourceBranch = patch.Source.Repo, patch.Source.Branch } return s.whileServed(repo, func() error { return s.db.AddPullPatch(repo.RepoDid, stored, ev) })}
// holding the repo lock means a wipe can't land between the check and the writefunc (s *Spindle) whileServed(repo *db.Repo, write func() error) error { defer s.lockRepo(repo.RepoDid).Unlock() if served, err := s.servedRepo(repo.RepoDid); err != nil || served == nil { return err } return write()}
// a nil ticket means there is no pull request to speak of, an issue or a record the knot lacksfunc (s *Spindle) syncPullTicket(ctx context.Context, k knotclient.Knot, repo *db.Repo, rkey string) (*knotTicket, *db.PullTicket, error) { var ticket knotTicket err := s.knots.GetRecord(ctx, k, repo.RepoDid, org_tangled.TrackTicketNSID, rkey, &ticket) if errors.Is(err, knotclient.ErrNotFound) { return nil, nil, s.db.ForgetPullTicket(repo.RepoDid, rkey) } if err != nil { return nil, nil, fmt.Errorf("reading ticket %s/%s: %w", repo.RepoDid, rkey, err) } if ticket.TargetBranch == nil { return nil, nil, nil } state := pullOpen if ticket.State != nil { state = *ticket.State } if !slices.Contains(pullStates, state) { s.l.Warn("skipping ticket with an unknown state", "repo", repo.RepoDid, "rkey", rkey, "state", state) return nil, nil, nil } ticket.State = &state
row, err := s.db.GetPullTicket(repo.RepoDid, rkey) if errors.Is(err, sql.ErrNoRows) { versions, err := s.knots.ListRecordVersions(ctx, k, repo.RepoDid, org_tangled.TrackTicketNSID, rkey) if errors.Is(err, knotclient.ErrNotFound) { return nil, nil, s.db.ForgetPullTicket(repo.RepoDid, rkey) } if err != nil { return nil, nil, fmt.Errorf("finding who opened %s/%s: %w", repo.RepoDid, rkey, err) } // nothing to compare a first sighting with, so it fires nothing until its patch lands row = &db.PullTicket{Author: versions[0].Editor, State: state, Version: versions[len(versions)-1].Cid} err = s.whileServed(repo, func() error { return s.db.AddPullTicket(repo.RepoDid, rkey, *row) }) return &ticket, row, err } if err != nil { return nil, nil, err } if err := s.walkPullTicket(ctx, k, repo, rkey, &ticket, row); err != nil { return nil, nil, err } return &ticket, row, nil}
const maxPullVersionWalk = 64
// events don't say what changed, and a close then reopen only shows up in the version walkfunc (s *Spindle) walkPullTicket(ctx context.Context, k knotclient.Knot, repo *db.Repo, rkey string, current *knotTicket, row *db.PullTicket) error { versions, err := s.knots.ListRecordVersions(ctx, k, repo.RepoDid, org_tangled.TrackTicketNSID, rkey) if errors.Is(err, knotclient.ErrNotFound) { return s.applyPullState(repo, rkey, current, row, "", current.Editor) } if err != nil { // using the newest state here would skip the middle transitions for good return fmt.Errorf("listing the versions of ticket %s/%s: %w", repo.RepoDid, rkey, err) } newest := len(versions) - 1 start := newest if i := slices.IndexFunc(versions, func(v knotclient.RecordVersion) bool { return v.Cid == row.Version }); i >= 0 { start = i + 1 } start = max(start, len(versions)-maxPullVersionWalk) for _, v := range versions[start:] { var at knotTicket err := s.knots.GetRecordAt(ctx, k, repo.RepoDid, org_tangled.TrackTicketNSID, rkey, v.Cid, &at) if errors.Is(err, knotclient.ErrNotFound) { continue } if err != nil { return fmt.Errorf("reading ticket %s/%s at %s: %w", repo.RepoDid, rkey, v.Cid, err) } state := pullOpen if at.State != nil { state = *at.State } if !slices.Contains(pullStates, state) { continue } at.State = &state if at.TargetBranch == nil { at.TargetBranch = current.TargetBranch } if err := s.applyPullState(repo, rkey, &at, row, v.Cid, v.Editor); err != nil { return err } } return nil}
func (s *Spindle) applyPullState(repo *db.Repo, rkey string, at *knotTicket, row *db.PullTicket, version string, sender syntax.DID) error { state := *at.State event, moved := pullTransition(row.State, state) if !moved { if version == "" || version == row.Version { return nil } advance := func() error { return s.db.AdvancePullTicket(repo.RepoDid, rkey, row.Version, version, state, nil) } if err := s.whileServed(repo, advance); err != nil { return err } row.Version = version return nil } ev, err := newPullEvent(repo, rkey, event, at, row.Author, sender) if err != nil { return err } err = s.whileServed(repo, func() error { if version == "" { return s.db.SetPullTicketState(repo.RepoDid, rkey, row.State, state, ev) } return s.db.AdvancePullTicket(repo.RepoDid, rkey, row.Version, version, state, &ev) }) if err != nil { return err } row.State, row.Version = state, version return nil}
func pullTransition(from, to string) (models.WebhookEvent, bool) { switch { case from == to: return "", false case to == pullMerged: return models.WebhookEventPullRequestMerged, true case to == pullClosed: return models.WebhookEventPullRequestClosed, true default: return models.WebhookEventPullRequestReopened, true }}
// sender is whoever wrote the version that caused the event, so not the author on a mergefunc newPullEvent(repo *db.Repo, rkey string, event models.WebhookEvent, ticket *knotTicket, author, sender syntax.DID) (db.PullEvent, error) { pr := models.WebhookPullRequest{ State: "open", TargetBranch: *ticket.TargetBranch, Owner: models.WebhookUser{Did: author.String()}, CreatedAt: ticket.CreatedAt, } switch *ticket.State { case pullClosed: pr.State = "closed" case pullMerged: pr.State = "merged" } if ticket.Title != nil && ticket.Title.MarkupMarkdown_Inline != nil { pr.Title = ticket.Title.MarkupMarkdown_Inline.Text } if ticket.Body != nil && ticket.Body.MarkupMarkdown != nil { pr.Body = ticket.Body.MarkupMarkdown.Text } pull, err := json.Marshal(pr) if err != nil { return db.PullEvent{}, err } return db.PullEvent{ RepoDid: repo.RepoDid, Ticket: rkey, Event: event, Author: author, Sender: sender, Pull: pull, Pipeline: models.PipelineId(tid.TID()), }, nil}
func (s *Spindle) runDuePullEvents(ctx context.Context, did syntax.DID) { events, err := s.db.DuePullEvents(did, s.retries.clock(), pullEventBatch) if err != nil { s.l.Error("reading the pull events", "repo", did, "err", err) return } if len(events) == 0 { return } repo, err := s.servedRepo(did) if err != nil { s.l.Error("looking up a repo with pull events", "repo", did, "err", err) return } // an unserved repo's events would stay due forever unless dropped if repo == nil { if err := s.db.DeletePullEventsByRepo(did); err != nil { s.l.Error("dropping the pull events of a repo no longer served", "repo", did, "err", err) } return } if s.runPullEvents(ctx, repo, events) && len(events) == pullEventBatch { // kick to the back of the queue, or one long backlog can hog a worker s.retries.jobs.kick(did) }}
// one notify slot per event, otherwise a backlog pins the poolfunc (s *Spindle) runPullEvents(ctx context.Context, repo *db.Repo, events []db.PullEvent) bool { for _, ev := range events { release, ok := s.acquireNotifySlot(repo.Knot) if !ok { return false } // only a shutdown skips the failure mark, a budget timeout still counts as one work, cancel := context.WithTimeout(ctx, notifyWorkBudget) err := s.runPullEvent(work, repo, ev) cancel() release() if err != nil && ctx.Err() == nil { s.l.Warn("a pull request event failed", "repo", repo.RepoDid, "pull", ev.Ticket, "event", ev.Event, "err", err) s.failPullEvent(ev) } } return true}
func (s *Spindle) bannedInPull(ev db.PullEvent) (bool, error) { for _, did := range []syntax.DID{ev.Author, ev.Sender} { if ban, err := s.db.GetBan(did); err != nil || ban != nil { return ban != nil, err } } return false, nil}
func (s *Spindle) runPullEvent(ctx context.Context, repo *db.Repo, ev db.PullEvent) error { if banned, err := s.bannedInPull(ev); err != nil { return fmt.Errorf("checking bans: %w", err) } else if banned { s.l.Warn("dropping a pull request event of a banned user", "repo", repo.RepoDid, "pull", ev.Ticket, "author", ev.Author, "sender", ev.Sender) return s.db.DeletePullEvent(ev.ID) } var pr models.WebhookPullRequest if err := json.Unmarshal(ev.Pull, &pr); err != nil { return err } pr.RoundNumber = ev.Rounds if ev.Head.Sha != "" { pr.Source = &models.WebhookPullRequestSource{Sha: ev.Head.Sha, Branch: ev.Head.SourceBranch, Repo: ev.Head.SourceRepo} } // marked before the handoff, so a crash in between loses the webhook instead of sending it twice if !ev.WebhookFired { if err := s.db.MarkPullEventWebhookFired(ev.ID); err != nil { return err } s.wh.FirePullRequest(ctx, ev.Event, repo, pr, ev.Sender) } if err := s.triggerPullPipeline(ctx, repo, ev, pr.TargetBranch); err != nil { return err } return s.db.DeletePullEvent(ev.ID)}
// only an error is worth retrying, everything that would end the same way again returns nilfunc (s *Spindle) triggerPullPipeline(ctx context.Context, repo *db.Repo, ev db.PullEvent, targetBranch string) error { l := s.l.With("repo", repo.RepoDid, "pull", ev.Ticket, "event", ev.Event) // the pipeline id was minted with the event, so finding it means a retry after the enqueue if exists, err := s.db.PipelineExists(ev.Pipeline); err != nil || exists { return err } if ev.Head.Sha == "" { return nil } if ban, err := s.db.IsBanned(repo.RepoDid, repo.Owner); err != nil { return fmt.Errorf("checking bans: %w", err) } else if ban != nil { l.Warn("dropping pull trigger for banned repo") return nil } allowed, err := s.mayTriggerPull(repo.RepoDid, ev.Author, ev.Sender) if err != nil { return fmt.Errorf("authorizing pull-triggered pipeline: %w", err) } if !allowed { l.Warn("rejecting pull-triggered pipeline: actor is not authorized", "actor", ev.Sender, "author", ev.Author) return nil }
triggerRepo, err := s.buildTriggerRepo(ctx, repo) if err != nil { return fmt.Errorf("building trigger repo: %w", err) } cloneUri, repoPath, _, err := s.resolveCheckout(ctx, repo.RepoDid, "") if err != nil { return err } trigger := pullTrigger(triggerRepo, ev, targetBranch) pipelineId, err := s.runPipeline(ctx, ev.Pipeline, repo.RepoDid, trigger, nil, cloneUri, repoPath, ev.Head.Sha, nil, triggerRepo) switch { case errors.Is(err, errRevRefused), errors.Is(err, db.ErrBannedSubject): l.Warn("dropping pull trigger", "sha", ev.Head.Sha, "err", err) return nil case err != nil: return err case pipelineId == "": l.Info("no workflow matched 'pull_request' trigger, skipping the event") return nil } l.Info("pipeline triggered", "pipeline", pipelineId) return nil}
// random users don't get to run pipelines, so the author needs push access, and// whoever caused the event has to be the author or have push access toofunc (s *Spindle) mayTriggerPull(repoDid, author, actor syntax.DID) (bool, error) { allowed, err := s.e.IsPushAllowed(author.String(), rbac.ThisServer, repoDid.String()) if err != nil || !allowed || actor == author { return allowed, err } return s.e.IsPushAllowed(actor.String(), rbac.ThisServer, repoDid.String())}
// the head is always checked out from the target repo, and a fork's branch name// would pass for the target's own branch in refs and cache scopes, so only a// branch of the target repo becomes the source branchfunc pullTrigger(repo *workflow.TriggerRepo, ev db.PullEvent, targetBranch string) workflow.TriggerMetadata { action := pullActions[ev.Event] sourceBranch := "" if ev.Head.SourceRepo == ev.RepoDid.String() { sourceBranch = ev.Head.SourceBranch } pull := fmt.Sprintf("at://%s/%s/%s", ev.RepoDid, org_tangled.TrackTicketNSID, ev.Ticket) actor := ev.Sender.String() return workflow.TriggerMetadata{ Kind: string(workflow.TriggerKindPullRequest), PullRequest: &workflow.PullRequestTriggerData{ Action: &action, Pull: &pull, SourceBranch: sourceBranch, SourceSha: ev.Head.Sha, TargetBranch: targetBranch, }, Repo: repo, Actor: &actor, }}
func (s *Spindle) failPullEvent(ev db.PullEvent) { retained, err := s.retries.failWith( func() (int, error) { return ev.Attempts, nil }, func(attempts int, due time.Time) error { return s.db.PutPullEventRetry(ev.ID, attempts, due) }, func() error { return s.db.DeletePullEvent(ev.ID) }, ) switch { case err != nil: s.l.Error("saving a pull event retry", "repo", ev.RepoDid, "pull", ev.Ticket, "err", err) case !retained: s.l.Error("giving up on a pull request event", "repo", ev.RepoDid, "pull", ev.Ticket, "event", ev.Event) }}
func (s *Spindle) failRecordMark(r db.RecordRetry) { retained, err := s.retries.failWith( func() (int, error) { return s.db.RecordRetryAttempts(r) }, func(attempts int, due time.Time) error { return s.db.PutRecordRetry(r, attempts, due) }, func() error { return s.db.DeleteRecordRetry(r) }, ) switch { case err != nil: s.l.Error("saving a record retry", "repo", r.RepoDid, "collection", r.Collection, "rkey", r.Rkey, "err", err) case !retained: s.l.Error("giving up on a pull request record", "repo", r.RepoDid, "collection", r.Collection, "rkey", r.Rkey) }}
func (s *Spindle) servedRepo(repoDid syntax.DID) (*db.Repo, error) { repo, err := s.db.GetRepoByDid(repoDid) if errors.Is(err, sql.ErrNoRows) { return nil, nil } return repo, err}