alice.pds.demo.boltless.dev
This repository has no description
spindle/engine: reject workflows that map to colliding keys
Signed-off-by: dawn <dawn@tangled.org>
author
dawn
date
(Jul 23, 2026, 8:42 PM +0300)
commit
bba3c2f9
bba3c2f976560c1d84b4593e8d2eec04467fc305
parent
6c271b5d
6c271b5d8eb5e4bb8be13425865c860dbc365237
change-id
rrtttnvy
rrtttnvyxslwmssrrlqokkvrszzzyuvz
+185
-4
+28
-4
spindle/engine/engine.go
+28
-4
spindle/engine/engine.go
···
47
47
l.Error("error creating s3 client", "err", err)
48
48
}
49
49
50
+
// wid.String() is lossy so two different names can map to the same key
51
+
// eg. "foo bar" and "foo-bar"...
52
+
wfCounts := make(map[string]int)
53
+
for _, wfs := range pipeline.Workflows {
54
+
for _, w := range wfs {
55
+
wid := models.WorkflowId{
56
+
PipelineId: pipelineId,
57
+
Name: w.Name,
58
+
}
59
+
wfCounts[wid.String()]++
60
+
}
61
+
}
62
+
50
63
var wg sync.WaitGroup
51
64
for eng, wfs := range pipeline.Workflows {
52
65
workflowTimeout := eng.WorkflowTimeout()
53
66
l.Info("using workflow timeout", "timeout", workflowTimeout)
54
67
55
68
for _, w := range wfs {
56
-
wg.Go(func() {
57
-
wid := models.WorkflowId{
58
-
PipelineId: pipelineId,
59
-
Name: w.Name,
69
+
w := w
70
+
wid := models.WorkflowId{
71
+
PipelineId: pipelineId,
72
+
Name: w.Name,
73
+
}
74
+
75
+
if wfCounts[wid.String()] > 1 {
76
+
l.Warn("skipping workflow due to name collision", "wid", wid, "key", wid.String())
77
+
dbErr := db.StatusFailed(wid, fmt.Sprintf("colliding workflow name: %s; rename to something else", wid.String()), -1, n)
78
+
if dbErr != nil {
79
+
l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr)
60
80
}
81
+
continue
82
+
}
83
+
84
+
wg.Go(func() {
61
85
62
86
defer func() {
63
87
if s3 != nil {
+157
spindle/engine/engine_test.go
+157
spindle/engine/engine_test.go
···
1
+
package engine
2
+
3
+
import (
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
+
19
+
type mockStep struct {
20
+
name string
21
+
command string
22
+
}
23
+
24
+
func (m mockStep) Name() string { return m.name }
25
+
func (m mockStep) Command() string { return m.command }
26
+
func (m mockStep) Kind() models.StepKind { return models.StepKindUser }
27
+
28
+
type 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
+
37
+
func (m *mockEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
38
+
return &models.Workflow{}, nil
39
+
}
40
+
41
+
func (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
+
52
+
func (m *mockEngine) WorkflowTimeout() time.Duration {
53
+
if m.timeout != 0 {
54
+
return m.timeout
55
+
}
56
+
return 5 * time.Second
57
+
}
58
+
59
+
func (m *mockEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
60
+
return nil
61
+
}
62
+
63
+
func (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
+
75
+
func 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
+
84
+
func 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, 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
+
}