alice.pds.demo.boltless.dev / core
This repository has no description
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

spindle: decouple execution from job ingestion

Signed-off-by: dawn <dawn@tangled.org>

author
dawn committer
Tangled date (Jul 21, 2026, 5:03 PM +0300) commit 17c82409 parent 0e430228 change-id lwurzyvr
+148 -69
+9
spindle/db/db.go
··· 112 112 repo_did text not null, 113 113 commit_id text not null 114 114 ); 115 + create table if not exists jobs ( 116 + id integer primary key autoincrement, 117 + repo_did text not null, 118 + pipeline_id_knot text not null, 119 + pipeline_id_rkey text not null, 120 + source_repo text, 121 + tpl text not null, 122 + created_at integer not null default (strftime('%s', 'now')) 123 + ); 115 124 116 125 create table if not exists workflows ( 117 126 id integer primary key autoincrement,
+68
spindle/db/jobs.go
··· 1 + package db 2 + 3 + import ( 4 + "context" 5 + "database/sql" 6 + "encoding/json" 7 + "tangled.org/core/api/tangled" 8 + "tangled.org/core/spindle/models" 9 + ) 10 + 11 + type JobRow struct { 12 + Id int64 13 + RepoDid string 14 + PipelineIdKnot string 15 + PipelineIdRkey string 16 + SourceRepo *tangled.Pipeline_TriggerRepo 17 + Tpl tangled.Pipeline 18 + } 19 + 20 + func (d *DB) EnqueueJob(ctx context.Context, repoDid string, pipelineId models.PipelineId, sourceRepo *tangled.Pipeline_TriggerRepo, tpl tangled.Pipeline) error { 21 + tplJson, err := json.Marshal(tpl) 22 + if err != nil { 23 + return err 24 + } 25 + _, err = d.ExecContext(ctx, ` 26 + insert into jobs (repo_did, pipeline_id_knot, pipeline_id_rkey, source_repo, tpl) 27 + values (?, ?, ?, ?, ?) 28 + `, repoDid, pipelineId.Knot, pipelineId.Rkey, string(sourceRepoJson(sourceRepo)), string(tplJson)) 29 + return err 30 + } 31 + func (d *DB) DequeueJob(ctx context.Context) (*JobRow, error) { 32 + var row JobRow 33 + var sourceRepoStr *string 34 + var tplJson string 35 + err := d.QueryRowContext(ctx, ` 36 + delete from jobs 37 + where id = ( 38 + select id from jobs 39 + order by id asc 40 + limit 1 41 + ) 42 + returning id, repo_did, pipeline_id_knot, pipeline_id_rkey, source_repo, tpl 43 + `).Scan(&row.Id, &row.RepoDid, &row.PipelineIdKnot, &row.PipelineIdRkey, &sourceRepoStr, &tplJson) 44 + if err != nil { 45 + if err == sql.ErrNoRows { 46 + return nil, nil 47 + } 48 + return nil, err 49 + } 50 + if err := json.Unmarshal([]byte(tplJson), &row.Tpl); err != nil { 51 + return nil, err 52 + } 53 + if sourceRepoStr != nil { 54 + row.SourceRepo = &tangled.Pipeline_TriggerRepo{} 55 + if err := json.Unmarshal([]byte(*sourceRepoStr), row.SourceRepo); err != nil { 56 + return nil, err 57 + } 58 + } 59 + return &row, nil 60 + } 61 + 62 + func sourceRepoJson(sr *tangled.Pipeline_TriggerRepo) []byte { 63 + if sr == nil { 64 + return nil 65 + } 66 + b, _ := json.Marshal(sr) 67 + return b 68 + }
+71 -69
spindle/server.go
··· 40 40 "tangled.org/core/spindle/engines/nixery" 41 41 "tangled.org/core/spindle/git" 42 42 "tangled.org/core/spindle/models" 43 - "tangled.org/core/spindle/queue" 44 43 "tangled.org/core/spindle/secrets" 45 44 "tangled.org/core/spindle/xrpc" 46 45 "tangled.org/core/tid" ··· 64 63 l *slog.Logger 65 64 n *notifier.Notifier 66 65 engs map[string]models.Engine 67 - jq *queue.Queue 68 66 cfg *config.Config 69 67 ks *eventconsumer.Consumer 70 68 res *idresolver.Resolver ··· 73 71 motd []byte 74 72 motdMu sync.RWMutex 75 73 rootCtx context.Context 74 + jobWake chan struct{} 76 75 } 77 76 78 77 // New creates a new Spindle server with the provided configuration and engines. ··· 116 115 return nil, fmt.Errorf("failed to run startup migrations: %w", err) 117 116 } 118 117 119 - jq := queue.NewQueue(cfg.Server.QueueSize, cfg.Server.MaxJobCount) 120 - logger.Info("initialized queue", "queueSize", cfg.Server.QueueSize, "numWorkers", cfg.Server.MaxJobCount) 121 - 122 118 collections := []string{ 123 119 tangled.SpindleMemberNSID, 124 120 tangled.RepoNSID, ··· 161 157 l: logger, 162 158 n: &n, 163 159 engs: engines, 164 - jq: jq, 165 160 cfg: cfg, 166 161 res: resolver, 167 162 verify: repoverify.New(resolver, cfg.Server.Dev), 168 163 vault: vault, 169 164 motd: defaultMotd, 170 165 rootCtx: ctx, 166 + jobWake: make(chan struct{}, 1), 171 167 } 172 168 173 169 err = e.AddSpindle(rbacDomain) ··· 236 232 return s.db 237 233 } 238 234 239 - // Queue returns the job queue instance. 240 - func (s *Spindle) Queue() *queue.Queue { 241 - return s.jq 242 - } 243 - 244 235 // Engines returns the map of available engines. 245 236 func (s *Spindle) Engines() map[string]models.Engine { 246 237 return s.engs ··· 278 269 // Start starts the Spindle server (blocking). 279 270 func (s *Spindle) Start(ctx context.Context) error { 280 271 // starts a job queue runner in the background 281 - s.jq.Start() 282 - defer s.jq.Stop() 272 + s.StartJobWorkers(ctx) 283 273 284 274 // Stop vault token renewal if it implements Stopper 285 275 if stopper, ok := s.vault.(secrets.Stopper); ok { ··· 758 748 return rawPipeline, nil 759 749 } 760 750 761 - // processPipeline enqueues the workflows in tpl. 762 - func (s *Spindle) processPipeline(repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId, sourceRepo *tangled.Pipeline_TriggerRepo) error { 763 - // derive security-relevant things like whether this run is trusted and can be passed 764 - // secrets to from the original metadata. 765 - pipelineEnv := models.PipelineEnvVarsForSource(tpl.TriggerMetadata, pipelineId, sourceRepo) 751 + func (s *Spindle) StartJobWorkers(ctx context.Context) { 752 + for range s.cfg.Server.MaxJobCount { 753 + go func() { 754 + for { 755 + job, err := s.db.DequeueJob(ctx) 756 + if err != nil { 757 + s.l.Error("failed to dequeue job", "error", err) 758 + } 759 + if job == nil { 760 + // sleep until a new job wakes us 761 + select { 762 + case <-ctx.Done(): 763 + return 764 + case <-s.jobWake: 765 + } 766 + continue 767 + } 768 + s.runJob(ctx, job) 769 + } 770 + }() 771 + } 772 + } 773 + 774 + func (s *Spindle) runJob(ctx context.Context, job *db.JobRow) { 775 + pipelineId := models.PipelineId{ 776 + Knot: job.PipelineIdKnot, 777 + Rkey: job.PipelineIdRkey, 778 + } 779 + 780 + pipelineEnv := models.PipelineEnvVarsForSource(job.Tpl.TriggerMetadata, pipelineId, job.SourceRepo) 766 781 trustedSource := true 767 - if tm := tpl.TriggerMetadata; tm != nil && tm.SourceRepo != nil && 768 - *tm.SourceRepo != "" && *tm.SourceRepo != repoDid.String() { 782 + if tm := job.Tpl.TriggerMetadata; tm != nil && tm.SourceRepo != nil && 783 + *tm.SourceRepo != "" && *tm.SourceRepo != job.RepoDid { 769 784 trustedSource = false 770 785 } 771 786 772 - // swap the repo with our sourceRepo if we are running a pipeline on a fork. 773 - // the metadata stays the same. we check whether the repo is trusted above, 774 - // so this only affects the clone URL. 775 - initTpl := tpl 776 - if sourceRepo != nil && tpl.TriggerMetadata != nil { 777 - tm := *tpl.TriggerMetadata 778 - tm.Repo = sourceRepo 787 + initTpl := job.Tpl 788 + if job.SourceRepo != nil && job.Tpl.TriggerMetadata != nil { 789 + tm := *job.Tpl.TriggerMetadata 790 + tm.Repo = job.SourceRepo 779 791 initTpl.TriggerMetadata = &tm 780 792 } 781 793 782 - // filter & init workflows 783 794 workflows := make(map[models.Engine][]models.Workflow) 784 - for _, w := range tpl.Workflows { 795 + for _, w := range job.Tpl.Workflows { 785 796 if w == nil { 786 797 continue 787 798 } 788 799 eng, ok := s.engs[w.Engine] 789 800 if !ok { 790 - err := s.db.StatusFailed(models.WorkflowId{ 801 + _ = s.db.StatusFailed(models.WorkflowId{ 791 802 PipelineId: pipelineId, 792 803 Name: w.Name, 793 804 }, fmt.Sprintf("unknown engine %#v", w.Engine), -1, s.n) 794 - if err != nil { 795 - return fmt.Errorf("db.StatusFailed: %w", err) 796 - } 797 - 798 805 continue 799 806 } 800 807 801 808 ewf, err := eng.InitWorkflow(*w, initTpl) 802 809 if err != nil { 803 - err = s.db.StatusFailed(models.WorkflowId{ 810 + _ = s.db.StatusFailed(models.WorkflowId{ 804 811 PipelineId: pipelineId, 805 812 Name: w.Name, 806 813 }, fmt.Sprintf("init workflow: %s", err), -1, s.n) 807 - if err != nil { 808 - return fmt.Errorf("db.StatusFailed: %w", err) 809 - } 810 - 811 814 continue 812 815 } 813 816 814 - // inject TANGLED_* env vars after InitWorkflow 815 - // This prevents user-defined env vars from overriding them 816 817 if ewf.Environment == nil { 817 818 ewf.Environment = make(map[string]string) 818 819 } 819 820 maps.Copy(ewf.Environment, pipelineEnv) 820 - 821 821 workflows[eng] = append(workflows[eng], *ewf) 822 822 } 823 823 824 - // enqueue pipeline 825 - ok := s.jq.Enqueue(repoDid, queue.Job{ 826 - Run: func() error { 827 - engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.rootCtx, &models.Pipeline{ 828 - RepoDid: repoDid, 829 - Workflows: workflows, 830 - TrustedSource: trustedSource, 831 - }, pipelineId) 832 - return nil 833 - }, 834 - OnFail: func(jobError error) { 835 - s.l.Error("pipeline run failed", "error", jobError) 836 - }, 837 - }) 838 - if !ok { 839 - return fmt.Errorf("failed to enqueue pipeline: queue is full") 824 + engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.rootCtx, &models.Pipeline{ 825 + RepoDid: syntax.DID(job.RepoDid), 826 + Workflows: workflows, 827 + TrustedSource: trustedSource, 828 + }, pipelineId) 829 + } 830 + 831 + // enqueues the workflows in tpl. 832 + func (s *Spindle) processPipeline(repoDid syntax.DID, tpl tangled.Pipeline, pipelineId models.PipelineId, sourceRepo *tangled.Pipeline_TriggerRepo) error { 833 + err := s.db.EnqueueJob(s.rootCtx, repoDid.String(), pipelineId, sourceRepo, tpl) 834 + if err != nil { 835 + return fmt.Errorf("failed to enqueue durable job: %w", err) 840 836 } 841 - s.l.Info("pipeline enqueued successfully", "id", pipelineId) 837 + s.l.Info("pipeline enqueued successfully to db", "id", pipelineId) 842 838 843 - // after successful enqueue, emit StatusPending for all workflows 844 - for _, ewfs := range workflows { 845 - for _, ewf := range ewfs { 846 - err := s.db.StatusPending(models.WorkflowId{ 847 - PipelineId: pipelineId, 848 - Name: ewf.Name, 849 - }, s.n) 850 - if err != nil { 851 - return fmt.Errorf("db.StatusPending: %w", err) 852 - } 839 + // wake up an idle worker to pick up more jobs if any 840 + select { 841 + case s.jobWake <- struct{}{}: 842 + default: 843 + } 844 + 845 + // pipelines visible from now on, they are sitting in queue 846 + for _, w := range tpl.Workflows { 847 + if w == nil { 848 + continue 849 + } 850 + if err := s.db.StatusPending(models.WorkflowId{ 851 + PipelineId: pipelineId, 852 + Name: w.Name, 853 + }, s.n); err != nil { 854 + return fmt.Errorf("db.StatusPending: %w", err) 853 855 } 854 856 } 855 857 return nil