engine.go
1package engine
2
3import (
4 "context"
5 "encoding/json"
6 "errors"
7 "fmt"
8 "os"
9 "path/filepath"
10
11 "git.kilimanjaro.io/ci/internal/lex"
12 "git.kilimanjaro.io/ci/internal/manifest"
13 "git.kilimanjaro.io/ci/internal/runlog"
14)
15
16// DefaultMaxTransitions is the loop guard (DESIGN.md ยง8 rule 4).
17const DefaultMaxTransitions = 50
18
19// Options change how a run executes.
20type Options struct {
21 // Stubs maps an action to a result file used instead of running it.
22 Stubs map[string]string
23 // Until stops the run before this action would run.
24 Until string
25 // NoFollow runs the first action only and reports what would follow.
26 NoFollow bool
27 // MaxTransitions overrides DefaultMaxTransitions.
28 MaxTransitions int
29 // Gates decides gates as they are reached; nil pauses at every gate.
30 Gates Gatekeeper
31 // Detach pauses at every gate even with a Gatekeeper (the runner's
32 // mode; locally `ci run --detach`).
33 Detach bool
34 // LocksDir holds concurrency locks; default <runs>/.locks.
35 LocksDir string
36 // Locker overrides the file locks in LocksDir (the runner's, for local
37 // runs that must not overlap CI deploys).
38 Locker Locker
39}
40
41// Engine runs one workflow run.
42type Engine struct {
43 M *manifest.Manifest
44 Dir Dir
45 Exec Executor
46 Log *runlog.Writer
47 BaseEnv []string
48 Opts Options
49
50 router *Router
51 st *State
52 locks map[string]func()
53}
54
55// Create starts a new run in dir: it writes the manifest copy, state.json and
56// the first event. Close the engine's Log when done.
57func Create(m *manifest.Manifest, dir Dir, meta Meta, ex Executor, opts Options) (*Engine, error) {
58 if err := dir.Create(); err != nil {
59 return nil, err
60 }
61 if _, err := os.Stat(dir.StateFile()); err == nil {
62 return nil, fmt.Errorf("run %s already exists", dir)
63 }
64 if err := os.WriteFile(dir.ManifestFile(), m.Raw, 0o644); err != nil {
65 return nil, err
66 }
67 log, err := runlog.Open(dir.Events(), meta.Run)
68 if err != nil {
69 return nil, err
70 }
71 t := now()
72 e := &Engine{
73 M: m, Dir: dir, Exec: ex, Log: log, BaseEnv: osEnviron(), Opts: opts,
74 router: &Router{M: m, Detached: meta.Parent != ""},
75 st: &State{Version: StateVersion, Meta: meta, Status: StatusRunning, Created: t},
76 }
77 for _, c := range meta.Checkouts {
78 e.emit(lex.RunEvent{Kind: lex.KindCheckout, Repo: c.Repo, Rev: c.Rev, Path: c.Path})
79 }
80 detail := fmt.Sprintf("%s %s", m.Name, meta.Repo)
81 if meta.Sha != "" {
82 detail += "@" + short(meta.Sha)
83 }
84 if meta.Dirty {
85 detail += " (working copy with changes)"
86 }
87 e.emit(lex.RunEvent{Kind: lex.KindRun, Status: StatusRunning, Content: detail})
88 if err := e.st.Save(dir); err != nil {
89 return nil, err
90 }
91 return e, nil
92}
93
94func short(sha string) string {
95 if len(sha) > 12 {
96 return sha[:12]
97 }
98 return sha
99}
100
101// State returns the run's current state.
102func (e *Engine) State() *State { return e.st }
103
104// Run runs the workflow from its first step.
105func (e *Engine) Run(ctx context.Context) (*State, error) {
106 return e.drive(ctx, e.router.Start())
107}
108
109// RunFrom runs the workflow starting at action, as if a person had chosen
110// it; routing then follows that action's result.
111func (e *Engine) RunFrom(ctx context.Context, action string) (*State, error) {
112 if _, ok := e.M.Actions[action]; !ok {
113 return nil, fmt.Errorf("no action %q in %s", action, e.M.Name)
114 }
115 return e.drive(ctx, e.router.Begin(action))
116}
117
118func (e *Engine) maxTransitions() int {
119 if e.Opts.MaxTransitions > 0 {
120 return e.Opts.MaxTransitions
121 }
122 return DefaultMaxTransitions
123}
124
125// drive follows decisions until the run ends.
126func (e *Engine) drive(ctx context.Context, d Decision) (*State, error) {
127 for {
128 if ctx.Err() != nil {
129 return e.finish(End{StatusCancelled, lex.ReasonCancel, "interrupted"}, "")
130 }
131 switch {
132 case d.End != nil:
133 return e.finish(*d.End, e.lastAction())
134 case d.Gate != nil:
135 next, paused, err := e.reachGate(ctx, d.Gate)
136 if err != nil {
137 return e.finish(End{StatusFailure, lex.ReasonError, err.Error()}, e.lastAction())
138 }
139 if paused {
140 return e.st, nil
141 }
142 d = next
143 case d.Run != nil:
144 p := *d.Run
145 if e.st.Seq >= e.maxTransitions() {
146 return e.finish(End{StatusFailure, lex.ReasonLimit,
147 fmt.Sprintf("more than %d transitions; is a result routing in a loop?", e.maxTransitions())}, p.From)
148 }
149 if p.Action == e.Opts.Until {
150 return e.finish(End{StatusCancelled, lex.ReasonUntil, "stopped before " + p.Action + " (--until)"}, p.From)
151 }
152 e.transition(lex.Transition{From: p.From, To: p.Action, Then: p.Then, Reason: p.Reason})
153 res, file, err := e.execute(ctx, p)
154 if err != nil {
155 if ctx.Err() != nil {
156 return e.finish(End{StatusCancelled, lex.ReasonCancel, "interrupted during " + p.Action}, p.Action)
157 }
158 return e.finish(End{StatusFailure, lex.ReasonError, err.Error()}, p.Action)
159 }
160 next := e.router.After(e.st, p, res, file)
161 if e.Opts.NoFollow {
162 status := StatusSuccess
163 if res.Status == lex.Failure {
164 status = StatusFailure
165 }
166 return e.finish(End{status, lex.ReasonStop, "not followed (--no-follow): would " + next.String()}, p.Action)
167 }
168 d = next
169 default:
170 return e.finish(End{StatusFailure, lex.ReasonError, "router returned no decision"}, e.lastAction())
171 }
172 }
173}
174
175func (e *Engine) lastAction() string {
176 if x := e.st.LastExec(); x != nil {
177 return x.Action
178 }
179 return ""
180}
181
182func (e *Engine) transition(t lex.Transition) {
183 e.st.Seq++
184 t.Pipeline, t.Run, t.Workflow, t.Seq, t.CreatedAt = e.st.Meta.Pipeline, e.st.Meta.Run, e.st.Meta.Workflow, e.st.Seq, now()
185 e.emit(lex.RunEvent{Kind: lex.KindTransition, Transition: &t})
186}
187
188func (e *Engine) emit(ev lex.RunEvent) {
189 if err := e.Log.Emit(ev); err != nil {
190 fmt.Fprintf(os.Stderr, "ci: writing %s: %v\n", e.Dir.Events(), err)
191 }
192}
193
194// finish ends the run.
195func (e *Engine) finish(end End, from string) (*State, error) {
196 e.releaseLocks()
197 e.transition(lex.Transition{From: from, Reason: end.Reason, Detail: end.Detail})
198 e.st.End = &end
199 e.st.Status = end.Status
200 e.st.Gate = nil
201 e.emit(lex.RunEvent{Kind: lex.KindRun, Status: end.Status, Content: end.Detail})
202 return e.st, e.st.Save(e.Dir)
203}
204
205// execute runs one action and returns its (validated) result and the
206// result file's path relative to the run dir. An error means the run can't
207// go on (cancellation, an engine fault); a failing action is a result.
208func (e *Engine) execute(ctx context.Context, p Pending) (lex.Result, string, error) {
209 e.st.Execs++
210 n := e.st.Execs
211 if i, ok := e.M.IsStep(p.Action); ok {
212 e.st.Cursor = i + 1
213 }
214 a := e.M.Actions[p.Action]
215 root := e.st.Meta.Root
216 dir := filepath.Clean(filepath.Join(root, a.Dir))
217 base := fmt.Sprintf("%03d-%s", n, p.Action)
218 resultRel := filepath.Join("results", base+".json")
219 resultFile := filepath.Join(e.Dir.Path(), resultRel)
220 inputsFile := filepath.Join(e.Dir.Results(), base+".inputs.json")
221 inputs := p.Inputs
222 if len(inputs) == 0 {
223 inputs = json.RawMessage(`{}`)
224 }
225 if err := os.WriteFile(inputsFile, append(inputs, '\n'), 0o644); err != nil {
226 return lex.Result{}, "", err
227 }
228 triggerFile := ""
229 if p.TriggerResult != "" {
230 triggerFile = filepath.Join(e.Dir.Path(), p.TriggerResult)
231 }
232 _ = os.Remove(resultFile)
233
234 x := Exec{N: n, Action: p.Action, Then: p.Then, Reason: p.Reason, Result: resultRel, Started: now()}
235 e.st.History = append(e.st.History, x)
236 if err := e.st.Save(e.Dir); err != nil {
237 return lex.Result{}, "", err
238 }
239 if a.Concurrency != nil {
240 if err := e.acquire(ctx, a.Concurrency.Group); err != nil {
241 return lex.Result{}, "", errors.Join(err, e.recordExec(x, StatusCancelled, -1))
242 }
243 }
244 e.emit(lex.RunEvent{Kind: lex.KindStart, Action: p.Action, Attempt: n, Command: a.Command})
245
246 code, runErr := 0, error(nil)
247 if stub, ok := e.Opts.Stubs[p.Action]; ok {
248 x.Stubbed = true
249 code, runErr = e.stub(stub, resultFile, p.Action, n)
250 } else {
251 stdout := e.Log.Lines(p.Action, n, "stdout")
252 stderr := e.Log.Lines(p.Action, n, "stderr")
253 code, runErr = e.Exec.Exec(ctx, Spec{
254 Root: root, Dir: dir, Command: a.Command,
255 Env: e.actionEnv(p, n, resultFile, inputsFile, triggerFile),
256 Stdout: stdout, Stderr: stderr,
257 })
258 stdout.Close()
259 stderr.Close()
260 }
261 if runErr != nil && ctx.Err() != nil {
262 if err := e.recordExec(x, StatusCancelled, -1); err != nil {
263 return lex.Result{}, "", errors.Join(runErr, err)
264 }
265 return lex.Result{}, "", runErr
266 }
267
268 var res lex.Result
269 if runErr != nil {
270 res = lex.Result{Version: lex.ResultVersion, Status: lex.Failure, Summary: "could not run: " + runErr.Error()}
271 code = -1
272 } else {
273 var err error
274 res, err = ReadResult(e.M, resultFile, code, e.Dir.Artifacts())
275 if err != nil {
276 res = ContractFailure(err)
277 }
278 }
279 if err := mergeOutputs(filepath.Join(e.stateDir(), "outputs.json"), res.Outputs); err != nil {
280 res = ContractFailure(fmt.Errorf("outputs: %w", err))
281 }
282 exit := code
283 e.emit(lex.RunEvent{Kind: lex.KindEnd, Action: p.Action, Attempt: n, ExitCode: &exit, Status: res.Status})
284 e.emit(lex.RunEvent{Kind: lex.KindResult, Action: p.Action, Attempt: n, Result: &res})
285 if err := e.recordExec(x, res.Status, code); err != nil {
286 return lex.Result{}, "", err
287 }
288 return res, resultRel, nil
289}
290
291func (e *Engine) recordExec(x Exec, status string, code int) error {
292 x.Status, x.ExitCode, x.Finished = status, code, now()
293 e.st.History[len(e.st.History)-1] = x
294 return e.st.Save(e.Dir)
295}
296
297// stub stands in for an action: the fixture becomes its result, and its
298// exit code follows the fixture's status.
299func (e *Engine) stub(fixture, resultFile, action string, n int) (int, error) {
300 b, err := os.ReadFile(fixture)
301 if err != nil {
302 return -1, fmt.Errorf("stub for %s: %w", action, err)
303 }
304 if err := os.WriteFile(resultFile, b, 0o644); err != nil {
305 return -1, err
306 }
307 e.emit(lex.RunEvent{Kind: lex.KindLog, Action: action, Attempt: n, Stream: "stderr",
308 Content: "ci: stubbed with " + fixture})
309 var probe struct {
310 Status string `json:"status"`
311 }
312 if err := json.Unmarshal(b, &probe); err == nil && probe.Status == lex.Failure {
313 return 1, nil
314 }
315 return 0, nil
316}
317
318// ErrNoRun is returned when a run directory has no state.
319var ErrNoRun = errors.New("no such run")
