Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520package spindle
import ( "context" _ "embed" "errors" "fmt" "io" "log/slog" "maps" "net/http" "net/url" "path/filepath" "regexp" "sort" "strconv" "strings" "sync" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "github.com/go-chi/chi/v5" "github.com/hashicorp/go-version" "go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp" "go.opentelemetry.io/otel/attribute" "go.opentelemetry.io/otel/codes" "golang.org/x/sync/singleflight" "tangled.org/core/api/org_tangled" "tangled.org/core/gitutil" "tangled.org/core/idresolver" "tangled.org/core/jetstream" "tangled.org/core/log" "tangled.org/core/notifier" "tangled.org/core/rbac" "tangled.org/core/spindle/artifactstore" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" "tangled.org/core/spindle/engine" "tangled.org/core/spindle/engines/dummy" "tangled.org/core/spindle/engines/nixery" "tangled.org/core/spindle/identity" "tangled.org/core/spindle/knotclient" "tangled.org/core/spindle/mill" "tangled.org/core/spindle/mill/executor" "tangled.org/core/spindle/models" "tangled.org/core/spindle/netguard" "tangled.org/core/spindle/observability" "tangled.org/core/spindle/oidc" pipelinecodec "tangled.org/core/spindle/pipeline" "tangled.org/core/spindle/quota" "tangled.org/core/spindle/secrets" "tangled.org/core/spindle/storage" "tangled.org/core/spindle/webhook" "tangled.org/core/spindle/xrpc" "tangled.org/core/tid" "tangled.org/core/workflow" "tangled.org/core/xrpc/serviceauth")
//go:embed motdvar defaultMotd []byte
const ( rbacDomain = "thisserver" executorShutdownTimeout = 8 * time.Minute serverShutdownTimeout = 10 * time.Second dequeueRetryDelay = time.Second
// sparse-checkout wants a leading slash to anchor the pattern at the root sparseWorkflowDir = "/" + workflow.WorkflowDir)
type executorClient interface { Connect(context.Context) Drain(context.Context) error RegisterMetrics(*observability.Metrics) SetQuotaClient(*executor.QuotaClient)}
type Spindle struct { jc *jetstream.JetstreamClient backfill *backfill db *db.DB e *rbac.Enforcer l *slog.Logger n *notifier.Notifier wh *webhook.Service metrics *observability.Metrics engs map[string]models.Engine debugKeys debugKeyCache jobWake chan struct{} jobWorkers sync.WaitGroup jobsMu sync.Mutex jobsByOwner map[string]int scheduler *pipelineScheduler cfg *config.Config res *idresolver.Resolver vault secrets.Manager cache storage.Storage motd []byte motdMu sync.RWMutex rootCtx context.Context rootCancel context.CancelFunc store artifactstore.Store stores *artifactstore.Stores reader artifactstore.Reader qm *quota.Manager // set only when this spindle hosts the mill or joins one as an executor mill *mill.Mill oidc *oidc.Issuer identity http.HandlerFunc exec executorClient repoLocks stripedLocks turns turnClock retries repoRetries knots *knotclient.Client
pushGroup singleflight.Group notifySlots notifySlots notifyWork sync.WaitGroup notifyPriorityOnce sync.Once notifyPriorityHosts map[string]bool notifyAdmission notifyAdmission workflowDefs workflowDefCache}
func newCacheStore(ctx context.Context, cfg *config.Config) (storage.Storage, error) { switch cfg.Cache.Backend { case "": return nil, nil case "disk": dir := cfg.Cache.DiskDir if dir == "" { dir = filepath.Join(filepath.Dir(cfg.Server.DBPath), "cache") } return storage.NewDisk(dir) case "s3": return storage.NewUnversionedS3(ctx, cfg.Cache.S3Bucket, cfg.Cache.S3Prefix) default: return nil, fmt.Errorf("storage: unknown backend %q", cfg.Cache.Backend) }}
// New creates a new Spindle server with the provided configuration and engines.func New(ctx context.Context, cfg *config.Config, d *db.DB, engines map[string]models.Engine, qm *quota.Manager) (*Spindle, error) { logger := log.FromContext(ctx) metrics := observability.GetMetrics(ctx) if metrics == nil { metrics = observability.NewMetrics() ctx = observability.WithMetrics(ctx, metrics) } n := notifier.New() lifecycleCtx, cancelLifecycle := context.WithCancel(context.WithoutCancel(ctx)) keepLifecycle := false defer func() { if !keepLifecycle { cancelLifecycle() } }()
if err := cleanupOrphanRepos(ctx, d, logger); err != nil { return nil, fmt.Errorf("failed to run startup cleanup: %w", err) }
cacheStore, err := newCacheStore(ctx, cfg) if err != nil { return nil, fmt.Errorf("failed to setup cache storage: %w", err) } if cacheStore != nil { logger.Info("cache storage enabled", "backend", cfg.Cache.Backend, "storeID", cfg.Cache.StoreID) }
var issuer *oidc.Issuer if cfg.Role != config.RoleExecutor && cfg.OIDC.KeyPath != "" { issuer, err = oidc.New(oidc.Options{ Issuer: cfg.OIDC.Issuer, KeyPath: cfg.OIDC.KeyPath, KeyID: cfg.OIDC.KeyID, }) if err != nil { return nil, fmt.Errorf("failed to setup OIDC issuer: %w", err) } logger.Info("minting CI ID tokens", "issuer", cfg.OIDC.Issuer, "kid", issuer.KeyID()) }
var id http.HandlerFunc if cfg.Role != config.RoleExecutor && cfg.Server.KeyPath != "" { id, err = identity.Load(cfg.Server.Hostname, cfg.Server.KeyPath) if err != nil { return nil, fmt.Errorf("failed to load spindle identity: %w", err) } logger.Info("serving did document", "did", cfg.Server.Did()) }
spindle := &Spindle{ db: d, l: logger, n: &n, wh: webhook.New(d, cfg.Server.Dev, pauseGate(d, &cfg.Server)), metrics: metrics, engs: engines, cfg: cfg, cache: cacheStore, motd: defaultMotd, rootCtx: lifecycleCtx, rootCancel: cancelLifecycle, jobWake: make(chan struct{}, 1), qm: qm, oidc: issuer, identity: id, } metrics.AttachDB(ctx, d) diskFallback := "" if cfg.Role == config.RoleStandalone { diskFallback = cfg.Server.LogDir if cfg.ArtifactStores.Disk.Dir == "" { logger.Warn("using SPINDLE_SERVER_LOG_DIR as the implicit disk artifact store; configure SPINDLE_ARTIFACT_STORES_DISK_DIR explicitly") } } stores, err := artifactstore.NewStores(cfg.ArtifactStores, diskFallback, cfg.LegacyS3.LogBucket) if err != nil { return nil, fmt.Errorf("failed to setup artifact stores: %w", err) } spindle.stores = stores if cfg.LegacyS3.LogBucket != "" { logger.Warn("SPINDLE_S3_LOG_BUCKET is deprecated; use SPINDLE_ARTIFACT_STORES_S3_BUCKET") } if cfg.Role != config.RoleExecutor { engine.StartCachePruner(ctx, logger, d, cacheStore, cfg.Cache.Retention, cfg.Cache.PruneInterval, cfg.Cache.MaxBytes) } if cfg.Role == config.RoleStandalone { spindle.reader = stores } else { name := cfg.Mill.ArtifactStore if name == "" { names := stores.Names() if len(names) != 1 { return nil, fmt.Errorf("%s requires SPINDLE_MILL_ARTIFACT_STORE when %d artifact stores are configured", cfg.Role, len(names)) } name = names[0] logger.Warn("SPINDLE_MILL_ARTIFACT_STORE is not set; inferred the only configured store", "store", name) } store, ok := stores.Store(name) if !ok { return nil, fmt.Errorf("SPINDLE_MILL_ARTIFACT_STORE=%q is not configured", name) } spindle.store = store spindle.reader = store } if cfg.Role == config.RoleExecutor { keepLifecycle = true return spindle, nil }
e, err := rbac.NewEnforcer(cfg.Server.DBPath) if err != nil { return nil, fmt.Errorf("failed to setup rbac enforcer: %w", err) } e.E.EnableAutoSave(true) spindle.e = e
switch cfg.Server.Secrets.Provider { case "openbao": if cfg.Server.Secrets.OpenBao.ProxyAddr == "" { return nil, fmt.Errorf("openbao proxy address is required when using openbao secrets provider") } spindle.vault, err = secrets.NewOpenBaoManager( cfg.Server.Secrets.OpenBao.ProxyAddr, logger, secrets.WithMountPath(cfg.Server.Secrets.OpenBao.Mount), ) if err != nil { return nil, fmt.Errorf("failed to setup openbao secrets provider: %w", err) } logger.Info("using openbao secrets provider", "proxy_address", cfg.Server.Secrets.OpenBao.ProxyAddr, "mount", cfg.Server.Secrets.OpenBao.Mount) case "sqlite", "": spindle.vault, err = secrets.NewSQLiteManager(cfg.Server.DBPath, secrets.WithTableName("secrets")) if err != nil { return nil, fmt.Errorf("failed to setup sqlite secrets provider: %w", err) } logger.Info("using sqlite secrets provider", "path", cfg.Server.DBPath) default: return nil, fmt.Errorf("unknown secrets provider: %s", cfg.Server.Secrets.Provider) }
collections := []string{ org_tangled.RepoDeclarationNSID, org_tangled.RepoManifestNSID, org_tangled.GraphCollaboratorNSID, org_tangled.TrackTicketNSID, org_tangled.ReviewPatchNSID, } // the trailing false is logDids, on would log a line for every did added jc, err := jetstream.NewJetstreamClient(cfg.Server.JetstreamEndpoint, collections, nil, log.SubLogger(logger, "jetstream"), d, true, false) if err != nil { return nil, fmt.Errorf("failed to setup jetstream client: %w", err) } spindle.jc = jc jc.AddDid(cfg.Server.Owner) for _, admin := range cfg.Server.Admins { jc.AddDid(admin) } // repo dids aren't known up front, so these skip the did filter jc.ExemptCollection(org_tangled.RepoManifestNSID) jc.ExemptCollection(org_tangled.GraphCollaboratorNSID) jc.ExemptCollection(org_tangled.TrackTicketNSID) jc.ExemptCollection(org_tangled.ReviewPatchNSID)
allowlisted, err := db.ListAllowlist(d) if err != nil { return nil, fmt.Errorf("failed to load the allow-list: %w", err) } for _, did := range allowlisted { jc.AddDid(did.String()) }
knownRepos, err := d.AllRepos() if err != nil { return nil, fmt.Errorf("failed to get known repos: %w", err) } for _, r := range knownRepos { if r.Owner != "" { jc.AddDid(r.Owner.String()) } }
spindle.res = idresolver.DefaultResolver(cfg.Server.PlcUrl) guarded := knotclient.HTTPClient(cfg.Server.Dev) spindle.knots = knotclient.New(spindle.res.Directory(), guarded, cfg.Server.Dev) spindle.scheduler = newPipelineScheduler(spindle)
err = e.AddSpindle(rbacDomain) if err != nil { return nil, fmt.Errorf("failed to set rbac domain: %w", err) } err = spindle.configureOwner() if err != nil { return nil, err } logger.Info("owner set", "did", cfg.Server.Owner)
spindle.backfill = newBackfill(spindle, log.SubLogger(logger, "backfill"), cfg.Server.Hydrant.Dir) jc.SetGapHandler(func(context.Context) { spindle.backfill.CatchUpLater() }) err = jc.StartJetstream(lifecycleCtx, spindle.ingest()) if err != nil { return nil, fmt.Errorf("failed to start jetstream consumer: %w", err) }
keepLifecycle = true return spindle, nil}func (s *Spindle) DB() *db.DB { return s.db}
func (s *Spindle) Engines() map[string]models.Engine { return s.engs}
func (s *Spindle) Vault() secrets.Manager { return s.vault}
func (s *Spindle) Notifier() *notifier.Notifier { return s.n}
func (s *Spindle) Enforcer() *rbac.Enforcer { return s.e}
func (s *Spindle) SetMotdContent(content []byte) { s.motdMu.Lock() defer s.motdMu.Unlock() s.motd = content}
func (s *Spindle) GetMotdContent() []byte { s.motdMu.RLock() defer s.motdMu.RUnlock() return s.motd}
// runs the server. blocksfunc (s *Spindle) Start(ctx context.Context) error { runCtx := s.rootCtx cancelRun := s.rootCancel if runCtx == nil || cancelRun == nil { runCtx, cancelRun = context.WithCancel(context.WithoutCancel(ctx)) s.rootCtx = runCtx s.rootCancel = cancelRun } defer cancelRun()
_, err := observability.StartMetricsServer(runCtx, s.cfg.Server.MetricsListenAddr, s.l, s.metrics.Registry()) if err != nil { return fmt.Errorf("starting metrics listener: %w", err) } if s.scheduler != nil { s.scheduler.Start(runCtx) }
// an executor dials out to its mill and takes work from it var execDone chan struct{} if s.exec != nil { execDone = make(chan struct{}) go func() { defer close(execDone) s.exec.Connect(runCtx) }() } waitForExecutor := func(waitCtx context.Context) error { if execDone == nil { return nil } select { case <-execDone: return nil case <-waitCtx.Done(): return waitCtx.Err() } }
if s.mill != nil && s.cfg.Mill.JumpListenAddr != "" { go s.mill.ServeJump(runCtx, s.cfg.Mill.JumpListenAddr, s.cfg.Mill.JumpHostKeyPath, s.cfg.Mill.DebugExecutorPort, s.cfg.Mill.MaxJumpConnections) }
if stopper, ok := s.vault.(secrets.Stopper); ok { defer stopper.Stop() }
if s.cfg.Role != config.RoleExecutor { s.resumeWipes(ctx)
failed, err := s.db.FailOrphanedWorkflows(s.n) if err != nil { return fmt.Errorf("failing orphaned workflows: %w", err) } if failed > 0 { s.l.Warn("failed workflows a restart left behind", "count", failed) }
owners, err := s.backfillOwners() if err != nil { return err } s.backfill.Track(owners...) if err := s.backfill.Resume(); err != nil { return err } var repoSync sync.WaitGroup repoSync.Add(2) go func() { defer repoSync.Done() s.retryFailedRepos(runCtx) }() go func() { defer repoSync.Done() s.backfill.run(runCtx) }() // hydrant has to be shut down before the process exits, or its db is // left for the next start to recover defer func() { cancelRun() repoSync.Wait() }() }
// start workers after resuming wipes so banned jobs cannot run workersCtx, cancelWorkers := context.WithCancel(runCtx) s.StartJobWorkers(workersCtx) defer func() { cancelWorkers() done := make(chan struct{}) go func() { s.jobWorkers.Wait() close(done) }() select { case <-done: case <-time.After(5 * time.Second): s.l.Warn("timed out waiting for job workers to stop") } }()
server := newHTTPServer(s.cfg.Server.ListenAddr, s.Router()) serverErr := make(chan error, 1) go func() { s.l.Info("starting spindle server", "address", s.cfg.Server.ListenAddr) serverErr <- server.ListenAndServe() }()
var listenErr error select { case listenErr = <-serverErr: case <-ctx.Done(): }
if s.exec != nil { drainCtx, cancelDrain := context.WithTimeout(context.Background(), s.cfg.Mill.DrainTimeout) if err := s.exec.Drain(drainCtx); err != nil { s.l.Warn("executor drain did not complete", "err", err) } cancelDrain() }
if s.cfg.Role != config.RoleExecutor { webhookCtx, cancelWebhooks := context.WithTimeout(context.Background(), serverShutdownTimeout) s.wh.Close(webhookCtx) cancelWebhooks() }
cancelRun() waitCtx, cancelWait := context.WithTimeout(context.Background(), executorShutdownTimeout) if err := waitForExecutor(waitCtx); err != nil { s.l.Warn("executor shutdown did not complete", "err", err) } cancelWait()
// the final cursor save has to land before the db closes if s.jc != nil { select { case <-s.jc.Done(): case <-time.After(serverShutdownTimeout): s.l.Warn("jetstream did not save its cursor before shutdown") } }
shutdownCtx, cancelShutdown := context.WithTimeout(context.Background(), serverShutdownTimeout) shutdownErr := server.Shutdown(shutdownCtx) cancelShutdown() // give notify work a moment before the run ends, since it can outlast the handler notifyDone := make(chan struct{}) go func() { s.notifyWork.Wait() close(notifyDone) }() select { case <-notifyDone: case <-time.After(5 * time.Second): s.l.Warn("timed out waiting for notify work to stop") } if listenErr == nil { listenErr = <-serverErr } if errors.Is(listenErr, http.ErrServerClosed) { listenErr = nil } if shutdownErr != nil { shutdownErr = fmt.Errorf("shutting down spindle server: %w", shutdownErr) } return errors.Join(listenErr, shutdownErr)}
// the first round erases hydrant's copy of anyone missing, so a load error fails the whole listfunc (s *Spindle) backfillOwners() ([]syntax.DID, error) { repos, err := s.db.AllRepos() if err != nil { return nil, fmt.Errorf("loading repo owners to backfill: %w", err) } allowed, err := db.ListAllowlist(s.db) if err != nil { return nil, fmt.Errorf("loading the allow-list to backfill: %w", err) } seen := make(map[syntax.DID]struct{}, len(repos)+len(allowed)+1) dids := make([]syntax.DID, 0, len(repos)+len(allowed)+1) add := func(did syntax.DID) { if did == "" { return } if _, ok := seen[did]; ok { return } seen[did] = struct{}{} dids = append(dids, did) }
for _, admin := range append([]string{s.cfg.Server.Owner}, s.cfg.Server.Admins...) { did, err := syntax.ParseDID(admin) if err != nil { s.l.Warn("backfill: invalid configured admin", "admin", admin, "err", err) continue } add(did) } for _, member := range allowed { add(member) } for _, repo := range repos { add(repo.Owner) } sort.Slice(dids, func(i, j int) bool { return dids[i] < dids[j] }) return dids, nil}
var errMicrovmUnsupported = errors.New("microvm engine is only supported on Linux")
func localEngines(ctx context.Context, cfg *config.Config, d *db.DB, qStore quota.ReservationStore, logger *slog.Logger) (map[string]models.Engine, error) { nixeryEng, err := nixery.New(ctx, cfg) if err != nil { return nil, err } engines := map[string]models.Engine{ "nixery": nixeryEng, "dummy": dummy.New(logger), } microvmEng, err := newMicrovmEngine(ctx, cfg, d, qStore) switch { case errors.Is(err, errMicrovmUnsupported): logger.Warn("microvm engine unavailable, workflows that use it will fail", "error", err) case err != nil: return nil, err default: engines["microvm"] = microvmEng } return engines, nil}
func Run(ctx context.Context) error { cfg, err := config.Load(ctx) if err != nil { return fmt.Errorf("failed to load config: %w", err) }
shutdownTracing, err := observability.InitTracing(ctx, cfg.Tracing) if err != nil { return fmt.Errorf("failed to initialize tracing: %w", err) } defer func() { shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := shutdownTracing(shutdownCtx); err != nil { log.FromContext(ctx).Error("failed to shut down tracing", "err", err) } }()
otelHandler, shutdownLogging, err := observability.InitLogging(ctx, cfg.Logging) if err != nil { return fmt.Errorf("failed to initialize logging: %w", err) } defer func() { shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() if err := shutdownLogging(shutdownCtx); err != nil { log.FromContext(ctx).Error("failed to shut down logging", "err", err) } }()
currentLogger := log.FromContext(ctx) combinedHandler := log.NewFanoutHandler(currentLogger.Handler(), otelHandler) combinedLogger := slog.New(combinedHandler) slog.SetDefault(combinedLogger) ctx = log.IntoContext(ctx, combinedLogger)
metrics := observability.NewMetrics() ctx = observability.WithMetrics(ctx, metrics)
if err := ensureGitVersion(); err != nil { return fmt.Errorf("ensuring git version: %w", err) }
releaseInstance, err := lockInstance(cfg.Server.DBPath) if err != nil { return err } defer releaseInstance()
d, err := db.Make(ctx, cfg.Server.DBPath) if err != nil { return fmt.Errorf("failed to setup db: %w", err) } if err := d.MigratePipelineLogFiles(cfg.Server.LogDir); err != nil { return fmt.Errorf("failed to migrate pipeline log files: %w", err) }
var qs *db.QuotaStore var qm *quota.Manager if cfg.Role != config.RoleExecutor { defaults, err := cfg.ToDefaults() if err != nil { return fmt.Errorf("invalid quota configuration: %w", err) } for scope, resources := range defaults { for resource, limit := range resources { metrics.SetQuotaDefaultLimit(string(scope), string(resource), limit) } } qs = db.NewQuotaStore(d, defaults) qm = quota.NewManager(qs, 50*time.Millisecond, metrics.QuotaObserver()) defer qm.Close()
if cfg.Role != config.RoleMill { if err := qs.Recover(ctx, nil); err != nil { return fmt.Errorf("failed to recover quota reservations: %w", err) } } }
logger := log.FromContext(ctx)
var engines map[string]models.Engine var m *mill.Mill var quotaClient *executor.QuotaClient
if cfg.Role == config.RoleMill { // on a mill host, engines place jobs on executors instead of running // them. all names share one Mill m = mill.New(log.SubLogger(logger, "mill"), mill.Config{ LogDir: cfg.Server.LogDir, MaxPending: cfg.Mill.MaxPending, ReconnectGrace: cfg.Mill.ReconnectGrace, CancelAckTimeout: cfg.Mill.CancelAckTimeout, CancelTeardownTimeout: cfg.Mill.CancelTeardownTimeout, CacheStoreID: cfg.Cache.StoreID, CacheMaxBytesPerOwner: cfg.Cache.MaxBytesPerOwner, CacheMaxBytes: cfg.Cache.MaxBytes, }) engines = map[string]models.Engine{ "nixery": mill.NewEngine("nixery", m), "microvm": mill.NewEngine("microvm", m), "dummy": mill.NewEngine("dummy", m), } } else { // standalone and executor both run real engines locally var engineQStore quota.ReservationStore if cfg.Role == config.RoleExecutor { quotaClient = executor.NewQuotaClient() } else { engineQStore = qs } engines, err = localEngines(ctx, cfg, d, engineQStore, logger) if err != nil { return err } }
s, err := New(ctx, cfg, d, engines, qm) if err != nil { return err } // an executor's workflows carry a check that asks the mill instead if cfg.Role != config.RoleExecutor { attachDebugAuthorizer(engines, s) }
if cfg.Role == config.RoleMill || cfg.Role == config.RoleStandalone { // flattens the store snapshot into bounded label tuples, the collector // drops anything outside the bounded scope, resource and status sets metrics.SetQuotaLoader(func() (observability.QuotaSnapshot, error) { snap, err := qs.MetricsSnapshot(context.Background()) if err != nil { return observability.QuotaSnapshot{}, err } var out observability.QuotaSnapshot for scope, resources := range snap.Usage { for resource, used := range resources { out.Usage = append(out.Usage, observability.QuotaUsage{ Scope: string(scope), Resource: resource, Used: used, }) } } for scope, resources := range snap.Subjects { for resource, statuses := range resources { for status, count := range statuses { out.Subjects = append(out.Subjects, observability.QuotaSubjectCount{ Scope: string(scope), Resource: string(resource), Status: status, Count: count, }) } } } return out, nil }) } if m != nil { // the engines built above hold the mill, but the mill's db and // notifier only exist after New, so attach them here m.Attach(s.DB(), s.Notifier(), qm) m.AttachCache(s.cache) s.mill = m s.mill.SetDebugKeyChecker(s) s.mill.RegisterMetrics(s.metrics) if err := m.RestoreState(); err != nil { return fmt.Errorf("restoring mill state: %w", err) } liveIDs := m.LiveQuotaReservationIDs() for i := range liveIDs { liveIDs[i] = quota.StorageReservationID(liveIDs[i]) } if err := qs.Recover(ctx, liveIDs); err != nil { return fmt.Errorf("failed to recover mill quota reservations: %w", err) } } if cfg.Role == config.RoleExecutor { s.exec, err = executor.New(cfg, engines, s.DB(), s.Notifier(), log.SubLogger(logger, "executor"), s.store, s.cache) if err != nil { return err } s.exec.SetQuotaClient(quotaClient) s.exec.RegisterMetrics(s.metrics) }
return s.Start(ctx)}
// no write timeout, log streams run for hours and would get cut offfunc newHTTPServer(addr string, handler http.Handler) *http.Server { return &http.Server{ Addr: addr, Handler: handler, ReadHeaderTimeout: 10 * time.Second, ReadTimeout: 30 * time.Second, IdleTimeout: 2 * time.Minute, }}
func (s *Spindle) Router() http.Handler { mux := chi.NewRouter() mux.Use(observability.HTTPMiddleware(s.metrics)) if s.cfg.Tracing.Endpoint != "" { mux.Use(observability.OTelRouteMiddleware) }
mux.HandleFunc("/", func(w http.ResponseWriter, r *http.Request) { w.Write(s.GetMotdContent()) }) mux.Get("/_health", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "application/json") _, _ = io.WriteString(w, "{\"status\":\"ok\"}\n") }) if s.cfg.Role != config.RoleExecutor { if s.mill != nil { mux.HandleFunc("/mill", s.mill.HandleExecutorConn) }
mux.Mount("/xrpc", s.XrpcRouter()) }
if s.identity != nil { mux.Get(identity.DocumentPath, s.identity) }
if s.oidc != nil { mux.Get(oidc.ConfigurationPath, s.oidc.ServeConfiguration) mux.Get(oidc.JWKSPath, s.oidc.ServeJWKS) }
if s.cfg.Tracing.Endpoint == "" { return mux } return otelhttp.NewHandler( mux, "HTTP", otelhttp.WithSpanNameFormatter(func(string, *http.Request) string { return "HTTP" }), )}
func (s *Spindle) XrpcRouter() http.Handler { serviceAuth := serviceauth.NewServiceAuth(s.l, s.res.Directory(), s.cfg.Server.Did().String())
l := log.SubLogger(s.l, "xrpc")
x := xrpc.Xrpc{ Logger: l, Db: s.db, Enforcer: s.e, Engines: s.engs, Config: s.cfg, ArtifactReader: s.reader, Resolver: s.res, Vault: s.vault, Notifier: s.Notifier(), Webhooks: s.wh, ServiceAuth: serviceAuth, Trigger: s, QuotaStore: db.NewQuotaStore(s.db, quota.Defaults{}), Wiper: s, Backfill: s.backfill, Knots: s.knots, Events: s, Repos: s, }
return x.Router()}
func (s *Spindle) buildTriggerRepo(ctx context.Context, repo *db.Repo) (*workflow.TriggerRepo, error) { rkey := string(repo.Rkey) repoDid := repo.RepoDid.String() return s.buildTriggerRepoFrom(ctx, repo.Knot, repo.Owner.String(), rkey, repoDid), nil}
func (s *Spindle) buildTriggerRepoFrom(ctx context.Context, knot, did, rkey, repoDid string) *workflow.TriggerRepo { // an unreadable manifest only costs the trigger its default branch defaultBranch, _ := s.defaultBranch(ctx, knotclient.Knot{Host: knot}, syntax.DID(repoDid))
var rkeyPtr *string if rkey != "" { rkeyPtr = &rkey } return &workflow.TriggerRepo{ Did: did, Knot: knot, Repo: rkeyPtr, RepoDid: &repoDid, DefaultBranch: defaultBranch, }}
func (s *Spindle) resolvePipelineSourceRepo(ctx context.Context, trigger *workflow.TriggerMetadata) (*workflow.TriggerRepo, error) { if trigger == nil { return nil, nil } if trigger.SourceRepo == nil || *trigger.SourceRepo == "" { return trigger.Repo, nil } repoDid, err := syntax.ParseDID(*trigger.SourceRepo) if err != nil { return nil, fmt.Errorf("parse sourceRepo %s: %w", *trigger.SourceRepo, err) } return s.resolveSourceRepoInfo(ctx, repoDid)}
func (s *Spindle) resolveSourceRepoInfo(ctx context.Context, repoDid syntax.DID) (*workflow.TriggerRepo, error) { repo, err := s.db.GetRepoByDid(repoDid) if err == nil { return s.buildTriggerRepo(ctx, repo) }
// the knot comes from the did doc, not the caller, or the source could point anywhere k, err := s.knots.KnotOf(ctx, repoDid) if err != nil { return nil, fmt.Errorf("verify sourceRepo %s: %w", repoDid, err) } var manifest org_tangled.RepoManifest if err := s.knots.GetRecord(ctx, k, repoDid, org_tangled.RepoManifestNSID, knotclient.ManifestRkey, &manifest); err != nil { return nil, fmt.Errorf("verify sourceRepo %s: reading manifest: %w", repoDid, err) } owner, _, rkey, err := manifestIdentity(&manifest) if err != nil { return nil, fmt.Errorf("verify sourceRepo %s: %w", repoDid, err) } declared, err := s.pdsDeclaredRepo(ctx, owner, rkey.String()) if err != nil { return nil, fmt.Errorf("verify sourceRepo %s: reading declaration %s/%s: %w", repoDid, owner, rkey, err) } if declared != repoDid { return nil, fmt.Errorf("verify sourceRepo %s: %s/%s declares %s", repoDid, owner, rkey, declared) } return s.buildTriggerRepoFrom(ctx, k.Host, owner.String(), rkey.String(), repoDid.String()), nil}
func (s *Spindle) runPipeline(ctx context.Context, pipelineId models.PipelineId, repoDid syntax.DID, trigger workflow.TriggerMetadata, changedFiles []string, repoCloneUri, repoPath, rev string, only []string, sourceRepo *workflow.TriggerRepo, blocked bool) (models.PipelineId, error) { l := log.FromContext(ctx)
// all triggers go through this gate, not just pushes if gate, err := s.db.Gate(&s.cfg.Server, repoDid); err != nil { return "", fmt.Errorf("checking admission: %w", err) } else if gate.Paused { l.Info("owner isn't admitted, starting nothing for their repo", "repo", repoDid) return "", nil }
compiler := workflow.Compiler{ ChangedFiles: changedFiles, Trigger: trigger, }
rawPipeline, _, err := s.loadPipeline(ctx, repoCloneUri, repoPath, rev) if err != nil { return "", fmt.Errorf("loading pipeline: %w", err) } if len(rawPipeline) == 0 { return "", nil }
tpl := compiler.Compile(compiler.Parse(rawPipeline)) // todo(dawn): pass compile error to workflow log for _, w := range compiler.Diagnostics.Errors { l.Error(w.String()) } for _, w := range compiler.Diagnostics.Warnings { l.Warn(w.String()) }
if len(only) > 0 { tpl.Workflows = filterWorkflows(tpl.Workflows, only) } if len(tpl.Workflows) == 0 { return "", nil }
record, err := pipelinecodec.FromTangled(pipelineId, time.Now(), tpl) if err != nil { return "", fmt.Errorf("creating pipeline: %w", err) } err = s.processPipeline(ctx, repoDid, tpl, record, sourceRepo, blocked) // both errors mean nothing to enqueue, since a pause or wipe can land mid checkout if errors.Is(err, db.ErrOwnerPaused) || errors.Is(err, db.ErrNotServed) { l.Info("nothing enqueued, the owner is paused or the repo is gone", "repo", repoDid) return "", nil } if err != nil { return "", err } return pipelineId, nil}
func filterWorkflows(workflows []*workflow.WorkflowSpec, only []string) []*workflow.WorkflowSpec { allowed := make(map[string]struct{}, len(only)) for _, n := range only { allowed[n] = struct{}{} } var filtered []*workflow.WorkflowSpec for _, w := range workflows { if w == nil { continue } if _, ok := allowed[w.Name]; ok { filtered = append(filtered, w) } } return filtered}
// TriggerManual dispatches a pipeline at sha, authorized against and recorded// under repoDid. sourceRepo, pull, and inputs are optional trigger payload.func (s *Spindle) TriggerManual(ctx context.Context, actor, repoDid syntax.DID, sha, ref string, workflows []string, inputs []*workflow.TriggerPair) (models.PipelineId, error) { repo, err := s.db.GetRepoByDid(repoDid) if err != nil { return "", fmt.Errorf("unknown repoDid %s: %w", repoDid, err) }
if ban, err := s.db.IsBanned(repoDid, repo.Owner); err != nil { return "", fmt.Errorf("checking bans: %w", err) } else if ban != nil { return "", fmt.Errorf("repo %s is banned", repoDid) } // checked before the checkout, otherwise a paused owner matches nothing instead of erroring if gate, err := s.db.Gate(&s.cfg.Server, repoDid); err != nil { return "", fmt.Errorf("checking admission: %w", err) } else if gate.Paused { return "", xrpc.ErrOwnerPaused }
triggerRepo, err := s.buildTriggerRepo(ctx, repo) if err != nil { return "", fmt.Errorf("building trigger repo: %w", err) }
var refPtr *string if ref != "" { refPtr = &ref } trigger := workflow.TriggerMetadata{ Kind: string(workflow.TriggerKindManual), Repo: triggerRepo, Actor: (*string)(&actor), Manual: &workflow.ManualTriggerData{ Sha: sha, Ref: refPtr, Inputs: inputs, }, }
repoCloneUri, repoPath := s.resolveCheckout(repo)
// a manual ref only becomes a signable claim once the server has seen the // sha inside the ref's history if refPtr != nil { contains, err := gitutil.RefContainsCommit(ctx, repoCloneUri, ref, sha) if err != nil { return "", fmt.Errorf("checking ref %s: %w", ref, err) } if !contains { return "", fmt.Errorf("ref %s does not contain commit %s", ref, sha) } }
pipelineId, err := s.runPipeline(ctx, models.PipelineId(tid.TID()), repoDid, trigger, nil, repoCloneUri, repoPath, sha, workflows, triggerRepo, false) if err != nil { return "", err } if pipelineId == "" { return "", xrpc.ErrNoMatchingWorkflows } return pipelineId, nil}
func (s *Spindle) resolveCheckout(repo *db.Repo) (cloneUri, repoPath string) { return s.knots.CloneURL(knotclient.Knot{Host: repo.Knot}, repo.RepoDid), s.newRepoPath(repo.RepoDid)}
// resolves the workflow definition at sha without executing it// returns a deterministic fingerprint over the resolved files.func (s *Spindle) ListWorkflowDefinitions(ctx context.Context, repoDid syntax.DID, ref string) ([]*models.WorkflowDefinition, error) { if ref == "" { ref = "HEAD" } return s.workflowDefs.list(repoDid, ref, func() ([]*models.WorkflowDefinition, error) { return s.loadWorkflowDefinitions(ctx, repoDid, ref) })}
func (s *Spindle) loadWorkflowDefinitions(ctx context.Context, repoDid syntax.DID, ref string) ([]*models.WorkflowDefinition, error) { rawPipeline, commit, err := s.readWorkflowDir(ctx, repoDid, ref) if err != nil { return nil, fmt.Errorf("loading pipeline: %w", err) } out := make([]*models.WorkflowDefinition, 0, len(rawPipeline)) for _, raw := range rawPipeline { definition, err := pipelinecodec.FileDefinition(raw.Name, raw.Contents, repoDid.String(), commit) if err != nil { return nil, fmt.Errorf("parsing workflow %s: %w", raw.Name, err) } out = append(out, definition) } return out, nil}
// the host comes from a did doc the repo owner controls and git dials it unguarded, so// every knot fetch comes through here and git's stderr only goes to the logfunc (s *Spindle) syncRepo(ctx context.Context, cloneUri, repoPath, rev string) error { err := s.guardedSync(ctx, cloneUri, repoPath, rev) if err != nil { log.FromContext(ctx).Error("sync git repo", "uri", cloneUri, "err", err) if refusedByKnot(err) { return errRevRefused } return errors.New("couldn't fetch the push from the knot") } return nil}
var ( errRevRefused = errors.New("the knot refused to serve that commit") // git only passes on the status of a refused smart http request, not the knot's reason gitHTTPStatus = regexp.MustCompile(`returned error: (\d{3})`))
// a 4xx means the knot won't serve that commit, a retry gets the same answer unless it was busyfunc refusedByKnot(err error) bool { msg := err.Error() // upload-pack answers 200 for an unknown sha, which leaves only the message to match if strings.Contains(msg, "not our ref") || strings.Contains(msg, "unadvertised object") { return true } // info/refs 404s only show up in the message text if strings.Contains(msg, "fatal: repository '") && strings.Contains(msg, "' not found") { return true } m := gitHTTPStatus.FindStringSubmatch(msg) if m == nil { return false } status, _ := strconv.Atoi(m[1]) return status >= 400 && status < 500 && status != http.StatusRequestTimeout && status != http.StatusTooManyRequests}
func (s *Spindle) guardedSync(ctx context.Context, cloneUri, repoPath, rev string) error { // a plain path has no host to dial if u, err := url.Parse(cloneUri); err == nil && u.Host != "" && !s.cfg.Server.Dev { if err := netguard.CheckHost(ctx, u.Host); err != nil { return err } } return gitutil.SparseSync(ctx, cloneUri, repoPath, rev, sparseWorkflowDir)}
func (s *Spindle) loadPipeline(ctx context.Context, repoUri, repoPath, rev string) (workflow.RawPipeline, string, error) { if err := s.syncRepo(ctx, repoUri, repoPath, rev); err != nil { return nil, "", err } // go-git chokes on sha256 repos, so git resolves the commit head, err := gitutil.Output(ctx, "-C", repoPath, "rev-parse", "HEAD") if err != nil { return nil, "", fmt.Errorf("resolving checked out commit: %w", err) } commit := strings.TrimSpace(string(head)) rawPipeline, err := readWorkflowTree(ctx, gitutil.Output, repoPath, commit) return rawPipeline, commit, err}
func (s *Spindle) newRepoPath(repo syntax.DID) string { return filepath.Join(s.cfg.Server.RepoDir, repo.String())}
const RequiredVersion = "2.49.0"
func ensureGitVersion() error { v, err := gitutil.Version() if err != nil { return fmt.Errorf("fetching git version: %w", err) } if v.LessThan(version.Must(version.NewVersion(RequiredVersion))) { return fmt.Errorf("installed git version %q is not supported, Spindle requires git version >= %q", v, RequiredVersion) } return nil}
func (s *Spindle) configureOwner() error { cfgOwner := s.cfg.Server.Owner
existing, err := s.e.GetSpindleUsersByRole("server:owner", rbacDomain) if err != nil { return err }
switch len(existing) { case 0: // no owner configured, continue case 1: // find existing owner existingOwner := existing[0]
// no ownership change, this is okay if existingOwner == s.cfg.Server.Owner { break }
// remove existing owner err = s.e.RemoveSpindleOwner(rbacDomain, existingOwner) if err != nil { return nil } default: return fmt.Errorf("more than one owner in DB, try deleting %q and starting over", s.cfg.Server.DBPath) }
return s.e.AddSpindleOwner(rbacDomain, cfgOwner)}
func (s *Spindle) StartJobWorkers(ctx context.Context) { for range s.cfg.Server.MaxJobCount { s.jobWorkers.Add(1) go func() { defer s.jobWorkers.Done() for { job, err := s.dequeueJob(ctx) if err != nil && ctx.Err() == nil { s.l.Error("failed to dequeue job", "error", err) } if job == nil { // polls because a failed dequeue can leave jobs queued with nobody to wake us var retry <-chan time.Time if err != nil { retry = time.After(dequeueRetryDelay) } select { case <-ctx.Done(): return case <-s.jobWake: case <-retry: } continue } s.metrics.RecordJobQueueActivity("dequeue") s.runJob(ctx, job) s.finishJob(job.Owner) } }() }}
func (s *Spindle) dequeueJob(ctx context.Context) (*db.JobRow, error) { s.jobsMu.Lock() defer s.jobsMu.Unlock() job, err := s.db.DequeueJob(ctx, s.jobsByOwner) if job != nil { if s.jobsByOwner == nil { s.jobsByOwner = make(map[string]int) } s.jobsByOwner[job.Owner]++ } return job, err}
func (s *Spindle) finishJob(owner string) { s.jobsMu.Lock() defer s.jobsMu.Unlock() s.jobsByOwner[owner]-- if s.jobsByOwner[owner] <= 0 { delete(s.jobsByOwner, owner) }}
func (s *Spindle) runJob(ctx context.Context, job *db.JobRow) { pipelineId := job.PipelineId repoDID := job.RepoDid if job.SourceRepo != nil && job.SourceRepo.RepoDid != nil && *job.SourceRepo.RepoDid != "" { repoDID = *job.SourceRepo.RepoDid }
jobCtx := observability.ExtractFromTraceparentAndTracestate(ctx, job.Traceparent, job.Tracestate) if job.CreatedAtNs > 0 { queueDelay := time.Since(time.Unix(0, job.CreatedAtNs)) if queueDelay < 0 { queueDelay = 0 } s.metrics.RecordJobDequeueLatency(jobCtx, queueDelay) }
jobCtx, span := observability.Tracer().Start(jobCtx, "job.run") if span.IsRecording() { attrs := []attribute.KeyValue{ attribute.Int64(observability.JobIDKey, job.Id), attribute.String(observability.PipelineIDKey, pipelineId.String()), } if repoDID != "" { attrs = append(attrs, attribute.String(observability.RepoDIDKey, repoDID)) } if job.RepoDid != "" && job.RepoDid != repoDID { attrs = append(attrs, attribute.String(observability.TargetRepoDIDKey, job.RepoDid)) } span.SetAttributes(attrs...) } failed := false defer func() { if failed { span.SetStatus(codes.Error, "job failed") } else { span.SetStatus(codes.Ok, "success") } span.End() }()
l := log.SubLogger(log.SubLogger(s.l, "job"), "engine").With( "job_id", job.Id, ) if job.RepoDid != "" && job.RepoDid != repoDID { l = l.With("target_repo_did", job.RepoDid) }
pipelineEnv := models.PipelineEnvVarsForSource(job.Tpl.TriggerMetadata, pipelineId, job.SourceRepo)
initTpl := job.Tpl if job.SourceRepo != nil && job.Tpl.TriggerMetadata != nil { tm := *job.Tpl.TriggerMetadata tm.Repo = job.SourceRepo initTpl.TriggerMetadata = &tm } trustedSource := models.TrustedPipelineSource(initTpl.TriggerMetadata, job.RepoDid)
workflows := make(map[models.Engine][]models.Workflow) for _, w := range job.Tpl.Workflows { if w == nil { continue } eng, ok := s.engs[w.Engine] if !ok { failed = true _ = s.db.StatusFailed(models.WorkflowId{ PipelineId: pipelineId, Name: w.Name, }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n) s.metrics.RecordWorkflowTerminal( fmt.Sprint(w.Engine), "failure", string(engine.FailureClassUser), string(engine.FailureReasonWorkflowInvalid), ) continue }
ewf, err := eng.InitWorkflow(*w, initTpl) if err != nil { failed = true _ = s.db.StatusFailed(models.WorkflowId{ PipelineId: pipelineId, Name: w.Name, }, fmt.Sprintf("init workflow: %s", err), -1, s.n) failureClass, failureReason := engine.FailureAttribution("failure", err) s.metrics.RecordWorkflowTerminal( fmt.Sprint(w.Engine), "failure", failureClass, failureReason, ) continue } ewf.RunID = fmt.Sprintf("%d", job.Id)
if ewf.Environment == nil { ewf.Environment = make(map[string]string) } maps.Copy(ewf.Environment, pipelineEnv)
ewf.Engine = w.Engine ewf.RepoDID = job.RepoDid if w.Audience != nil { ewf.Audience = *w.Audience } if !trustedSource { ewf.Caches = nil } workflows[eng] = append(workflows[eng], *ewf) }
engine.StartWorkflows(l, s.vault, s.cfg, s.qm, s.stores, s.db, s.n, s.cache, nil, jobCtx, &models.Pipeline{ RepoDid: syntax.DID(job.RepoDid), Workflows: workflows, TrustedSource: trustedSource, TriggerMetadata: job.Tpl.TriggerMetadata, MintIDToken: s.idTokenMinter(job, trustedSource), }, pipelineId)}
// only trusted source runs receive a token. unrecordable identities withhold// it; signing failures fail the workflow.func (s *Spindle) idTokenMinter(job *db.JobRow, trustedSource bool) func(time.Duration, string) (string, error) { if s.oidc == nil || !trustedSource || job.Tpl.TriggerMetadata == nil { return nil }
identity := models.RunIdentityFor(job.Tpl.TriggerMetadata, job.RepoDid) if identity.SHA == "" { s.l.Warn("id tokens are off for a run with no recorded commit", "job_id", job.Id) return nil }
run := oidc.Run{ Repository: identity.Repository, Ref: identity.Ref, SHA: identity.SHA, Actor: identity.Actor, } return func(workflowTimeBudget time.Duration, audience string) (string, error) { token, err := s.oidc.Mint(run, workflowTimeBudget, audience) if errors.Is(err, oidc.ErrUnrecordableRun) { s.l.Warn("withholding an id token the issuer will not sign", "job_id", job.Id, "err", err) return "", nil } return token, err }}
// reruns the spec a pipeline was enqueued with under a new id, on behalf of actor. a pipeline// that was blocked has to be approved first, even once cancelled. a retry would skip the approvalfunc (s *Spindle) RetryPipeline(ctx context.Context, actor, repoDid syntax.DID, id models.PipelineId, workflows []string) (models.PipelineId, error) { stored, err := s.db.GetStoredPipeline(id) if err != nil { return "", err } if stored.Spec == nil || stored.Spec.TriggerMetadata == nil || stored.NeededApproval && !stored.Approved { return "", xrpc.ErrNotRetryable } tpl := *stored.Spec trigger := *tpl.TriggerMetadata trigger.Actor = (*string)(&actor) if stored.NeededApproval && trigger.PullRequest != nil { pull := *trigger.PullRequest pull.SourceBranch = "" trigger.PullRequest = &pull } tpl.TriggerMetadata = &trigger if len(workflows) > 0 { tpl.Workflows = filterWorkflows(tpl.Workflows, workflows) } if len(tpl.Workflows) == 0 { return "", xrpc.ErrNoMatchingWorkflows }
retryId := models.PipelineId(tid.TID()) record, err := pipelinecodec.FromTangled(retryId, time.Now(), tpl) if err != nil { return "", fmt.Errorf("creating pipeline: %w", err) } if err := s.processPipeline(ctx, repoDid, tpl, record, stored.SourceRepo, false); err != nil { return "", err } return retryId, nil}
// the guards run again, an owner paused or banned since the pipeline was blocked stays blockedfunc (s *Spindle) ApprovePipeline(ctx context.Context, repoDid syntax.DID, id models.PipelineId) error { traceparent, tracestate := observability.InjectToTraceparentAndTracestate(ctx) err := func() error { defer s.lockRepo(repoDid).Unlock() return s.db.ApprovePipeline(s.rootCtx, repoDid.String(), id, traceparent, tracestate, s.n, db.Admitted(&s.cfg.Server)) }() if err != nil { return err } s.l.Info("pipeline approved", "id", id)
select { case s.jobWake <- struct{}{}: default: } return nil}
func (s *Spindle) processPipeline(ctx context.Context, repoDid syntax.DID, tpl workflow.PipelineSpec, record *models.PipelineRecord, sourceRepo *workflow.TriggerRepo, blocked bool) error { traceparent, tracestate := observability.InjectToTraceparentAndTracestate(ctx) // only the insert holds the repo lock, which keeps a wipe from landing halfway through err := func() error { defer s.lockRepo(repoDid).Unlock() return s.db.EnqueuePipeline(s.rootCtx, repoDid.String(), record, sourceRepo, tpl, blocked, traceparent, tracestate, s.n, db.Admitted(&s.cfg.Server)) }() if err != nil { return fmt.Errorf("failed to enqueue durable job: %w", err) } if blocked { s.l.Info("pipeline is blocked until approved", "id", record.ID) return nil } s.l.Info("pipeline enqueued successfully to db", "id", record.ID)
select { case s.jobWake <- struct{}{}: default: }
return nil}