Something went wrong. Try again.
Monorepo for Tangled forked from tangled.org/core
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196package engine
import ( "context" "log/slog" "sync" "testing" "time"
"tangled.org/core/spindle/config" "tangled.org/core/spindle/models" "tangled.org/core/spindle/observability")
type queueRun struct { statuses map[string]models.StatusKind finished []slog.Record metrics *observability.Metrics}
// runQueued starts workflows a and b behind a single slotfunc runQueued(t *testing.T, timeout time.Duration, runStep func(ctx context.Context, wid models.WorkflowId) error, during func(holder, waiter models.WorkflowId)) queueRun { t.Helper() var records []slog.Record logger := slog.New(&captureHandler{records: &records, mu: &sync.Mutex{}}) metrics := observability.NewMetrics() testDB := newTestDB(t) eng := &slottedEngine{mockEngine: &mockEngine{timeout: timeout}, slots: NewSemaphoreSlotter(1)} first := make(chan models.WorkflowId, 1) var once sync.Once eng.runStepFunc = func(ctx context.Context, wid models.WorkflowId, _ int, _ models.WorkflowLogger) error { once.Do(func() { first <- wid }) return runStep(ctx, wid) }
pipelineId := models.PipelineId("queue-timeout") step := []models.Step{mockStep{name: "step"}} pipeline := &models.Pipeline{Workflows: map[models.Engine][]models.Workflow{ eng: {{Name: "a", Steps: step}, {Name: "b", Steps: step}}, }} cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
var wg sync.WaitGroup wg.Go(func() { StartWorkflows(logger, nil, cfg, nil, nil, testDB, nil, nil, nil, observability.WithMetrics(context.Background(), metrics), pipeline, pipelineId) }) if during != nil { holder := <-first waiter := models.WorkflowId{PipelineId: pipelineId, Name: "a"} if holder.Name == "a" { waiter.Name = "b" } during(holder, waiter) } wg.Wait()
run := queueRun{statuses: map[string]models.StatusKind{}, metrics: metrics} for _, name := range []string{"a", "b"} { st, err := testDB.GetStatus(models.WorkflowId{PipelineId: pipelineId, Name: name}) if err != nil { t.Fatalf("GetStatus(%s) error = %v", name, err) } run.statuses[name] = st } for _, r := range records { if r.Message == "workflow finished" { run.finished = append(run.finished, r) } } return run}
func TestQueueWaitDoesNotCountAgainstTheWorkflowTimeout(t *testing.T) { // each run takes 250ms. the second one finishes 500ms after both were queued, past the 400ms timeout statuses := runQueued(t, 400*time.Millisecond, func(ctx context.Context, _ models.WorkflowId) error { select { case <-time.After(250 * time.Millisecond): return nil case <-ctx.Done(): return ctx.Err() } }, nil).statuses for name, st := range statuses { if st != models.StatusKindSuccess { t.Errorf("workflow %s status = %s, want success", name, st) } }}
func TestWorkflowThatRunsPastItsTimeoutAfterQueueingStillTimesOut(t *testing.T) { timeout := 400 * time.Millisecond var mu sync.Mutex var ran time.Duration holderSeen := false statuses := runQueued(t, timeout, func(ctx context.Context, _ models.WorkflowId) error { mu.Lock() holder := !holderSeen holderSeen = true mu.Unlock() if holder { time.Sleep(250 * time.Millisecond) return nil } start := time.Now() <-ctx.Done() mu.Lock() ran = time.Since(start) mu.Unlock() return ctx.Err() }, nil).statuses
timedOut := 0 for _, st := range statuses { if st == models.StatusKindTimeout { timedOut++ } } if timedOut != 1 { t.Errorf("statuses = %v, want exactly one timeout", statuses) } if ran < timeout-50*time.Millisecond { t.Errorf("queued workflow ran for %s before timing out, want about %s", ran, timeout) }}
func TestCancelInterruptsAQueuedWorkflow(t *testing.T) { release := make(chan struct{}) var mu sync.Mutex var ranSteps []string statuses := runQueued(t, 5*time.Second, func(ctx context.Context, wid models.WorkflowId) error { mu.Lock() ranSteps = append(ranSteps, wid.Name) mu.Unlock() select { case <-release: return nil case <-ctx.Done(): return ctx.Err() } }, func(_, waiter models.WorkflowId) { // nothing reports when b starts waiting for the slot. this sleep gives it time to get there time.Sleep(300 * time.Millisecond) CancelWorkflow(waiter) time.Sleep(100 * time.Millisecond) close(release) }).statuses
if len(ranSteps) != 1 { t.Errorf("steps ran for %v, want only the slot holder", ranSteps) } cancelled := 0 for _, st := range statuses { if st == models.StatusKindCancelled { cancelled++ } } if cancelled != 1 { t.Errorf("statuses = %v, want exactly one cancelled", statuses) }}
func TestQueueWaitIsLeftOutOfTheWorkflowDuration(t *testing.T) { // each run takes 250ms and the second one waits about 250ms for the slot first run := runQueued(t, 5*time.Second, func(context.Context, models.WorkflowId) error { time.Sleep(250 * time.Millisecond) return nil }, nil)
if len(run.finished) != 2 { t.Fatalf("got %d workflow finished logs, want 2", len(run.finished)) } for _, r := range run.finished { r.Attrs(func(a slog.Attr) bool { if a.Key == "duration_seconds" && a.Value.Float64() > 0.4 { t.Errorf("logged duration %vs, want about 0.25s of running", a.Value.Float64()) } return true }) }
families, err := run.metrics.Registry().Gather() if err != nil { t.Fatal(err) } for _, family := range families { if family.GetName() != "spindle_workflow_duration_seconds" { continue } for _, m := range family.GetMetric() { if sum := m.GetHistogram().GetSampleSum(); sum > 0.65 { t.Errorf("recorded %vs for two 250ms runs, want about 0.5s", sum) } } }}