Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336package spindle
import ( "context" "database/sql" "errors" "fmt" "math/rand/v2" "os" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "tangled.org/core/rbac" "tangled.org/core/spindle/db" "tangled.org/core/spindle/engine" "tangled.org/core/spindle/models" "tangled.org/core/spindle/secrets")
// a restart can finish the wipe since the row is in the dbfunc (s *Spindle) WipeRepo(ctx context.Context, repoDid syntax.DID, reason string) error { rows, err := s.db.ReposByDid(repoDid) if err != nil { return fmt.Errorf("list repo rows: %w", err) } // force skips the turn check, otherwise an older refresh could undo the wipe err = s.beginTurn(repoDid).force(func() error { return s.claimWipe(ctx, repoDid, rows, reason) }) if err != nil { return fmt.Errorf("claim %s: %w", repoDid, err) } return s.runWipe(ctx, repoDid)}
// the caller runs cleanup after it drops the turn lockfunc (s *Spindle) claimWipeIfRows(ctx context.Context, repoDid syntax.DID, reason string) (cleanup func() error, err error) { rows, err := s.db.ReposByDid(repoDid) if err != nil { return nil, fmt.Errorf("looking up repo rows: %w", err) } if len(rows) == 0 { return nil, nil } if err := s.claimWipe(ctx, repoDid, rows, reason); err != nil { return nil, fmt.Errorf("claim %s: %w", repoDid, err) } return func() error { return s.runWipe(context.WithoutCancel(ctx), repoDid) }, nil}
func (s *Spindle) revokeRepoAccess(ctx context.Context, repoDid syntax.DID, rows []db.Repo) error { var errs []error collabs, err := s.db.ListCollaboratorsByRepoDid(repoDid) errs = append(errs, err) for _, c := range collabs { errs = append(errs, s.e.RemoveCollaborator(c.Subject.String(), rbac.ThisServer, repoDid.String())) } errs = append(errs, s.db.DeleteRepoCollaboratorsByRepoDid(repoDid)) for _, r := range rows { errs = append(errs, s.e.RemoveRepo(r.Owner.String(), rbac.ThisServer, repoDid.String())) } if s.scheduler != nil { errs = append(errs, s.scheduler.RemoveRepo(ctx, repoDid.String())) } return errors.Join(errs...)}
// the wipe row keeps a refresh from serving the repo during cleanupfunc (s *Spindle) claimWipe(ctx context.Context, repoDid syntax.DID, rows []db.Repo, reason string) error { if err := s.revokeRepoAccess(ctx, repoDid, rows); err != nil { return err } if err := s.db.ClaimWipe(repoDid, reason, wipeSteps, s.retries.clock()); err != nil { return err } for _, r := range rows { if err := s.releaseOwnerInterest(ctx, r.Owner); err != nil { s.l.Warn("releasing owner interest failed", "owner", r.Owner, "err", err) } } return nil}
// order matters, a failed step stays in the row for a retryvar wipeSteps = []string{"workflows", "leases", "artifacts", "jobs", "quota", "secrets", "webhooks", "pulls", "pipelines", "clone"}
// each step has to be safe to repeat since a retry can rerun a half finished onefunc (s *Spindle) runWipe(ctx context.Context, repoDid syntax.DID) error { w, err := s.db.GetWipe(repoDid) if errors.Is(err, sql.ErrNoRows) { return nil } if err != nil { return fmt.Errorf("reading the wipe of %s: %w", repoDid, err) } l := s.l.With("repo", repoDid, "reason", w.Reason) l.Warn("wiping repo")
var errs []error for _, step := range w.Steps { if err := s.wipeStep(ctx, repoDid, w.Reason, step); err != nil { l.Error("wipe step failed", "step", step, "err", err) errs = append(errs, fmt.Errorf("%s: %w", step, err)) continue } if err := s.db.FinishWipeStep(repoDid, step); err != nil { errs = append(errs, fmt.Errorf("%s: %w", step, err)) } } if len(errs) > 0 { attempts := w.Attempts + 1 if err := s.db.FailWipe(repoDid, attempts, s.retries.clock().Add(wipeBackoff(attempts))); err != nil { errs = append(errs, err) } return fmt.Errorf("wiped %s with %d step errors: %w", repoDid, len(errs), errors.Join(errs...)) } l.Info("repo wiped") return nil}
// no attempt cap, backoff stops growing at repoRetryMaxfunc wipeBackoff(attempts int) time.Duration { backoff := min(repoRetryBase<<min(attempts-1, 10), repoRetryMax) return backoff - rand.N(backoff/4)}
func (s *Spindle) wipeStep(ctx context.Context, repoDid syntax.DID, reason, step string) error { switch step { case "workflows": _, err := s.cancelRepoWorkflows(ctx, repoDid, reason) return err case "leases": return s.wipeLeases(ctx, repoDid) case "artifacts": refs, err := s.db.ListArtifactRefsByRepo(repoDid.String()) if err != nil { return err } var errs []error for _, ref := range refs { errs = append(errs, s.stores.Delete(ctx, ref)...) } if err := errors.Join(errs...); err != nil { // rows stay so a retry still knows which objects to delete return err } return s.db.DeleteArtifactRefsByRepo(repoDid.String()) case "jobs": return s.db.DeleteJobsByRepo(ctx, repoDid.String()) case "quota": return s.db.DeleteQuotaStateForRepo(ctx, repoDid.String()) case "secrets": return s.vault.RemoveAllSecrets(ctx, secrets.RepoIdentifier(repoDid.String())) case "webhooks": return s.db.DeleteWebhooksByRepo(repoDid) case "pulls": return s.db.DeletePullsByRepo(repoDid) case "pipelines": pipes, err := s.db.ListPipelineWorkflows(repoDid.String()) if err != nil { return err } ids := make([]models.PipelineId, 0, len(pipes)) for _, p := range pipes { ids = append(ids, models.PipelineId(p.PipelineID)) } return s.db.DeletePipelinesByRepo(repoDid.String(), ids) case "clone": return os.RemoveAll(s.newRepoPath(repoDid)) } return fmt.Errorf("unknown wipe step %q", step)}
// lease rows go last, they name the logs to deletefunc (s *Spindle) wipeLeases(ctx context.Context, repoDid syntax.DID) error { refs, err := s.db.ListArtifactRefsByRepo(repoDid.String()) if err != nil { return err } isArtifact := make(map[string]bool, len(refs)) for _, ref := range refs { isArtifact[ref] = true } leases, err := s.db.ListMillLeases() if err != nil { return err } var errs []error for _, lease := range leases { if lease.RepoDID != repoDid.String() { continue } if ref := "logs/" + lease.LeaseID + ".log"; !isArtifact[ref] { errs = append(errs, s.stores.Delete(ctx, ref)...) } } if err := errors.Join(errs...); err != nil { return err } _, err = s.db.DeleteMillLeasesByRepo(repoDid.String()) return err}
func (s *Spindle) WipeOwner(ctx context.Context, ownerDid syntax.DID, reason string) error { repos, err := s.db.AllRepos() if err != nil { return fmt.Errorf("list repos: %w", err) } var errs []error for _, r := range repos { if r.Owner != ownerDid || r.RepoDid == "" { continue } if err := s.WipeRepo(ctx, r.RepoDid, reason); err != nil { errs = append(errs, err) } } return errors.Join(errs...)}
func (s *Spindle) cancelRepoWorkflows(ctx context.Context, repoDid syntax.DID, reason string) ([]models.WorkflowId, error) { var wids []models.WorkflowId seen := map[models.WorkflowId]bool{} add := func(rkey, name string) models.WorkflowId { wid := models.WorkflowId{PipelineId: models.PipelineId(rkey), Name: name} if !seen[wid] { seen[wid] = true wids = append(wids, wid) } return wid }
pipes, err := s.db.ListPipelineWorkflows(repoDid.String()) if err != nil { return nil, fmt.Errorf("list pipeline workflows: %w", err) } for _, p := range pipes { add(string(p.PipelineID), p.Name) }
leases, err := s.db.ListMillLeases() if err != nil { return nil, fmt.Errorf("list mill leases: %w", err) } leased := map[models.WorkflowId]bool{} for _, lease := range leases { if lease.RepoDID == repoDid.String() { leased[add(lease.PipelineID, lease.Workflow)] = true } }
var errs []error for _, wid := range wids { st, err := s.db.GetStatus(wid) finished := err == nil && st.IsFinish() if !finished { if err := s.db.StatusCancelled(wid, reason, -1, s.n); err != nil { errs = append(errs, err) } } // a finished workflow keeps its lease while a debug shell holds the VM if !finished || leased[wid] { engine.CancelWorkflow(wid) for _, eng := range s.engs { if err := eng.DestroyWorkflow(ctx, wid); err != nil { errs = append(errs, err) } } } if err := os.Remove(models.LogFilePath(s.cfg.Server.LogDir, wid)); err != nil && !os.IsNotExist(err) { errs = append(errs, fmt.Errorf("remove log for %s: %w", wid, err)) } } return wids, errors.Join(errs...)}
func (s *Spindle) releaseOwnerInterest(ctx context.Context, ownerDid syntax.DID) error { if ownerDid == "" { return nil } n, err := s.db.CountReposByOwner(ownerDid) if err != nil { return err } if n > 0 { return nil } onList, err := db.AllowlistContains(s.db, ownerDid) if err != nil { return err } // admins are served without an allow-list entry if !onList && !s.cfg.Server.IsAdmin(ownerDid.String()) && s.backfill != nil { s.backfill.Untrack(ownerDid) } return nil}
// only db work here, the cleanup runs on the repo job once kickedfunc (s *Spindle) resumeWipes(ctx context.Context) { bans, err := s.db.BanList() if err != nil { s.l.Error("failed to list bans for reconcile", "err", err) return } if len(bans) == 0 { return } banned := make(map[syntax.DID]bool, len(bans)) for _, b := range bans { banned[b.SubjectDid] = true } repos, err := s.db.AllRepos() if err != nil { s.l.Error("failed to list repos for the ban reconcile", "err", err) return } claimed := map[syntax.DID]bool{} for _, r := range repos { if r.RepoDid == "" || claimed[r.RepoDid] || !banned[r.RepoDid] && !banned[r.Owner] { continue } claimed[r.RepoDid] = true err := s.beginTurn(r.RepoDid).force(func() error { rows, err := s.db.ReposByDid(r.RepoDid) if err != nil { return err } return s.claimWipe(ctx, r.RepoDid, rows, "banned") }) if err != nil { s.l.Error("ban reconcile failed", "repo", r.RepoDid, "err", err) continue } s.retries.jobs.kick(r.RepoDid) }}