Something went wrong. Try again.
Monorepo for Tangled tangled.org
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216package engine
import ( "context" "encoding/json" "errors" "log/slog" "os" "path/filepath" "reflect" "slices" "strings" "sync" "testing" "time"
"github.com/bluesky-social/indigo/atproto/syntax" "go.opentelemetry.io/otel" sdktrace "go.opentelemetry.io/otel/sdk/trace" "go.opentelemetry.io/otel/sdk/trace/tracetest" "tangled.org/core/spindle/config" "tangled.org/core/spindle/db" "tangled.org/core/spindle/models" "tangled.org/core/spindle/observability" "tangled.org/core/spindle/quota" "tangled.org/core/spindle/secrets" "tangled.org/core/spindle/storage" "tangled.org/core/workflow")
type mockStep struct { name string command string}
func (m mockStep) Name() string { return m.name }func (m mockStep) Command() string { return m.command }func (m mockStep) Kind() models.StepKind { return models.StepKindUser }
type emptyVault struct{}
func (emptyVault) AddSecret(context.Context, secrets.UnlockedSecret) error { return nil}
func (emptyVault) RemoveSecret(context.Context, secrets.Secret[any]) error { return nil}
func (emptyVault) RemoveAllSecrets(context.Context, secrets.RepoIdentifier) error { return nil}
func (emptyVault) GetSecretsLocked(context.Context, secrets.RepoIdentifier) ([]secrets.LockedSecret, error) { return nil, nil}
func (emptyVault) GetSecretsUnlocked(context.Context, secrets.RepoIdentifier) ([]secrets.UnlockedSecret, error) { return nil, nil}
type staticCacheController struct { bindings []models.CacheBinding}
func (c staticCacheController) Plan(context.Context, *models.Pipeline, *models.Workflow) ([]models.CacheBinding, error) { return c.bindings, nil}
func (staticCacheController) Apply(context.Context, CacheUpdate) error { return nil}
type mockEngine struct { mu sync.Mutex setupCalls []models.WorkflowId setupTrusted []bool runStepCalls []models.WorkflowId runStepSecrets [][]secrets.UnlockedSecret setupFunc func(ctx context.Context, wid models.WorkflowId) error runStepFunc func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error destroyFunc func(ctx context.Context, wid models.WorkflowId) error runWorkflowFunc func(*models.Workflow) timeout time.Duration}
func (m *mockEngine) InitWorkflow(twf workflow.WorkflowSpec, tpl workflow.PipelineSpec) (*models.Workflow, error) { owner := "did:web:alice" repo := "did:web:alice/repo" if md := tpl.TriggerMetadata; md != nil && md.Repo != nil { owner = md.Repo.Did if md.Repo.RepoDid != nil { repo = *md.Repo.RepoDid } } return &models.Workflow{ OwnerDID: owner, RepoDID: repo, }, nil}
func (m *mockEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error { m.mu.Lock() m.setupCalls = append(m.setupCalls, wid) m.setupTrusted = append(m.setupTrusted, wf.TrustedSource) fn := m.setupFunc m.mu.Unlock() if fn != nil { return fn(ctx, wid) } return nil}
func (m *mockEngine) WorkflowTimeout() time.Duration { if m.timeout != 0 { return m.timeout } return 5 * time.Second}
func (m *mockEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { if m.destroyFunc != nil { return m.destroyFunc(ctx, wid) } return nil}
func (m *mockEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error { m.mu.Lock() m.runStepCalls = append(m.runStepCalls, wid) m.runStepSecrets = append(m.runStepSecrets, slices.Clone(secrets)) runStepFn := m.runStepFunc runWorkflowFn := m.runWorkflowFunc m.mu.Unlock()
if runWorkflowFn != nil { runWorkflowFn(w) } if runStepFn != nil { return runStepFn(ctx, wid, idx, wfLogger) } return nil}
func newTestDB(t *testing.T) *db.DB { t.Helper() d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db")) if err != nil { t.Fatalf("failed to create test db: %v", err) } return d}
func TestStartWorkflows_CollisionRejection(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
eng := &mockEngine{} pipelineId := models.PipelineId("test-rkey")
// two names that normalize to the same wid must not both run wfColliding1 := models.Workflow{ Name: "test-job", Steps: []models.Step{mockStep{name: "step1"}}, } wfColliding2 := models.Workflow{ Name: "test job", Steps: []models.Step{mockStep{name: "step1"}}, } wfUnique := models.Workflow{ Name: "unique_job", Steps: []models.Step{mockStep{name: "step1"}}, }
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: {wfColliding1, wfColliding2, wfUnique}, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
eng.mu.Lock() setupCalls := append([]models.WorkflowId(nil), eng.setupCalls...) eng.mu.Unlock()
for _, call := range setupCalls { if call.Name == "test-job" || call.Name == "test job" { t.Fatalf("expected colliding workflow %s to not be started", call.Name) } }
hasUnique := false for _, call := range setupCalls { if call.Name == "unique_job" { hasUnique = true } } if !hasUnique { t.Fatalf("expected unique workflow unique_job to be started") }
widColliding1 := models.WorkflowId{PipelineId: pipelineId, Name: "test-job"} widColliding2 := models.WorkflowId{PipelineId: pipelineId, Name: "test job"} widUnique := models.WorkflowId{PipelineId: pipelineId, Name: "unique_job"}
status1, err := testDB.GetStatus(widColliding1) if err != nil || status1 != models.StatusKindFailed { t.Fatalf("expected colliding1 status to be failed, got status=%v err=%v", status1, err) }
status2, err := testDB.GetStatus(widColliding2) if err != nil || status2 != models.StatusKindFailed { t.Fatalf("expected colliding2 status to be failed, got status=%v err=%v", status2, err) }
statusUnique, err := testDB.GetStatus(widUnique) if err != nil || statusUnique != models.StatusKindSuccess { t.Fatalf("expected unique status to be success, got status=%v err=%v", statusUnique, err) }}
func TestCancelWorkflow_NotOverwritten(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
stepStarted := make(chan struct{}) eng := &mockEngine{ runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error { close(stepStarted) <-ctx.Done() return ctx.Err() }, }
pipelineId := models.PipelineId("test-rkey")
wid := models.WorkflowId{ PipelineId: pipelineId, Name: "cancel_test_job", }
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: { { Name: "cancel_test_job", Steps: []models.Step{mockStep{name: "step1"}}, }, }, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} doneChan := make(chan struct{}) go func() { StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId) close(doneChan) }()
select { case <-stepStarted: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for step to start") }
_ = testDB.StatusCancelled(wid, "User canceled the workflow", -1, nil) CancelWorkflow(wid)
select { case <-doneChan: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for StartWorkflows to complete") }
// the runner writes StatusCancelled itself when it sees the canceled ctx // the handler writes nothing for a live wf, so nothing lands after to overwrite it st, err := testDB.GetStatus(wid) if err != nil { t.Fatalf("GetStatus error = %v", err) } if st != models.StatusKindCancelled { t.Fatalf("expected status to be cancelled, got %s", st) }}
func TestStartWorkflowsScopesSecretsToTrustedPipelineRepo(t *testing.T) { vault, err := secrets.NewSQLiteManager(":memory:") if err != nil { t.Fatal(err) } repoA := syntax.DID("did:plc:workflowrepoa") repoB := syntax.DID("did:plc:workflowrepob") for _, secret := range []secrets.UnlockedSecret{ {Repo: secrets.RepoIdentifier(repoA), Key: "A_SECRET", Value: "alpha"}, {Repo: secrets.RepoIdentifier(repoB), Key: "B_SECRET", Value: "bravo"}, } { if err := vault.AddSecret(context.Background(), secret); err != nil { t.Fatal(err) } } run := func(trusted bool) [][]secrets.UnlockedSecret { t.Helper() database := newTestDB(t) defer database.Close() engine := &mockEngine{} pipeline := &models.Pipeline{ RepoDid: repoA, TrustedSource: trusted, Workflows: map[models.Engine][]models.Workflow{ engine: {{Name: "ci.yml", Steps: []models.Step{mockStep{name: "test"}}}}, }, } StartWorkflows(slog.Default(), vault, &config.Config{Server: config.Server{LogDir: t.TempDir()}}, nil, nil, database, nil, nil, nil, context.Background(), pipeline, models.PipelineId("3mu2xwiorc2xl")) engine.mu.Lock() defer engine.mu.Unlock() return slices.Clone(engine.runStepSecrets) }
trustedSecrets := run(true) if len(trustedSecrets) != 1 || len(trustedSecrets[0]) != 1 || trustedSecrets[0][0].Key != "A_SECRET" || trustedSecrets[0][0].Value != "alpha" { t.Fatalf("trusted secrets = %#v", trustedSecrets) } if untrustedSecrets := run(false); len(untrustedSecrets) != 1 || len(untrustedSecrets[0]) != 0 { t.Fatalf("untrusted secrets = %#v", untrustedSecrets) }}
func TestStartWorkflowsMarksOnlyTrustedWorkflowsTrusted(t *testing.T) { for _, tc := range []struct { name string repoDid syntax.DID trusted bool want bool }{ {"trusted push", "did:plc:repo", true, true}, {"untrusted source", "did:plc:repo", false, false}, {"trusted source without a repo did", "", true, false}, } { t.Run(tc.name, func(t *testing.T) { database := newTestDB(t) defer database.Close() vault, err := secrets.NewSQLiteManager(":memory:") if err != nil { t.Fatal(err) } eng := &mockEngine{} pipeline := &models.Pipeline{ RepoDid: tc.repoDid, TrustedSource: tc.trusted, Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: "ci.yml", Steps: []models.Step{mockStep{name: "test"}}}}, }, } StartWorkflows(slog.Default(), vault, &config.Config{Server: config.Server{LogDir: t.TempDir()}}, nil, nil, database, nil, nil, nil, context.Background(), pipeline, models.PipelineId("3mu2xwiorc2xl")) eng.mu.Lock() defer eng.mu.Unlock() if !slices.Equal(eng.setupTrusted, []bool{tc.want}) { t.Fatalf("setup saw TrustedSource %v, want [%t]", eng.setupTrusted, tc.want) } }) }}
func TestStartWorkflows_FlushesLogBeforeTerminalStatus(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) logDir := t.TempDir() vault, err := secrets.NewSQLiteManager(filepath.Join(t.TempDir(), "secrets.db")) if err != nil { t.Fatal(err) }
repoDid := syntax.DID("did:plc:test") if err := vault.AddSecret(context.Background(), secrets.UnlockedSecret{ Key: "LONG_SECRET", Value: strings.Repeat("s", 256), Repo: secrets.RepoIdentifier(repoDid.String()), }); err != nil { t.Fatal(err) }
destroyStarted := make(chan struct{}) releaseDestroy := make(chan struct{}) var releaseOnce sync.Once release := func() { releaseOnce.Do(func() { close(releaseDestroy) }) } t.Cleanup(release)
eng := &mockEngine{ runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error { if _, err := wfLogger.DataWriter(idx, "stdout").Write([]byte("ssh debug hint\n")); err != nil { return err } return errors.New("step failed") }, destroyFunc: func(ctx context.Context, wid models.WorkflowId) error { close(destroyStarted) <-releaseDestroy return nil }, }
pipelineId := models.PipelineId("test-rkey") wid := models.WorkflowId{PipelineId: pipelineId, Name: "failed_job"} pipeline := &models.Pipeline{ RepoDid: repoDid, TrustedSource: true, Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: wid.Name, Steps: []models.Step{mockStep{name: "step1"}}}}, }, } cfg := &config.Config{Server: config.Server{LogDir: logDir}}
done := make(chan struct{}) go func() { StartWorkflows(logger, vault, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId) close(done) }()
select { case <-destroyStarted: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for workflow destruction") }
status, err := testDB.GetStatus(wid) if err != nil { t.Fatal(err) } if status != models.StatusKindFailed { t.Fatalf("expected failed status, got %s", status) }
log, err := os.ReadFile(models.LogFilePath(logDir, wid)) if err != nil { t.Fatal(err) } if !strings.Contains(string(log), "ssh debug hint") { t.Fatalf("terminal status was visible before the log was flushed: %s", log) }
release() select { case <-done: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for StartWorkflows to complete") }}
func TestStartWorkflows_EndLinesCarryStepOutcome(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) logDir := t.TempDir()
eng := &mockEngine{ runStepFunc: func(ctx context.Context, wid models.WorkflowId, idx int, wfLogger models.WorkflowLogger) error { if idx == 1 { hint := mockStep{name: "Debug shell"} wfLogger.ControlWriter(2, hint, models.StepStatusStart).Write([]byte{0}) wfLogger.ControlWriter(2, hint, models.StepStatusEnd).Write([]byte{0}) return errors.New("exited with code 1") } return nil }, }
pipelineId := models.PipelineId("outcome-rkey") wid := models.WorkflowId{PipelineId: pipelineId, Name: "outcome"} pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: wid.Name, Steps: []models.Step{mockStep{name: "build"}, mockStep{name: "test"}}}}, }, } cfg := &config.Config{Server: config.Server{LogDir: logDir}} StartWorkflows(logger, emptyVault{}, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
raw, err := os.ReadFile(models.LogFilePath(logDir, wid)) if err != nil { t.Fatal(err) } got := map[string]models.StepOutcome{} for _, encoded := range strings.Split(strings.TrimSpace(string(raw)), "\n") { var line models.LogLine if err := json.Unmarshal([]byte(encoded), &line); err != nil { t.Fatalf("decode log line %q: %v", encoded, err) } if line.Kind == models.LogKindControl && line.StepStatus == models.StepStatusEnd { got[line.Content] = line.StepOutcome } } want := map[string]models.StepOutcome{ "build": models.StepOutcomeSuccess, "test": models.StepOutcomeFailed, "Debug shell": "", } if !reflect.DeepEqual(got, want) { t.Fatalf("end line outcomes = %v, want %v", got, want) }}
func TestStartWorkflowsCapsTheWorkflowTimeBudget(t *testing.T) { deadline := make(chan time.Time, 1) eng := &mockEngine{ timeout: 24 * time.Hour, setupFunc: func(ctx context.Context, _ models.WorkflowId) error { got, ok := ctx.Deadline() if !ok { return errors.New("workflow context has no deadline") } deadline <- got return nil }, } pipeline := &models.Pipeline{Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: "bounded", Steps: []models.Step{mockStep{name: "step"}}}}, }}
StartWorkflows(slog.Default(), nil, &config.Config{Server: config.Server{LogDir: t.TempDir()}}, nil, nil, newTestDB(t), nil, nil, nil, context.Background(), pipeline, "bounded")
remaining := time.Until(<-deadline) if remaining > models.MaxWorkflowTimeBudget || remaining < models.MaxWorkflowTimeBudget-time.Minute { t.Fatalf("workflow time budget = %s, want %s", remaining, models.MaxWorkflowTimeBudget) }}
func TestSetupTimeout_ReportsTimeout(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
// setup blocks past the workflow timeout, so it should land as timeout not failed eng := &mockEngine{ timeout: 100 * time.Millisecond, setupFunc: func(ctx context.Context, wid models.WorkflowId) error { <-ctx.Done() return ctx.Err() }, }
pipelineId := models.PipelineId("test-rkey") wid := models.WorkflowId{PipelineId: pipelineId, Name: "timeout_job"}
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: "timeout_job", Steps: []models.Step{mockStep{name: "step1"}}}}, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
st, err := testDB.GetStatus(wid) if err != nil { t.Fatalf("GetStatus error = %v", err) } if st != models.StatusKindTimeout { t.Fatalf("expected status to be timeout, got %s", st) }
if len(eng.runStepCalls) != 0 { t.Fatalf("expected no steps to run after setup timeout, got %d", len(eng.runStepCalls)) }}
type cacheSave struct { keys []string ctxErr error deadline time.Time}
type cacheMockEngine struct { *mockEngine blockSave bool mu sync.Mutex saves []cacheSave events []string}
func (e *cacheMockEngine) record(event string) { e.mu.Lock() defer e.mu.Unlock() e.events = append(e.events, event)}
func (e *cacheMockEngine) RestoreCache(context.Context, models.WorkflowId, *models.Workflow, storage.Storage, []models.CacheBinding, models.WorkflowLogger) error { return nil}
func (e *cacheMockEngine) SaveCache(ctx context.Context, _ models.WorkflowId, _ *models.Workflow, _ storage.Storage, caches []models.CacheBinding, _ models.WorkflowLogger) error { save := cacheSave{ctxErr: ctx.Err()} save.deadline, _ = ctx.Deadline() for _, c := range caches { save.keys = append(save.keys, c.Key) } if e.blockSave { <-ctx.Done() } e.mu.Lock() e.saves = append(e.saves, save) e.mu.Unlock() e.record("save") return ctx.Err()}
type cachedRun struct { status models.StatusKind saves []cacheSave events []string log string}
func runCachedWorkflow(t *testing.T, pipelineID models.PipelineId, blockSave bool, runStep func(ctx context.Context, wid models.WorkflowId) error) cachedRun { t.Helper() eng := &cacheMockEngine{blockSave: blockSave} eng.mockEngine = &mockEngine{ timeout: 100 * time.Millisecond, runStepFunc: func(ctx context.Context, wid models.WorkflowId, _ int, _ models.WorkflowLogger) error { return runStep(ctx, wid) }, destroyFunc: func(context.Context, models.WorkflowId) error { eng.record("destroy") return nil }, } wid := models.WorkflowId{PipelineId: pipelineID, Name: "cached"} pipeline := &models.Pipeline{ RepoDid: "did:plc:cached", TrustedSource: true, Workflows: map[models.Engine][]models.Workflow{ eng: {{ Name: wid.Name, Steps: []models.Step{mockStep{name: "build"}}, Caches: []models.CacheEntry{ {Key: "always", Paths: []string{"a"}, When: "always"}, {Key: "on-success", Paths: []string{"b"}}, }, }}, }, } controller := staticCacheController{bindings: []models.CacheBinding{ {EntryIndex: 0, Key: "always", SaveKey: "objects/always", When: "always"}, {EntryIndex: 1, Key: "on-success", SaveKey: "objects/on-success"}, }} testDB := newTestDB(t) cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} StartWorkflows(discardLogger, emptyVault{}, cfg, nil, nil, testDB, nil, &fakeStorage{objects: map[string][]byte{}}, controller, context.Background(), pipeline, pipelineID)
status, err := testDB.GetStatus(wid) if err != nil { t.Fatal(err) } log, err := os.ReadFile(models.LogFilePath(cfg.Server.LogDir, wid)) if err != nil { t.Fatal(err) } eng.mu.Lock() defer eng.mu.Unlock() return cachedRun{status: status, saves: eng.saves, events: eng.events, log: string(log)}}
var timeoutSteps = map[string]func(context.Context, models.WorkflowId) error{ "deadline passed": func(ctx context.Context, _ models.WorkflowId) error { <-ctx.Done() return ErrTimedOut }, "engine reported": func(context.Context, models.WorkflowId) error { return ErrTimedOut },}
func timeoutPipelineID(prefix, name string) models.PipelineId { return models.PipelineId(prefix + "-" + strings.ReplaceAll(name, " ", "-"))}
func TestStartWorkflowsSavesAlwaysCachesAfterTimeout(t *testing.T) { t.Parallel() for name, runStep := range timeoutSteps { t.Run(name, func(t *testing.T) { t.Parallel() start := time.Now() run := runCachedWorkflow(t, timeoutPipelineID("timeout", name), false, runStep) if run.status != models.StatusKindTimeout { t.Fatalf("status = %s, want timeout", run.status) } if len(run.saves) != 1 || !slices.Equal(run.saves[0].keys, []string{"always"}) { t.Fatalf("saves = %+v, want only the always cache", run.saves) } if run.saves[0].ctxErr != nil { t.Fatalf("save context already done: %v", run.saves[0].ctxErr) } if run.saves[0].deadline.IsZero() || run.saves[0].deadline.After(time.Now().Add(CacheSaveTimeout)) || run.saves[0].deadline.Before(start.Add(CacheSaveTimeout)) { t.Fatalf("save deadline = %v, want %s after the timeout", run.saves[0].deadline, CacheSaveTimeout) } if !slices.Equal(run.events, []string{"save", "destroy"}) { t.Fatalf("events = %v, want save then a single destroy", run.events) } if !strings.Contains(run.log, CacheSaveStep.Name()) { t.Fatalf("log has no %q step: %s", CacheSaveStep.Name(), run.log) } }) }}
func TestStartWorkflowsDestroysAfterABlockedSave(t *testing.T) { saveTimeout := CacheSaveTimeout CacheSaveTimeout = 200 * time.Millisecond t.Cleanup(func() { CacheSaveTimeout = saveTimeout })
for name, runStep := range timeoutSteps { t.Run(name, func(t *testing.T) { start := time.Now() run := runCachedWorkflow(t, timeoutPipelineID("blocked", name), true, runStep) if run.status != models.StatusKindTimeout { t.Fatalf("status = %s, want timeout", run.status) } if !slices.Equal(run.events, []string{"save", "destroy"}) { t.Fatalf("events = %v, want save then a single destroy", run.events) } if took := time.Since(start); took > 5*time.Second { t.Fatalf("workflow took %v with a blocked save", took) } }) }}
func TestStartWorkflowsSavesNothingAfterCancel(t *testing.T) { t.Parallel() run := runCachedWorkflow(t, "cancel-cached", false, func(ctx context.Context, wid models.WorkflowId) error { CancelWorkflow(wid) <-ctx.Done() return ErrWorkflowCanceled }) if run.status != models.StatusKindCancelled { t.Fatalf("status = %s, want cancelled", run.status) } if !slices.Equal(run.events, []string{"destroy"}) { t.Fatalf("events = %v, want only a destroy", run.events) }}
func TestStartWorkflowsSavesAlwaysCachesAfterFailure(t *testing.T) { t.Parallel() run := runCachedWorkflow(t, "failed-cached", false, func(context.Context, models.WorkflowId) error { return errors.New("exited with code 1") }) if run.status != models.StatusKindFailed { t.Fatalf("status = %s, want failed", run.status) } if len(run.saves) != 1 || !slices.Equal(run.saves[0].keys, []string{"always"}) { t.Fatalf("saves = %+v, want only the always cache", run.saves) } if !slices.Equal(run.events, []string{"save", "destroy"}) { t.Fatalf("events = %v, want save then a single destroy", run.events) }}
func TestStartWorkflows_SpanEnrichment(t *testing.T) { oldProvider := otel.GetTracerProvider() recorder := tracetest.NewSpanRecorder() provider := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(recorder)) otel.SetTracerProvider(provider) t.Cleanup(func() { _ = provider.Shutdown(context.Background()) otel.SetTracerProvider(oldProvider) })
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
eng := &mockEngine{} pipelineId := models.PipelineId("test-rkey")
pipeline := &models.Pipeline{ RepoDid: "did:web:test-repo", Workflows: map[models.Engine][]models.Workflow{ eng: { { Name: "test-workflow", Steps: []models.Step{ mockStep{name: "step-1"}, }, OwnerDID: "did:web:test-owner", RepoDID: "did:web:test-repo", }, }, }, TrustedSource: false, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId)
spans := recorder.Ended() var workflowRunSpan, stepRunSpan sdktrace.ReadOnlySpan for _, span := range spans { if span.Name() == "workflow.run" { workflowRunSpan = span } else if span.Name() == "step.run" { stepRunSpan = span } }
if workflowRunSpan == nil { t.Fatal("workflow.run span not recorded") } if stepRunSpan == nil { t.Fatal("step.run span not recorded") }
workflowAttrs := make(map[string]string) for _, attr := range workflowRunSpan.Attributes() { workflowAttrs[string(attr.Key)] = attr.Value.AsString() }
wid := models.WorkflowId{ PipelineId: pipelineId, Name: "test-workflow", }
expectedWfAttrs := map[string]string{ observability.OwnerDIDKey: "did:web:test-owner", observability.RepoDIDKey: "did:web:test-repo", observability.PipelineIDKey: pipelineId.String(), observability.WorkflowIDKey: wid.String(), } for k, wantVal := range expectedWfAttrs { if gotVal := workflowAttrs[k]; gotVal != wantVal { t.Errorf("workflow attribute %q = %q, want %q", k, gotVal, wantVal) } }
stepAttrs := make(map[string]string) for _, attr := range stepRunSpan.Attributes() { stepAttrs[string(attr.Key)] = attr.Value.AsString() }
expectedStepAttrs := map[string]string{ observability.OwnerDIDKey: "did:web:test-owner", observability.RepoDIDKey: "did:web:test-repo", observability.PipelineIDKey: pipelineId.String(), observability.WorkflowIDKey: wid.String(), observability.StepNameKey: "step-1", } for k, wantVal := range expectedStepAttrs { if gotVal := stepAttrs[k]; gotVal != wantVal { t.Errorf("step attribute %q = %q, want %q", k, gotVal, wantVal) } } hasStepIndex := false for _, attr := range stepRunSpan.Attributes() { if string(attr.Key) == observability.StepIndexKey && attr.Value.AsInterface() == int64(0) { hasStepIndex = true } } if !hasStepIndex { t.Error("step.run span missing step_index attribute or it is not 0") }}
type mockResourceUsageReporter struct { *mockEngine workflows, memory, vcpus, disk int64 usage WorkflowResourceUsage usageOk bool}
func (m *mockResourceUsageReporter) QuotaResources(wf *models.Workflow) quota.Resources { return quota.Resources{ quota.ResourceWorkflows: m.workflows, quota.ResourceMemoryMiB: m.memory, quota.ResourceVCPUs: m.vcpus, quota.ResourceDiskMiB: m.disk, }}
func (m *mockResourceUsageReporter) WorkflowResourceUsage(wf *models.Workflow) (WorkflowResourceUsage, bool) { return m.usage, m.usageOk}
func TestStartWorkflows_ResourceTracking(t *testing.T) { oldProvider := otel.GetTracerProvider() recorder := tracetest.NewSpanRecorder() provider := sdktrace.NewTracerProvider( sdktrace.WithSpanProcessor(recorder), sdktrace.WithSampler(sdktrace.AlwaysSample()), ) otel.SetTracerProvider(provider) t.Cleanup(func() { _ = provider.Shutdown(context.Background()) otel.SetTracerProvider(oldProvider) })
testDB := newTestDB(t)
mu := &sync.Mutex{} records := []slog.Record{} handler := &captureHandler{ mu: mu, records: &records, } logger := slog.New(handler)
eng := &mockResourceUsageReporter{ mockEngine: &mockEngine{}, workflows: 2, memory: 1024, vcpus: 4, disk: 512, usage: WorkflowResourceUsage{ CPUUsec: 12345, MemoryCurrentBytes: 2048, MemoryPeakBytes: 4096, SwapCurrentBytes: 100, SwapPeakBytes: 200, PIDsCurrent: 10, IOReadBytes: 300, IOWriteBytes: 400, IOReadOps: 30, IOWriteOps: 40, VolumeAllocatedBytes: 1000, CgroupAvailable: true, VolumeAvailable: true, }, usageOk: true, }
pipelineId := models.PipelineId("test-rkey")
pipeline := &models.Pipeline{ RepoDid: "did:web:test-repo", Workflows: map[models.Engine][]models.Workflow{ eng: { { Name: "test-workflow", Steps: []models.Step{ mockStep{name: "step-1"}, }, OwnerDID: "did:web:test-owner", RepoDID: "did:web:test-repo", }, }, }, TrustedSource: false, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
metrics := observability.NewMetrics() ctx := observability.WithMetrics(context.Background(), metrics)
StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, ctx, pipeline, pipelineId)
handler.mu.Lock() records = *handler.records handler.mu.Unlock()
var startedLog, finishedLog slog.Record for _, r := range records { if r.Message == "workflow started" { startedLog = r } else if r.Message == "workflow finished" { finishedLog = r } }
if startedLog.Message == "" { t.Fatal("workflow started log not found") } if finishedLog.Message == "" { t.Fatal("workflow finished log not found") }
getAttr := func(r slog.Record, key string) any { var val any r.Attrs(func(a slog.Attr) bool { if a.Key == key { val = a.Value.Any() return false } return true }) return val }
if getAttr(startedLog, "requested_workflows") != int64(2) { t.Errorf("started log requested_workflows = %v, want 2", getAttr(startedLog, "requested_workflows")) } if getAttr(startedLog, "requested_vcpus") != int64(4) { t.Errorf("started log requested_vcpus = %v, want 4", getAttr(startedLog, "requested_vcpus")) }
if getAttr(finishedLog, "result") != "success" { t.Errorf("finished log result = %v, want success", getAttr(finishedLog, "result")) } if getAttr(finishedLog, "actual_cpu_usec") != uint64(12345) { t.Errorf("finished log actual_cpu_usec = %v, want 12345", getAttr(finishedLog, "actual_cpu_usec")) } if getAttr(finishedLog, "actual_memory_peak_bytes") != uint64(4096) { t.Errorf("finished log actual_memory_peak_bytes = %v, want 4096", getAttr(finishedLog, "actual_memory_peak_bytes")) } if getAttr(finishedLog, "error") != "" || getAttr(finishedLog, "failure_class") != string(FailureClassNone) { t.Errorf("finished log error = %q, failure_class = %q, want no error", getAttr(finishedLog, "error"), getAttr(finishedLog, "failure_class")) } duration, _ := getAttr(finishedLog, "duration_seconds").(float64) waited, _ := getAttr(finishedLog, "wait_seconds").(float64) if waited < 0 || duration < 0 { t.Errorf("finished log wait %v, duration %v, want both >= 0", waited, duration) }
spans := recorder.Ended() var workflowRunSpan sdktrace.ReadOnlySpan for _, span := range spans { if span.Name() == "workflow.run" { workflowRunSpan = span break } }
if workflowRunSpan == nil { t.Fatal("workflow.run span not found") }
attrs := make(map[string]any) for _, attr := range workflowRunSpan.Attributes() { attrs[string(attr.Key)] = attr.Value.AsInterface() }
if attrs[observability.ReqWorkflowsKey] != int64(2) { t.Errorf("span attr %s = %v, want 2", observability.ReqWorkflowsKey, attrs[observability.ReqWorkflowsKey]) } if attrs[observability.ReqVCPUsKey] != int64(4) { t.Errorf("span attr %s = %v, want 4", observability.ReqVCPUsKey, attrs[observability.ReqVCPUsKey]) }
if attrs[observability.ActCPUUsecKey] != int64(12345) { t.Errorf("span attr %s = %v, want 12345", observability.ActCPUUsecKey, attrs[observability.ActCPUUsecKey]) } if attrs[observability.ActMemoryPeakBytesKey] != int64(4096) { t.Errorf("span attr %s = %v, want 4096", observability.ActMemoryPeakBytesKey, attrs[observability.ActMemoryPeakBytesKey]) }}
type captureHandler struct { attrs []slog.Attr records *[]slog.Record mu *sync.Mutex}
func (h *captureHandler) Enabled(ctx context.Context, level slog.Level) bool { return true}
func (h *captureHandler) Handle(ctx context.Context, r slog.Record) error { h.mu.Lock() defer h.mu.Unlock() r.AddAttrs(h.attrs...) *h.records = append(*h.records, r) return nil}
func (h *captureHandler) WithAttrs(attrs []slog.Attr) slog.Handler { newAttrs := append([]slog.Attr(nil), h.attrs...) newAttrs = append(newAttrs, attrs...) return &captureHandler{ attrs: newAttrs, records: h.records, mu: h.mu, }}
func (h *captureHandler) WithGroup(name string) slog.Handler { return h}
func TestStartWorkflowsPassesPlannedCacheBindingsToEngine(t *testing.T) { t.Parallel()
database := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) expected := []models.CacheBinding{{ EntryIndex: 0, Paths: []string{"deps"}, Key: "deps", SaveID: "cache-id", SaveKey: "objects/cache-id", }} var got []models.CacheBinding eng := &mockEngine{ runWorkflowFunc: func(workflow *models.Workflow) { got = append([]models.CacheBinding(nil), workflow.CacheBindings...) }, } pipelineID := models.PipelineId("test-rkey") pipeline := &models.Pipeline{ RepoDid: "did:web:example.com", TrustedSource: true, Workflows: map[models.Engine][]models.Workflow{ eng: {{ Name: "cache-job", Engine: "microvm", Steps: []models.Step{mockStep{name: "step1"}}, Caches: []models.CacheEntry{{Key: "deps", Paths: []string{"deps"}}}, }}, }, } cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} store := &fakeStorage{objects: make(map[string][]byte)}
StartWorkflows( logger, emptyVault{}, cfg, nil, nil, database, nil, store, staticCacheController{bindings: expected}, context.Background(), pipeline, pipelineID, )
if !reflect.DeepEqual(got, expected) { t.Fatalf("cache bindings passed to engine = %#v, want %#v", got, expected) }}
func TestStartWorkflowsDestroysWorkflowWhenSetupIsCancelled(t *testing.T) { t.Parallel()
testDB := newTestDB(t) logger := slog.New(slog.NewTextHandler(os.Stderr, nil))
setupStarted := make(chan struct{}) destroyed := make(chan struct{}, 1) eng := &mockEngine{ setupFunc: func(ctx context.Context, wid models.WorkflowId) error { close(setupStarted) <-ctx.Done() return ctx.Err() }, destroyFunc: func(_ context.Context, wid models.WorkflowId) error { destroyed <- struct{}{} return nil }, }
pipelineId := models.PipelineId("test-rkey")
wid := models.WorkflowId{ PipelineId: pipelineId, Name: "cancel_setup_job", }
pipeline := &models.Pipeline{ Workflows: map[models.Engine][]models.Workflow{ eng: { { Name: "cancel_setup_job", Steps: []models.Step{mockStep{name: "step1"}}, }, }, }, }
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}} doneChan := make(chan struct{}) go func() { StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, context.Background(), pipeline, pipelineId) close(doneChan) }()
select { case <-setupStarted: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for workflow setup to start") }
CancelWorkflow(wid)
select { case <-doneChan: case <-time.After(5 * time.Second): t.Fatal("timed out waiting for StartWorkflows to complete") }
select { case <-destroyed: case <-time.After(5 * time.Second): t.Fatal("cancelled setup did not destroy the workflow") }
st, err := testDB.GetStatus(wid) if err != nil { t.Fatalf("GetStatus error = %v", err) } if st != models.StatusKindCancelled { t.Fatalf("expected status to be cancelled, got %s", st) }}