package 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/nixcache"
"tangled.org/core/spindle/observability"
"tangled.org/core/spindle/oidc"
"tangled.org/core/spindle/pemkey"
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 motd
var defaultMotd []byte
const (
rbacDomain = "thisserver"
executorShutdownTimeout = 8 * time.Minute
serverShutdownTimeout = 10 * time.Second
dequeueRetryDelay = time.Second
// attic checks the token per request, and the last upload can land after the budget
nixCacheTokenGrace = 5 * time.Minute
// 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
cacheCtl engine.CacheController
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
nixCache *nixcache.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 nixIssuer *nixcache.Issuer
if cfg.Role != config.RoleExecutor && cfg.NixCache.AtticURL != "" {
key, err := pemkey.LoadRSA(cfg.NixCache.AtticKeyPath, "attic key")
if err != nil {
return nil, fmt.Errorf("setting up attic cache tokens: %w", err)
}
nixIssuer = &nixcache.Issuer{
Client: nixcache.Client{BaseURL: cfg.NixCache.AtticURL},
Key: key,
RunUploadBytes: cfg.NixCache.RunUploadMiB << 20,
}
logger.Info("minting attic cache tokens", "url", cfg.NixCache.AtticURL)
}
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,
nixCache: nixIssuer,
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. blocks
func (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 list
func (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
var publishing []db.NixPublishing
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 {
publishing, err = qs.Recover(ctx, nil)
if err != nil {
return fmt.Errorf("failed to recover quota reservations: %w", err)
}
}
}
logger := log.FromContext(ctx)
if cfg.SharesNixUploadWithOpenAdmission() {
logger.Warn("SPINDLE_NIX_CACHE_UPLOAD_URL is shared by every repository and admission is open; any workflow can poison it")
}
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,
CacheQuota: qs,
CacheMaxBytes: cfg.Cache.MaxBytes,
NixCaches: cfg.NixCache.AtticURL != "",
})
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 s.nixCache != nil {
s.nixCache.Limits = qs
s.nixCache.Grants = qs
}
if s.cache != nil && cfg.Role != config.RoleExecutor {
s.cacheCtl = engine.NewLocalCacheController(d, qs, s.cache, cfg.Server.RepoDir, cfg.Cache.MaxEntriesPerOwner, cfg.Cache.MaxBytes, logger)
}
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])
}
publishing, err = qs.Recover(ctx, liveIDs)
if err != nil {
return fmt.Errorf("failed to recover mill quota reservations: %w", err)
}
}
var cacheKinds []quota.CacheKind
if s.cacheCtl != nil {
cacheKinds = append(cacheKinds, engine.CacheEntries(d, s.cache))
}
if s.nixCache != nil {
dropped, bytes, err := qs.DropUnindexedNixCharges(ctx)
if err != nil {
return fmt.Errorf("dropping upload url nix cache charges: %w", err)
}
if dropped > 0 {
logger.Info("dropped nix cache charges left from the upload url", "count", dropped, "bytes", bytes)
}
pruner := &nixcache.Pruner{
Index: qs,
Attic: s.nixCache,
MaxBytes: cfg.NixCache.MaxMiB << 20,
Retention: cfg.NixCache.Retention,
Grace: nixCacheTokenGrace,
Grants: qs,
Publishing: publishing,
Reservations: qm,
Log: logger,
}
pruner.Start(ctx, cfg.Cache.PruneInterval)
cacheKinds = append(cacheKinds, pruner)
} else {
// without attic nothing else settles these uploads
for _, row := range publishing {
if err := qm.Release(ctx, row.ReservationID); err != nil {
return fmt.Errorf("releasing interrupted attic upload %s: %w", row.ReservationID, err)
}
}
}
if qs != nil {
evictor := "a.Evictor{Usage: qs, Limits: qs, Kinds: cacheKinds, Log: logger}
evictor.Start(ctx, cfg.Cache.PruneInterval)
}
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 off
func 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 log
func (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 busy
func 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, s.cacheCtl, jobCtx, &models.Pipeline{
RepoDid: syntax.DID(job.RepoDid),
Workflows: workflows,
TrustedSource: trustedSource,
TriggerMetadata: job.Tpl.TriggerMetadata,
MintIDToken: s.idTokenMinter(job, trustedSource),
GrantNixCache: s.nixCacheGranter(job, trustedSource),
}, pipelineId)
}
// only trusted source runs push to the repo's cache, and attic records the run as the uploader
func (s *Spindle) nixCacheGranter(job *db.JobRow, trustedSource bool) func(context.Context, models.WorkflowId, time.Duration) (*models.NixCacheGrant, error) {
if s.nixCache == nil {
return nil
}
return func(ctx context.Context, wid models.WorkflowId, workflowTimeBudget time.Duration) (*models.NixCacheGrant, error) {
sub := string(wid.PipelineId) + "/" + wid.Name
return s.nixCache.Grant(ctx, job.RepoDid, job.Owner, sub, trustedSource, time.Now().Add(workflowTimeBudget+nixCacheTokenGrace))
}
}
// 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 approval
func (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 blocked
func (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
}