alice.pds.demo.boltless.dev
This repository has no description
1package engine
2
3import (
4 "context"
5 "log/slog"
6 "os"
7 "path/filepath"
8 "sync"
9 "testing"
10 "time"
11
12 "tangled.org/core/api/tangled"
13 "tangled.org/core/spindle/config"
14 "tangled.org/core/spindle/db"
15 "tangled.org/core/spindle/models"
16 "tangled.org/core/spindle/secrets"
17)
18
19type mockStep struct {
20 name string
21 command string
22}
23
24func (m mockStep) Name() string { return m.name }
25func (m mockStep) Command() string { return m.command }
26func (m mockStep) Kind() models.StepKind { return models.StepKindUser }
27
28type mockEngine struct {
29 mu sync.Mutex
30 setupCalls []models.WorkflowId
31 runStepCalls []models.WorkflowId
32 setupFunc func(ctx context.Context, wid models.WorkflowId) error
33 runStepFunc func(ctx context.Context, wid models.WorkflowId, idx int) error
34 timeout time.Duration
35}
36
37func (m *mockEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
38 return &models.Workflow{}, nil
39}
40
41func (m *mockEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error {
42 m.mu.Lock()
43 m.setupCalls = append(m.setupCalls, wid)
44 fn := m.setupFunc
45 m.mu.Unlock()
46 if fn != nil {
47 return fn(ctx, wid)
48 }
49 return nil
50}
51
52func (m *mockEngine) WorkflowTimeout() time.Duration {
53 if m.timeout != 0 {
54 return m.timeout
55 }
56 return 5 * time.Second
57}
58
59func (m *mockEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
60 return nil
61}
62
63func (m *mockEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
64 m.mu.Lock()
65 m.runStepCalls = append(m.runStepCalls, wid)
66 fn := m.runStepFunc
67 m.mu.Unlock()
68
69 if fn != nil {
70 return fn(ctx, wid, idx)
71 }
72 return nil
73}
74
75func newTestDB(t *testing.T) *db.DB {
76 t.Helper()
77 d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db"))
78 if err != nil {
79 t.Fatalf("failed to create test db: %v", err)
80 }
81 return d
82}
83
84func TestStartWorkflows_CollisionRejection(t *testing.T) {
85 t.Parallel()
86
87 testDB := newTestDB(t)
88 logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
89
90 eng := &mockEngine{}
91 pipelineId := models.PipelineId{
92 Knot: "test-knot",
93 Rkey: "test-rkey",
94 }
95
96 // two names that normalize to the same wid must not both run
97 wfColliding1 := models.Workflow{
98 Name: "test-job",
99 Steps: []models.Step{mockStep{name: "step1"}},
100 }
101 wfColliding2 := models.Workflow{
102 Name: "test job",
103 Steps: []models.Step{mockStep{name: "step1"}},
104 }
105 wfUnique := models.Workflow{
106 Name: "unique_job",
107 Steps: []models.Step{mockStep{name: "step1"}},
108 }
109
110 pipeline := &models.Pipeline{
111 Workflows: map[models.Engine][]models.Workflow{
112 eng: {wfColliding1, wfColliding2, wfUnique},
113 },
114 }
115
116 cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
117 StartWorkflows(logger, nil, cfg, testDB, nil, nil, context.Background(), pipeline, pipelineId)
118
119 eng.mu.Lock()
120 setupCalls := append([]models.WorkflowId(nil), eng.setupCalls...)
121 eng.mu.Unlock()
122
123 for _, call := range setupCalls {
124 if call.Name == "test-job" || call.Name == "test job" {
125 t.Fatalf("expected colliding workflow %s to not be started", call.Name)
126 }
127 }
128
129 hasUnique := false
130 for _, call := range setupCalls {
131 if call.Name == "unique_job" {
132 hasUnique = true
133 }
134 }
135 if !hasUnique {
136 t.Fatalf("expected unique workflow unique_job to be started")
137 }
138
139 widColliding1 := models.WorkflowId{PipelineId: pipelineId, Name: "test-job"}
140 widColliding2 := models.WorkflowId{PipelineId: pipelineId, Name: "test job"}
141 widUnique := models.WorkflowId{PipelineId: pipelineId, Name: "unique_job"}
142
143 status1, err := testDB.GetStatus(widColliding1)
144 if err != nil || status1.Status != string(models.StatusKindFailed) {
145 t.Fatalf("expected colliding1 status to be failed, got status=%v err=%v", status1, err)
146 }
147
148 status2, err := testDB.GetStatus(widColliding2)
149 if err != nil || status2.Status != string(models.StatusKindFailed) {
150 t.Fatalf("expected colliding2 status to be failed, got status=%v err=%v", status2, err)
151 }
152
153 statusUnique, err := testDB.GetStatus(widUnique)
154 if err != nil || statusUnique.Status != string(models.StatusKindSuccess) {
155 t.Fatalf("expected unique status to be success, got status=%v err=%v", statusUnique, err)
156 }
157}
158
159func TestCancelWorkflow_NotOverwritten(t *testing.T) {
160 t.Parallel()
161
162 testDB := newTestDB(t)
163 logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
164
165 stepStarted := make(chan struct{})
166 eng := &mockEngine{
167 runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int) error {
168 close(stepStarted)
169 <-ctx.Done()
170 return ctx.Err()
171 },
172 }
173
174 pipelineId := models.PipelineId{
175 Knot: "test-knot",
176 Rkey: "test-rkey",
177 }
178
179 wid := models.WorkflowId{
180 PipelineId: pipelineId,
181 Name: "cancel_test_job",
182 }
183
184 pipeline := &models.Pipeline{
185 Workflows: map[models.Engine][]models.Workflow{
186 eng: {
187 {
188 Name: "cancel_test_job",
189 Steps: []models.Step{mockStep{name: "step1"}},
190 },
191 },
192 },
193 }
194
195 cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
196 doneChan := make(chan struct{})
197 go func() {
198 StartWorkflows(logger, nil, cfg, testDB, nil, nil, context.Background(), pipeline, pipelineId)
199 close(doneChan)
200 }()
201
202 select {
203 case <-stepStarted:
204 case <-time.After(5 * time.Second):
205 t.Fatal("timed out waiting for step to start")
206 }
207
208 _ = testDB.StatusCancelled(wid, "User canceled the workflow", -1, nil)
209 CancelWorkflow(wid)
210
211 select {
212 case <-doneChan:
213 case <-time.After(5 * time.Second):
214 t.Fatal("timed out waiting for StartWorkflows to complete")
215 }
216
217 // the runner writes StatusCancelled itself when it sees the canceled ctx
218 // the handler writes nothing for a live wf, so nothing lands after to overwrite it
219 st, err := testDB.GetStatus(wid)
220 if err != nil {
221 t.Fatalf("GetStatus error = %v", err)
222 }
223 if st.Status != string(models.StatusKindCancelled) {
224 t.Fatalf("expected status to be cancelled, got %s", st.Status)
225 }
226}
227
228func TestSetupTimeout_ReportsTimeout(t *testing.T) {
229 t.Parallel()
230
231 testDB := newTestDB(t)
232 logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
233
234 // setup blocks past the workflow timeout, so it should land as timeout not failed
235 eng := &mockEngine{
236 timeout: 100 * time.Millisecond,
237 setupFunc: func(ctx context.Context, wid models.WorkflowId) error {
238 <-ctx.Done()
239 return ctx.Err()
240 },
241 }
242
243 pipelineId := models.PipelineId{Knot: "test-knot", Rkey: "test-rkey"}
244 wid := models.WorkflowId{PipelineId: pipelineId, Name: "timeout_job"}
245
246 pipeline := &models.Pipeline{
247 Workflows: map[models.Engine][]models.Workflow{
248 eng: {{Name: "timeout_job", Steps: []models.Step{mockStep{name: "step1"}}}},
249 },
250 }
251
252 cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
253 StartWorkflows(logger, nil, cfg, testDB, nil, nil, context.Background(), pipeline, pipelineId)
254
255 st, err := testDB.GetStatus(wid)
256 if err != nil {
257 t.Fatalf("GetStatus error = %v", err)
258 }
259 if st.Status != string(models.StatusKindTimeout) {
260 t.Fatalf("expected status to be timeout, got %s", st.Status)
261 }
262
263 if len(eng.runStepCalls) != 0 {
264 t.Fatalf("expected no steps to run after setup timeout, got %d", len(eng.runStepCalls))
265 }
266}