Monorepo for Tangled
spindle/engine: add acquire modes and logger-provider seam
+209
-69
+38
-36
spindle/engine/engine.go
+38
-36
spindle/engine/engine.go
···
18
18
var (
19
19
ErrTimedOut = errors.New("timed out")
20
20
ErrWorkflowFailed = errors.New("workflow failed")
21
+
ErrCancelled = errors.New("workflow cancelled")
21
22
)
22
23
23
-
type workflowFinalizer interface {
24
-
FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error
24
+
// workflowLoggerProvider lets an engine supply the WorkflowLogger StartWorkflows
25
+
// uses, instead of the default on-disk one. Engines that don't implement it get
26
+
// a file logger under cfg.Server.LogDir. The mill engine supplies a no-op logger
27
+
// because it doesn't run steps locally — it relays the executor's real log file
28
+
// into place itself, so a second logger here would only write competing lines.
29
+
type workflowLoggerProvider interface {
30
+
WorkflowLogger(wid models.WorkflowId) models.WorkflowLogger
31
+
}
32
+
33
+
func reportWorkflowStatusError(l *slog.Logger, database *db.DB, n *notifier.Notifier, wid models.WorkflowId, err error) {
34
+
if errors.Is(err, ErrTimedOut) {
35
+
dbErr := database.StatusTimeout(wid, n)
36
+
if dbErr != nil {
37
+
l.Error("failed to set workflow status to timeout", "wid", wid, "err", dbErr)
38
+
}
39
+
} else if errors.Is(err, ErrCancelled) {
40
+
dbErr := database.StatusCancelled(wid, err.Error(), -1, n)
41
+
if dbErr != nil {
42
+
l.Error("failed to set workflow status to cancelled", "wid", wid, "err", dbErr)
43
+
}
44
+
} else {
45
+
dbErr := database.StatusFailed(wid, err.Error(), -1, n)
46
+
if dbErr != nil {
47
+
l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
48
+
}
49
+
}
25
50
}
26
51
27
52
func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) {
···
68
93
}
69
94
}()
70
95
71
-
wfLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues)
72
-
if err != nil {
96
+
var wfLogger models.WorkflowLogger
97
+
if p, ok := eng.(workflowLoggerProvider); ok {
98
+
wfLogger = p.WorkflowLogger(wid)
99
+
} else if fileLogger, err := models.NewFileWorkflowLogger(cfg.Server.LogDir, wid, secretValues); err != nil {
73
100
l.Warn("failed to setup step logger; logs will not be persisted", "error", err)
74
101
wfLogger = models.NullLogger{}
75
102
} else {
76
103
l.Info("setup step logger; logs will be persisted", "logDir", cfg.Server.LogDir, "wid", wid)
77
-
defer wfLogger.Close()
104
+
wfLogger = fileLogger
105
+
defer fileLogger.Close()
78
106
}
79
107
80
108
l.Info("waiting for slot", "wid", wid)
81
109
slot := WorkflowSlot(NoopSlot{})
82
110
if s, ok := eng.(WorkflowSlotter); ok {
83
111
var err error
84
-
slot, err = s.AcquireWorkflowSlot(ctx, wid, &w)
112
+
slot, err = s.AcquireWorkflowSlot(ctx, wid, &w, Wait)
85
113
if err != nil {
86
114
l.Error("failed to acquire slot", "wid", wid, "err", err)
87
-
dbErr := db.StatusFailed(wid, err.Error(), -1, n)
88
-
if dbErr != nil {
89
-
l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
90
-
}
115
+
reportWorkflowStatusError(l, db, n, wid, err)
91
116
return
92
117
}
93
118
}
94
119
defer slot.Release()
95
120
96
-
err = db.StatusRunning(wid, n)
121
+
err := db.StatusRunning(wid, n)
97
122
if err != nil {
98
123
l.Error("failed to set workflow status to running", "wid", wid, "err", err)
99
124
return
···
110
135
l.Error("failed to destroy workflow after setup failure", "error", destroyErr)
111
136
}
112
137
113
-
dbErr := db.StatusFailed(wid, err.Error(), -1, n)
114
-
if dbErr != nil {
115
-
l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
116
-
}
138
+
reportWorkflowStatusError(l, db, n, wid, err)
117
139
return
118
140
}
119
141
defer eng.DestroyWorkflow(ctx, wid)
···
139
161
}
140
162
141
163
if err != nil {
142
-
if errors.Is(err, ErrTimedOut) {
143
-
dbErr := db.StatusTimeout(wid, n)
144
-
if dbErr != nil {
145
-
l.Error("failed to set workflow status to timeout", "wid", wid, "err", dbErr)
146
-
}
147
-
} else {
148
-
dbErr := db.StatusFailed(wid, err.Error(), -1, n)
149
-
if dbErr != nil {
150
-
l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
151
-
}
152
-
}
153
-
return
154
-
}
155
-
}
156
-
157
-
if finalizer, ok := eng.(workflowFinalizer); ok {
158
-
if err := finalizer.FinalizeWorkflow(ctx, wid, &w, wfLogger); err != nil {
159
-
dbErr := db.StatusFailed(wid, err.Error(), -1, n)
160
-
if dbErr != nil {
161
-
l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
162
-
}
164
+
reportWorkflowStatusError(l, db, n, wid, err)
163
165
return
164
166
}
165
167
}
+14
-7
spindle/engine/scheduler.go
+14
-7
spindle/engine/scheduler.go
···
50
50
}
51
51
}
52
52
53
-
func (s *ResourceScheduler[R]) Acquire(ctx context.Context, req R) (WorkflowSlot, error) {
53
+
// Acquire reserves resources for req. With Wait it blocks (queueing with aging)
54
+
// until the request fits; with NoWait it tries once and returns
55
+
// ErrNoWorkflowSlots if it can't be served right now, never queueing (the mill
56
+
// owns the backlog, so a NoWait caller either has room this instant or says no).
57
+
func (s *ResourceScheduler[R]) Acquire(ctx context.Context, req R, mode AcquireMode) (WorkflowSlot, error) {
54
58
if s == nil {
55
59
return NoopSlot{}, nil
56
60
}
···
60
64
s.mu.Unlock()
61
65
return nil, fmt.Errorf("%w: request=%v budget=%v max=%v", ErrNoWorkflowSlots, req, s.budget, s.max)
62
66
}
63
-
if len(s.queue) == 0 && s.used.Add(req).Fits(s.budget) {
67
+
// can serve immediately? a NoWait caller ignores the queue (it never waits
68
+
// behind aged waiters); a Wait caller only jumps in when the queue is empty.
69
+
if s.used.Add(req).Fits(s.budget) && (mode == NoWait || len(s.queue) == 0) {
64
70
s.used = s.used.Add(req)
65
71
s.mu.Unlock()
66
72
return &resourceLease[R]{scheduler: s, req: req}, nil
73
+
}
74
+
if mode == NoWait {
75
+
used := s.used
76
+
s.mu.Unlock()
77
+
return nil, fmt.Errorf("%w: request=%v used=%v budget=%v", ErrNoWorkflowSlots, req, used, s.budget)
67
78
}
68
79
69
80
waiter := &resourceWaiter[R]{req: req, ready: make(chan struct{}), enqueuedAt: s.now()}
···
129
140
}
130
141
131
142
func (s *ResourceScheduler[R]) remove(waiter *resourceWaiter[R]) {
132
-
for i, candidate := range s.queue {
133
-
if candidate != waiter {
134
-
continue
135
-
}
143
+
if i := slices.Index(s.queue, waiter); i >= 0 {
136
144
s.queue = slices.Delete(s.queue, i, i+1)
137
-
return
138
145
}
139
146
}
+80
-9
spindle/engine/scheduler_test.go
+80
-9
spindle/engine/scheduler_test.go
···
36
36
37
37
scheduler := NewResourceScheduler(ru{}, ru{}, 0)
38
38
39
-
slot, err := scheduler.Acquire(context.Background(), ru{a: 1 << 20, b: 1 << 20})
39
+
slot, err := scheduler.Acquire(context.Background(), ru{a: 1 << 20, b: 1 << 20}, Wait)
40
40
if err != nil {
41
41
t.Fatalf("Acquire() error = %v", err)
42
42
}
···
48
48
49
49
scheduler := NewResourceScheduler(ru{a: 1024, b: 10_000}, ru{a: 512, b: 5_000}, 0)
50
50
51
-
_, err := scheduler.Acquire(context.Background(), ru{a: 768, b: 100})
51
+
_, err := scheduler.Acquire(context.Background(), ru{a: 768, b: 100}, Wait)
52
52
if !errors.Is(err, ErrNoWorkflowSlots) {
53
53
t.Fatalf("Acquire() error = %v, want ErrNoWorkflowSlots", err)
54
54
}
55
55
56
-
_, err = scheduler.Acquire(context.Background(), ru{a: 128, b: 12_000})
56
+
_, err = scheduler.Acquire(context.Background(), ru{a: 128, b: 12_000}, Wait)
57
57
if !errors.Is(err, ErrNoWorkflowSlots) {
58
58
t.Fatalf("Acquire() error = %v, want ErrNoWorkflowSlots", err)
59
59
}
···
64
64
65
65
scheduler := NewResourceScheduler(ru{a: 1024}, ru{}, 0)
66
66
67
-
first, err := scheduler.Acquire(context.Background(), ru{a: 1024})
67
+
first, err := scheduler.Acquire(context.Background(), ru{a: 1024}, Wait)
68
68
if err != nil {
69
69
t.Fatalf("first Acquire() error = %v", err)
70
70
}
···
85
85
86
86
scheduler := NewResourceScheduler(ru{a: 1}, ru{}, 0)
87
87
88
-
slot, err := scheduler.Acquire(context.Background(), ru{a: 1})
88
+
slot, err := scheduler.Acquire(context.Background(), ru{a: 1}, Wait)
89
89
if err != nil {
90
90
t.Fatalf("Acquire() error = %v", err)
91
91
}
···
93
93
slot.Release()
94
94
slot.Release()
95
95
96
-
second, err := scheduler.Acquire(context.Background(), ru{a: 1})
96
+
second, err := scheduler.Acquire(context.Background(), ru{a: 1}, Wait)
97
97
if err != nil {
98
98
t.Fatalf("Acquire() after double release error = %v", err)
99
99
}
···
105
105
106
106
scheduler := NewResourceScheduler(ru{a: 1024}, ru{}, time.Hour) // disable aging so we test pure backfill
107
107
108
-
hold, err := scheduler.Acquire(context.Background(), ru{a: 512})
108
+
hold, err := scheduler.Acquire(context.Background(), ru{a: 512}, Wait)
109
109
if err != nil {
110
110
t.Fatalf("hold Acquire() error = %v", err)
111
111
}
···
128
128
fakeNow := time.Now()
129
129
scheduler.now = func() time.Time { return fakeNow }
130
130
131
-
hold, err := scheduler.Acquire(context.Background(), ru{a: 512})
131
+
hold, err := scheduler.Acquire(context.Background(), ru{a: 512}, Wait)
132
132
if err != nil {
133
133
t.Fatalf("hold Acquire() error = %v", err)
134
134
}
···
151
151
big.Release()
152
152
}
153
153
154
+
func TestResourceSchedulerTryRejectsWhenNoRoomNow(t *testing.T) {
155
+
t.Parallel()
156
+
157
+
scheduler := NewResourceScheduler(ru{a: 1024}, ru{}, 0)
158
+
159
+
first, err := scheduler.Acquire(context.Background(), ru{a: 1024}, NoWait)
160
+
if err != nil {
161
+
t.Fatalf("first Acquire(NoWait) error = %v", err)
162
+
}
163
+
164
+
// budget is full now: a try must reject immediately, never queue.
165
+
if _, err := scheduler.Acquire(context.Background(), ru{a: 1}, NoWait); !errors.Is(err, ErrNoWorkflowSlots) {
166
+
t.Fatalf("Acquire(NoWait) error = %v, want ErrNoWorkflowSlots", err)
167
+
}
168
+
169
+
first.Release()
170
+
171
+
second, err := scheduler.Acquire(context.Background(), ru{a: 1}, NoWait)
172
+
if err != nil {
173
+
t.Fatalf("Acquire(NoWait) after release error = %v", err)
174
+
}
175
+
second.Release()
176
+
}
177
+
178
+
func TestResourceSchedulerTryRejectsRequestsThatCanNeverFit(t *testing.T) {
179
+
t.Parallel()
180
+
181
+
scheduler := NewResourceScheduler(ru{a: 1024, b: 10_000}, ru{a: 512, b: 5_000}, 0)
182
+
183
+
if _, err := scheduler.Acquire(context.Background(), ru{a: 768, b: 100}, NoWait); !errors.Is(err, ErrNoWorkflowSlots) {
184
+
t.Fatalf("Acquire(NoWait) error = %v, want ErrNoWorkflowSlots", err)
185
+
}
186
+
}
187
+
188
+
// a try must not be gated behind an aged, blocked waiter sitting in the queue;
189
+
// it looks only at live usage vs budget.
190
+
func TestResourceSchedulerTryIgnoresQueuedWaiters(t *testing.T) {
191
+
t.Parallel()
192
+
193
+
scheduler := NewResourceScheduler(ru{a: 1024}, ru{}, time.Hour)
194
+
195
+
hold, err := scheduler.Acquire(context.Background(), ru{a: 512}, Wait)
196
+
if err != nil {
197
+
t.Fatalf("hold Acquire() error = %v", err)
198
+
}
199
+
defer hold.Release()
200
+
201
+
// a big blocking waiter parks in the queue.
202
+
bigCh := acquireAsync(context.Background(), scheduler, ru{a: 768})
203
+
assertAcquireBlocked(t, bigCh)
204
+
205
+
// a try that fits alongside the held 512 must succeed despite the queued waiter.
206
+
slot, err := scheduler.Acquire(context.Background(), ru{a: 256}, NoWait)
207
+
if err != nil {
208
+
t.Fatalf("Acquire(NoWait) error = %v, want success past queued waiter", err)
209
+
}
210
+
slot.Release()
211
+
}
212
+
213
+
func TestResourceSchedulerZeroLimitsTryDoesNotReject(t *testing.T) {
214
+
t.Parallel()
215
+
216
+
scheduler := NewResourceScheduler(ru{}, ru{}, 0)
217
+
218
+
slot, err := scheduler.Acquire(context.Background(), ru{a: 1 << 20, b: 1 << 20}, NoWait)
219
+
if err != nil {
220
+
t.Fatalf("Acquire(NoWait) error = %v", err)
221
+
}
222
+
slot.Release()
223
+
}
224
+
154
225
func acquireAsync(ctx context.Context, scheduler *ResourceScheduler[ru], req ru) <-chan acquireResult {
155
226
ch := make(chan acquireResult, 1)
156
227
go func() {
157
-
slot, err := scheduler.Acquire(ctx, req)
228
+
slot, err := scheduler.Acquire(ctx, req, Wait)
158
229
ch <- acquireResult{slot: slot, err: err}
159
230
}()
160
231
return ch
+23
-2
spindle/engine/slot.go
+23
-2
spindle/engine/slot.go
···
13
13
Release()
14
14
}
15
15
16
+
// AcquireMode selects whether acquiring a slot may block.
17
+
type AcquireMode int
18
+
19
+
const (
20
+
// Wait blocks (and may queue) until a slot frees, honouring ctx. This is the
21
+
// standalone behaviour and the zero value.
22
+
Wait AcquireMode = iota
23
+
// NoWait tries once and returns ErrNoWorkflowSlots if no slot is free right
24
+
// now, never queueing. Executors use this: the mill owns the backlog, so an
25
+
// executor either has a seat this instant or it says no.
26
+
NoWait
27
+
)
28
+
16
29
type WorkflowSlotter interface {
17
-
AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow) (WorkflowSlot, error)
30
+
AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode AcquireMode) (WorkflowSlot, error)
18
31
}
19
32
20
33
type releaseFunc func()
···
41
54
return &SemaphoreSlotter{slots: make(chan struct{}, maxConcurrent)}
42
55
}
43
56
44
-
func (a *SemaphoreSlotter) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow) (WorkflowSlot, error) {
57
+
func (a *SemaphoreSlotter) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode AcquireMode) (WorkflowSlot, error) {
45
58
if a == nil || a.slots == nil {
46
59
return NoopSlot{}, nil
60
+
}
61
+
if mode == NoWait {
62
+
select {
63
+
case a.slots <- struct{}{}:
64
+
return releaseFunc(func() { <-a.slots }), nil
65
+
default:
66
+
return nil, ErrNoWorkflowSlots
67
+
}
47
68
}
48
69
select {
49
70
case a.slots <- struct{}{}:
+42
-5
spindle/engine/slot_test.go
+42
-5
spindle/engine/slot_test.go
···
15
15
slotter := NewSemaphoreSlotter(0)
16
16
17
17
for range 10 {
18
-
slot, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil)
18
+
slot, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, Wait)
19
19
if err != nil {
20
20
t.Fatalf("AcquireWorkflowSlot() error = %v", err)
21
21
}
···
28
28
29
29
slotter := NewSemaphoreSlotter(1)
30
30
31
-
first, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil)
31
+
first, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, Wait)
32
32
if err != nil {
33
33
t.Fatalf("first AcquireWorkflowSlot() error = %v", err)
34
34
}
···
42
42
acquired := make(chan WorkflowSlot, 1)
43
43
errs := make(chan error, 1)
44
44
go func() {
45
-
slot, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil)
45
+
slot, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, Wait)
46
46
if err != nil {
47
47
errs <- err
48
48
return
···
64
64
65
65
slotter := NewSemaphoreSlotter(1)
66
66
67
-
first, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil)
67
+
first, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, Wait)
68
68
if err != nil {
69
69
t.Fatalf("first AcquireWorkflowSlot() error = %v", err)
70
70
}
···
73
73
ctx, cancel := context.WithCancel(context.Background())
74
74
cancel()
75
75
76
-
_, err = slotter.AcquireWorkflowSlot(ctx, zeroWorkflowID(), nil)
76
+
_, err = slotter.AcquireWorkflowSlot(ctx, zeroWorkflowID(), nil, Wait)
77
77
if !errors.Is(err, context.Canceled) {
78
78
t.Fatalf("AcquireWorkflowSlot() error = %v, want context.Canceled", err)
79
79
}
80
+
}
81
+
82
+
func TestSemaphoreSlotterTryDisabledDoesNotReject(t *testing.T) {
83
+
t.Parallel()
84
+
85
+
slotter := NewSemaphoreSlotter(0)
86
+
87
+
for range 10 {
88
+
slot, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, NoWait)
89
+
if err != nil {
90
+
t.Fatalf("AcquireWorkflowSlot(NoWait) error = %v", err)
91
+
}
92
+
slot.Release()
93
+
}
94
+
}
95
+
96
+
func TestSemaphoreSlotterTryRejectsWhenFull(t *testing.T) {
97
+
t.Parallel()
98
+
99
+
slotter := NewSemaphoreSlotter(1)
100
+
101
+
first, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, NoWait)
102
+
if err != nil {
103
+
t.Fatalf("first AcquireWorkflowSlot(NoWait) error = %v", err)
104
+
}
105
+
106
+
if _, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, NoWait); !errors.Is(err, ErrNoWorkflowSlots) {
107
+
t.Fatalf("AcquireWorkflowSlot(NoWait) error = %v, want ErrNoWorkflowSlots", err)
108
+
}
109
+
110
+
first.Release()
111
+
112
+
second, err := slotter.AcquireWorkflowSlot(context.Background(), zeroWorkflowID(), nil, NoWait)
113
+
if err != nil {
114
+
t.Fatalf("AcquireWorkflowSlot(NoWait) after release error = %v", err)
115
+
}
116
+
second.Release()
80
117
}
81
118
82
119
func assertNotAcquired(t *testing.T, acquired <-chan WorkflowSlot, errs <-chan error) {
+8
spindle/engines/dummy/engine.go
+8
spindle/engines/dummy/engine.go
···
8
8
9
9
"gopkg.in/yaml.v3"
10
10
"tangled.org/core/api/tangled"
11
+
"tangled.org/core/spindle/engine"
11
12
"tangled.org/core/spindle/models"
12
13
"tangled.org/core/spindle/secrets"
13
14
)
···
77
78
78
79
func (e *DummyEngine) WorkflowTimeout() time.Duration {
79
80
return 5 * time.Minute
81
+
}
82
+
83
+
// AcquireWorkflowSlot lets the dummy engine satisfy WorkflowSlotter (so it can
84
+
// run standalone or be driven by the executor shim). The dummy has no capacity
85
+
// limit, so it always grants a no-op slot regardless of mode.
86
+
func (e *DummyEngine) AcquireWorkflowSlot(_ context.Context, _ models.WorkflowId, _ *models.Workflow, _ engine.AcquireMode) (engine.WorkflowSlot, error) {
87
+
return engine.NoopSlot{}, nil
80
88
}
81
89
82
90
func (e *DummyEngine) DestroyWorkflow(_ context.Context, wid models.WorkflowId) error {
+2
-5
spindle/engines/microvm/budget.go
+2
-5
spindle/engines/microvm/budget.go
···
66
66
return budget, maxReq, cfg.AgingThreshold
67
67
}
68
68
69
-
func (e *Engine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow) (engine.WorkflowSlot, error) {
69
+
func (e *Engine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode engine.AcquireMode) (engine.WorkflowSlot, error) {
70
70
state, ok := wf.Data.(*workflowState)
71
71
if !ok || state == nil {
72
72
return nil, fmt.Errorf("microVM workflow state is not initialized")
···
75
75
return engine.NoopSlot{}, nil
76
76
}
77
77
req := resourcesForImage(state.ImageSpec)
78
-
if req.MemoryMiB < 0 || req.VCPUs < 0 || req.DiskMiB < 0 {
79
-
return nil, fmt.Errorf("microVM resource request must not be negative: %s", req)
80
-
}
81
-
return e.scheduler.Acquire(ctx, req)
78
+
return e.scheduler.Acquire(ctx, req, mode)
82
79
}
83
80
84
81
func resourcesForImage(spec ImageSpec) Resources {
-4
spindle/engines/microvm/engine.go
-4
spindle/engines/microvm/engine.go
···
542
542
return cleanupErr
543
543
}
544
544
545
-
func (e *Engine) FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, w *models.Workflow, wfLogger models.WorkflowLogger) error {
546
-
return nil
547
-
}
548
-
549
545
func (e *Engine) WorkflowTimeout() time.Duration {
550
546
d, err := time.ParseDuration(e.cfg.MicroVMPipelines.WorkflowTimeout)
551
547
if err != nil {
+2
-1
spindle/engines/nixery/engine.go
+2
-1
spindle/engines/nixery/engine.go
···
187
187
ctx context.Context,
188
188
wid models.WorkflowId,
189
189
wf *models.Workflow,
190
+
mode engine.AcquireMode,
190
191
) (engine.WorkflowSlot, error) {
191
192
if e.slotter == nil {
192
193
return engine.NoopSlot{}, nil
193
194
}
194
195
195
-
return e.slotter.AcquireWorkflowSlot(ctx, wid, wf)
196
+
return e.slotter.AcquireWorkflowSlot(ctx, wid, wf, mode)
196
197
}
197
198
198
199
func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) (err error) {