package 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)
}
}