alice.pds.demo.boltless.dev
This repository has no description
1package mill
2
3import (
4 "context"
5 "log/slog"
6 "time"
7
8 "tangled.org/core/api/tangled"
9 "tangled.org/core/spindle/engine"
10 "tangled.org/core/spindle/models"
11 "tangled.org/core/spindle/secrets"
12)
13
14// raw pipeline/workflow carried forward, executor runs the real InitWorkflow
15type millWorkflowState struct {
16 RawWorkflow tangled.Pipeline_Workflow
17 RawPipeline tangled.Pipeline
18 Lease *RemoteLease
19}
20
21// stand-in for a real engine, registered under the real names
22// ("microvm", "nixery"), all sharing one Mill
23type Engine struct {
24 name string
25 mill *Mill
26 l *slog.Logger
27}
28
29func (e *Engine) AuthorsRemoteStatus() {}
30
31func NewEngine(name string, mill *Mill) *Engine {
32 return &Engine{name: name, mill: mill, l: mill.l.With("engine", "mill:"+name)}
33}
34
35// synthetic one-step workflow so processPipeline injects TANGLED_* env
36// and marks pending normally. the real InitWorkflow runs exactly once, on
37// the executor inside ReserveSeat, and commit reuses that workflow
38func (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) {
39 return &models.Workflow{
40 Name: twf.Name,
41 Environment: map[string]string{},
42 Steps: []models.Step{remoteStep{}},
43 Data: &millWorkflowState{
44 RawWorkflow: twf,
45 RawPipeline: tpl,
46 },
47 }, nil
48}
49
50// no-op logger for the synthetic workflow. the executor's lines stream
51// into this wid's log directly, a local logger would just write competing
52// lines
53func (e *Engine) WorkflowLogger(wid models.WorkflowId) models.WorkflowLogger {
54 return models.NullLogger{}
55}
56
57// the placement seam, blocks on remote placement which the user sees as
58// "pending". only StartWorkflows calls this, always Wait
59func (e *Engine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, _ engine.AcquireMode) (engine.WorkflowSlot, error) {
60 return e.mill.place(ctx, e.name, wid, wf)
61}
62
63// no-op. real setup happens on the executor
64func (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error {
65 e.l.Info("remote job placed, awaiting commit", "wid", wid)
66 return nil
67}
68
69// hands over the secrets and blocks on the terminal result streamed over the
70// session
71func (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, unlocked []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error {
72 return e.mill.commitAndWait(ctx, w, unlocked)
73}
74
75// deliberately generous, the executor enforces the real timeout. the mill
76// only caps a hung or silent executor, true death is caught by reconnect grace
77func (e *Engine) WorkflowTimeout() time.Duration {
78 return e.mill.cfg.JobTimeout
79}
80
81// cancels a still-running attempt. no-op if already terminal
82func (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error {
83 e.mill.destroy(wid)
84 return nil
85}