Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110package mill
import ( "context" "log/slog" "time"
"tangled.org/core/api/tangled" "tangled.org/core/spindle/engine" "tangled.org/core/spindle/models" "tangled.org/core/spindle/secrets")
// raw pipeline/workflow carried forward, executor runs the real InitWorkflowtype millWorkflowState struct { RawWorkflow tangled.Pipeline_Workflow RawPipeline tangled.Pipeline Lease *RemoteLease}
// stand-in for a real engine, registered under the real names// ("microvm", "nixery"), all sharing one Milltype Engine struct { name string mill *Mill l *slog.Logger}
func (e *Engine) AuthorsRemoteStatus() {}
func (e *Engine) MetricEngineName() string { return e.name}
func NewEngine(name string, mill *Mill) *Engine { return &Engine{name: name, mill: mill, l: mill.l.With("engine", "mill:"+name)}}
// synthetic one-step workflow so processPipeline injects TANGLED_* env// and marks pending normally. the real InitWorkflow runs exactly once, on// the executor inside ReserveSeat, and commit reuses that workflowfunc (e *Engine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) { var ownerDID, repoDID string if md := tpl.TriggerMetadata; md != nil && md.Repo != nil { ownerDID = md.Repo.Did repoDID = ownerDID if md.Repo.RepoDid != nil && *md.Repo.RepoDid != "" { repoDID = *md.Repo.RepoDid } } caches, err := models.ParseCacheEntries(twf.Raw) if err != nil { return nil, err } if e.mill.cfg.CacheStoreID == "" { caches = nil }
return &models.Workflow{ Name: twf.Name, Environment: map[string]string{}, OwnerDID: ownerDID, RepoDID: repoDID, Steps: []models.Step{remoteStep{}}, Caches: caches, Engine: e.name, Data: &millWorkflowState{ RawWorkflow: twf, RawPipeline: tpl, }, }, nil}
// no-op logger for the synthetic workflow. the executor's lines stream// into this wid's log directly, a local logger would just write competing// linesfunc (e *Engine) WorkflowLogger(wid models.WorkflowId) models.WorkflowLogger { return models.NullLogger{}}
// the placement seam, blocks on remote placement which the user sees as// "pending". only StartWorkflows calls this, always Waitfunc (e *Engine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, _ engine.AcquireMode) (engine.WorkflowSlot, error) { return e.mill.place(ctx, e.name, wid, wf)}
// real setup happens on the executorfunc (e *Engine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error { e.l.Info("remote job placed, awaiting commit", "wid", wid) return nil}
// hands over the secrets and blocks on the terminal result streamed over the// sessionfunc (e *Engine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, unlocked []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error { return e.mill.commitAndWait(ctx, w, unlocked)}
// deliberately generous, the executor enforces the real timeout. the mill// only caps a hung or silent executor, true death is caught by reconnect gracefunc (e *Engine) WorkflowTimeout() time.Duration { return e.mill.cfg.JobTimeout}
// cancels a still-running attempt. no-op if already terminalfunc (e *Engine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { e.mill.destroy(wid) return nil}