package 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 db
func (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 lock
func (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 cleanup
func (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 retry
var 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 one
func (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 repoRetryMax
func 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 delete
func (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 kicked
func (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)
}
}